Switch src/documents/bulk*.py from os.path to pathlib.Path (#7862)

Also:
* Ensure that the ruff PTH check remains enabled for these files and
all files added in the future.
* Add some type annotations.
This commit is contained in:
Sebastian Steinbeißer
2024-11-12 18:04:07 +01:00
committed by GitHub
parent d1f255a22e
commit 74d0c9fda5
3 changed files with 132 additions and 46 deletions

View File

@@ -1,14 +1,21 @@
import os
from pathlib import Path
from typing import TYPE_CHECKING
from typing import NoReturn
from zipfile import ZipFile
from documents.models import Document
if TYPE_CHECKING:
from collections.abc import Callable
class BulkArchiveStrategy:
def __init__(self, zipf: ZipFile, follow_formatting: bool = False):
self.zipf = zipf
def __init__(self, zipf: ZipFile, follow_formatting: bool = False) -> None:
self.zipf: ZipFile = zipf
if follow_formatting:
self.make_unique_filename = self._formatted_filepath
self.make_unique_filename: Callable[..., Path | str] = (
self._formatted_filepath
)
else:
self.make_unique_filename = self._filename_only
@@ -17,7 +24,7 @@ class BulkArchiveStrategy:
doc: Document,
archive: bool = False,
folder: str = "",
):
) -> str:
"""
Constructs a unique name for the given document to be used inside the
zip file.
@@ -26,7 +33,7 @@ class BulkArchiveStrategy:
"""
counter = 0
while True:
filename = folder + doc.get_public_filename(archive, counter)
filename: str = folder + doc.get_public_filename(archive, counter)
if filename in self.zipf.namelist():
counter += 1
else:
@@ -37,7 +44,7 @@ class BulkArchiveStrategy:
doc: Document,
archive: bool = False,
folder: str = "",
):
) -> Path:
"""
Constructs a full file path for the given document to be used inside
the zipfile.
@@ -45,24 +52,30 @@ class BulkArchiveStrategy:
The path is already unique, as handled when a document is consumed or updated
"""
if archive and doc.has_archive_version:
in_archive_path = os.path.join(folder, doc.archive_filename)
if TYPE_CHECKING:
assert doc.archive_filename is not None
in_archive_path: Path = Path(folder) / doc.archive_filename
else:
in_archive_path = os.path.join(folder, doc.filename)
if TYPE_CHECKING:
assert doc.filename is not None
in_archive_path = Path(folder) / doc.filename
return in_archive_path
def add_document(self, doc: Document):
def add_document(self, doc: Document) -> NoReturn:
raise NotImplementedError # pragma: no cover
class OriginalsOnlyStrategy(BulkArchiveStrategy):
def add_document(self, doc: Document):
def add_document(self, doc: Document) -> None:
self.zipf.write(doc.source_path, self.make_unique_filename(doc))
class ArchiveOnlyStrategy(BulkArchiveStrategy):
def add_document(self, doc: Document):
def add_document(self, doc: Document) -> None:
if doc.has_archive_version:
if TYPE_CHECKING:
assert doc.archive_path is not None
self.zipf.write(
doc.archive_path,
self.make_unique_filename(doc, archive=True),
@@ -72,8 +85,10 @@ class ArchiveOnlyStrategy(BulkArchiveStrategy):
class OriginalAndArchiveStrategy(BulkArchiveStrategy):
def add_document(self, doc: Document):
def add_document(self, doc: Document) -> None:
if doc.has_archive_version:
if TYPE_CHECKING:
assert doc.archive_path is not None
self.zipf.write(
doc.archive_path,
self.make_unique_filename(doc, archive=True, folder="archive/"),