Merge pull request #444 from OP-TED/feature/TED-1219

refactor notice packager

Author: CaptainOfHacks

ted_sws/notice_packager/adapters/archiver.py

@@ -12,7 +12,7 @@
 import abc
 import os
 from pathlib import Path
-from typing import List, Union
+from typing import List
 from zipfile import ZipFile, ZIP_DEFLATED
 
 ARCHIVE_ZIP_FORMAT = "zip"
@@ -22,35 +22,21 @@
 ARCHIVE_MODE_APPEND = 'a'
 ARCHIVE_MODE = ARCHIVE_MODE_WRITE
 
-PATH_TYPE = Union[Path, str]
-LIST_TYPE = List[PATH_TYPE]
-
 
 class ArchiverABC(abc.ABC):
     """
     This abstract class provides methods definitions and infos for available archivers
     """
 
     @abc.abstractmethod
-    def process_archive(self, archive_name: PATH_TYPE, files: LIST_TYPE, mode: str):
+    def process_archive(self, archive_name: Path, files: List[Path], mode: str) -> Path:
         """
         This method adds the files (based on provided archive mode) to archive
         """
 
 
-class ArchiverFactory:
-    @classmethod
-    def get_archiver(cls, archive_format=ARCHIVE_DEFAULT_FORMAT):
-        """Factory Method to return the needed Archiver, based on archive format"""
-        archivers = {
-            "zip": ZipArchiver
-        }
-
-        return archivers[archive_format]()
-
-
 class ZipArchiver(ArchiverABC):
-    def process_archive(self, archive_name: PATH_TYPE, files: LIST_TYPE, mode: str = ARCHIVE_MODE) -> str:
+    def process_archive(self, archive_name: Path, files: List[Path], mode: str = ARCHIVE_MODE) -> Path:
         with ZipFile(archive_name, mode=mode, compression=ARCHIVE_ZIP_COMPRESSION) as archive:
             for file in files:
                 if os.path.isfile(file):

ted_sws/notice_packager/adapters/template_generator.py

@@ -18,22 +18,22 @@
 
 class TemplateGenerator:
     @classmethod
-    def __generate_template(cls, template, data: PackagerMetadata = None):
+    def __generate_template(cls, template, data: PackagerMetadata = None) -> str:
         template_render = TEMPLATES.get_template(template).render(data.dict())
         return template_render
 
     @classmethod
-    def mets_xml_dmd_rdf_generator(cls, data: PackagerMetadata = None):
+    def mets_xml_dmd_rdf_generator(cls, data: PackagerMetadata = None) -> str:
         template = 'mets_xml_dmd_rdf.jinja2'
         return cls.__generate_template(template, data)
 
     @classmethod
-    def tmd_rdf_generator(cls, data: PackagerMetadata = None):
+    def tmd_rdf_generator(cls, data: PackagerMetadata = None) -> str:
         template = 'tmd_rdf.jinja2'
         return cls.__generate_template(template, data)
 
     @classmethod
-    def mets2action_mets_xml_generator(cls, data: PackagerMetadata = None):
+    def mets2action_mets_xml_generator(cls, data: PackagerMetadata = None) -> str:
         action = data.notice.action.type
         validate_notice_action_type(action)
 

ted_sws/notice_packager/services/notice_packager.py

@@ -11,19 +11,17 @@
 
 import base64
 import binascii
-import os.path
+import pathlib
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from typing import Union
+from typing import List
 
 from ted_sws.core.model.manifestation import METSManifestation
 from ted_sws.core.model.notice import Notice
-from ted_sws.data_manager.adapters.notice_repository import NoticeRepositoryABC
 from ted_sws.notice_metadata_processor.model.metadata import ExtractedMetadata
 from ted_sws.notice_metadata_processor.services.xml_manifestation_metadata_extractor import \
     XMLManifestationMetadataExtractor
-from ted_sws.notice_packager.adapters.archiver import ArchiverFactory, ARCHIVE_ZIP_FORMAT, PATH_TYPE, \
-    LIST_TYPE as PATH_LIST_TYPE
+from ted_sws.notice_packager.adapters.archiver import ZipArchiver
 from ted_sws.notice_packager.adapters.template_generator import TemplateGenerator
 from ted_sws.notice_packager.model.metadata import ACTION_CREATE
 from ted_sws.notice_packager.services.metadata_transformer import MetadataTransformer
@@ -35,182 +33,91 @@
 FILE_TMD_FORMAT = "techMDID001.tmd.rdf"
 FILE_RDF_FORMAT = "{notice_id}.ttl"
 
-NOTICE_METADATA_TYPE = ExtractedMetadata
-IN_DATA_TYPE = Union[NOTICE_METADATA_TYPE, Notice, str]
-RDF_CONTENT_TYPE = Union[str, bytes]
-
-
-def create_notice_package(in_data: IN_DATA_TYPE, rdf_content: RDF_CONTENT_TYPE = None,
-                          extra_files: PATH_LIST_TYPE = None, action: str = ACTION_CREATE,
-                          save_to: PATH_TYPE = None, notice_repository: NoticeRepositoryABC = None) -> str:
-    """
-    :param in_data: can be Notice object, ExtractedMetadata object or notice_id string
-    :param rdf_content: base64 encoded bytes content of rdf file
-    :param extra_files: additional files paths to be added to archive
-    :param action:
-    :param save_to: can be:
-                    None - base64 encoded string of archive content,
-                    "" - temporary archive path,
-                    string (path to archive: containing archive name or just the folder) - archive path
-    :param notice_repository:
-    :return: base64 encoded archive or path to archive
-    """
-
-    tmp_dir = TemporaryDirectory()
-    tmp_dir_path = Path(tmp_dir.name)
-
-    notice_packager = NoticePackager(in_data, action, tmp_dir_path, notice_repository)
-    notice_packager.add_template_files()
-    notice_packager.add_rdf_content(rdf_content)
-    notice_packager.add_extra_files(extra_files)
-
-    return notice_packager.pack(save_to)
-
 
 def package_notice(notice: Notice) -> Notice:
     """
         This function generate METSPackage and set Notice METSManifestation.
     """
-    mets_manifestation_content = create_notice_package(in_data=notice,
-                                                       rdf_content=notice.distilled_rdf_manifestation.object_data.encode(
-                                                           "utf-8"))
+
+    notice_packager = NoticePackager(notice, ACTION_CREATE)
+    notice_packager.add_template_files()
+    notice_packager.add_rdf_content(notice.distilled_rdf_manifestation.object_data.encode("utf-8"))
+    mets_manifestation_content = notice_packager.pack()
     notice.set_mets_manifestation(mets_manifestation=METSManifestation(object_data=mets_manifestation_content))
     return notice
 
 
-def package_notice_and_save_to(notice: Notice, save_to: PATH_TYPE = None) -> str:
-    """
-        This function package a Notice to save_to location.
-    """
-    return create_notice_package(in_data=notice,
-                                 rdf_content=notice.distilled_rdf_manifestation.object_data.encode(
-                                     "utf-8"),
-                                 save_to=save_to)
-
-
 class NoticePackager:
     """
     This class will manage the steps/methods of notice packager creation
     """
 
-    def __init__(self, in_data: IN_DATA_TYPE, action: str, tmp_dir_path: Path, notice_repository: NoticeRepositoryABC):
-        self.notice_metadata: NOTICE_METADATA_TYPE = self.__validated_in_data(in_data, notice_repository)
-        self.archiver = ArchiverFactory.get_archiver(ARCHIVE_ZIP_FORMAT)
-        metadata_transformer = MetadataTransformer(self.notice_metadata)
+    def __init__(self, notice: Notice, action: str):
+        self.tmp_dir = TemporaryDirectory()
+        self.tmp_dir_path = Path(self.tmp_dir.name)
+        notice_metadata: ExtractedMetadata = XMLManifestationMetadataExtractor(
+            xml_manifestation=notice.xml_manifestation).to_metadata()
+        metadata_transformer = MetadataTransformer(notice_metadata)
         self.template_metadata = metadata_transformer.template_metadata(action=action)
-
         self.notice_id = self.template_metadata.notice.id
         self.notice_action = self.template_metadata.notice.action.type
-
-        self.tmp_dir_path = tmp_dir_path
-        self.files: PATH_LIST_TYPE = []
-
-    def __write_template_to_file(self, file_path, template_generator, template_metadata):
-        self.__write_to_file(file_path, template_generator(template_metadata))
-
-    @classmethod
-    def __write_to_file(cls, file_path, data, mode: str = 'x'):
-        with open(file_path, mode) as file:
-            file.write(data)
-            file.close()
-
-    @classmethod
-    def __validated_in_data(cls, in_data: IN_DATA_TYPE, notice_repository: NoticeRepositoryABC) -> NOTICE_METADATA_TYPE:
-        accepted_types = IN_DATA_TYPE.__args__
-        if not isinstance(in_data, accepted_types):
-            raise TypeError('Notice Packager accepts input data of "%s" types only' % accepted_types)
-
-        # here, needed notice_metadata is extracted from provided in_data
-        notice_metadata = None
-        if isinstance(in_data, str):  # notice_id
-            '''
-            if we get notice_id as in_data, 
-            a Notice must be assigned to in_data for next step of the validation flow
-            '''
-            # get Notice from DB
-            notice_id = in_data
-            if isinstance(notice_repository, NoticeRepositoryABC):
-                in_data = notice_repository.get(reference=notice_id)
-            else:
-                raise TypeError('Notice Repository must be sent, if providing notice_id "%s"' % notice_id)
-
-        if isinstance(in_data, Notice):  # Notice
-            '''
-            if we get Notice object as in_data, 
-            notice_metadata should be extracted from it
-            '''
-            notice = in_data
-            notice_metadata = XMLManifestationMetadataExtractor(
-                xml_manifestation=notice.xml_manifestation).to_metadata()
-        elif isinstance(in_data, NOTICE_METADATA_TYPE):  # ExtractedMetadata
-            notice_metadata = in_data
-
-        if not isinstance(notice_metadata, NOTICE_METADATA_TYPE):
-            raise TypeError('Notice Metadata must be of "%s" type' % NOTICE_METADATA_TYPE.__name__)
-
-        return notice_metadata
+        self.files: List[pathlib.Path] = []
 
     def add_template_files(self):
         file_mets_xml_dmd_rdf = self.tmp_dir_path / FILE_METS_XML_FORMAT.format(notice_id=self.notice_id)
-        self.__write_template_to_file(file_mets_xml_dmd_rdf, TemplateGenerator.mets_xml_dmd_rdf_generator,
-                                      self.template_metadata)
-
         file_tmd_rdf = self.tmp_dir_path / FILE_TMD_FORMAT.format()
-        self.__write_template_to_file(file_tmd_rdf, TemplateGenerator.tmd_rdf_generator, self.template_metadata)
-
         file_mets2action_mets_xml = self.tmp_dir_path / FILE_METS_ACTION_FORMAT.format(
             work_identifier=self.template_metadata.work.identifier,
             action=self.notice_action
         )
-        self.__write_template_to_file(file_mets2action_mets_xml, TemplateGenerator.mets2action_mets_xml_generator,
-                                      self.template_metadata)
-
+        encoding_type = "utf-8"
+        file_mets_xml_dmd_rdf.write_text(TemplateGenerator.mets_xml_dmd_rdf_generator(self.template_metadata),
+                                         encoding=encoding_type
+                                         )
+        file_tmd_rdf.write_text(TemplateGenerator.tmd_rdf_generator(self.template_metadata), encoding=encoding_type)
+        file_mets2action_mets_xml.write_text(TemplateGenerator.mets2action_mets_xml_generator(self.template_metadata),
+                                             encoding=encoding_type)
         self.files = [
             file_mets_xml_dmd_rdf,
             file_tmd_rdf,
             file_mets2action_mets_xml
         ]
 
-    def add_rdf_content(self, rdf_content: RDF_CONTENT_TYPE):
+    def add_rdf_content(self, rdf_content: bytes):
+        """
+
+        :param rdf_content:
+        :return:
+        """
         if rdf_content is not None:
-            if isinstance(rdf_content, str):
-                rdf_content = bytes(rdf_content, 'utf-8')
             try:
                 rdf_content_bytes = base64.b64decode(rdf_content, validate=True)
             except binascii.Error:
                 rdf_content_bytes = rdf_content
             rdf_file_path = self.tmp_dir_path / FILE_RDF_FORMAT.format(notice_id=self.notice_id)
-            self.__write_to_file(rdf_file_path, rdf_content_bytes, 'xb')
+            rdf_file_path.write_bytes(rdf_content_bytes)
             self.files.append(rdf_file_path)
 
-    def add_extra_files(self, extra_files: PATH_LIST_TYPE):
+    def add_extra_files(self, extra_files: List[pathlib.Path]):
+        """
+
+        :param extra_files:
+        :return:
+        """
         if extra_files is not None:
             self.files += extra_files
 
-    def pack(self, save_to: PATH_TYPE) -> str:
+    def get_archive_name(self) -> str:
         archive_name = ARCHIVE_NAME_FORMAT.format(
             work_identifier=self.template_metadata.work.identifier,
             action=self.template_metadata.notice.action.type
         )
-
-        archive_path = self.tmp_dir_path / archive_name
-        package_path = self.archiver.process_archive(archive_path, self.files)
-
-        with open(package_path, "rb") as f:
-            raw_archive_content = f.read()
-
-        if save_to is None:
-            # If no save_to path is specified, then just return the package content as string
-            archive_content = base64.b64encode(raw_archive_content)
-            return str(archive_content, 'utf-8')
-        else:
-            if save_to:
-                # If save_to path is not empty, then create, write the package and return the destination path
-                save_to_path = Path(save_to)
-                if os.path.isdir(save_to_path):
-                    save_to_path /= archive_name
-                self.__write_to_file(save_to_path, raw_archive_content, 'wb')
-                return str(save_to_path)
-            else:
-                # If save_to is empty (save_to="") then return the created package source path
-                return package_path
+        return archive_name
+
+    def pack(self) -> str:
+        archiver = ZipArchiver()
+        archive_path = self.tmp_dir_path / self.get_archive_name()
+        package_path = archiver.process_archive(archive_path, self.files)
+        raw_archive_content = package_path.read_bytes()
+        archive_content = base64.b64encode(raw_archive_content)
+        return str(archive_content, 'utf-8')

tests/features/notice_packager/test_notice_packager.py

@@ -7,9 +7,8 @@
     when,
 )
 
-from ted_sws.core.model.manifestation import METSManifestation
 from ted_sws.core.model.notice import Notice, NoticeStatus
-from ted_sws.notice_packager.services.notice_packager import create_notice_package
+from ted_sws.notice_packager.services.notice_packager import package_notice
 
 
 @scenario('test_notice_packager.feature', 'Package a TED notice in a METS package')
@@ -33,11 +32,8 @@ def the_notice_status_is_eligible_for_packaging(package_eligible_notice):
 @when('the notice packaging is executed', target_fixture="packaged_notice")
 def the_notice_packaging_is_executed(package_eligible_notice):
     """the notice packaging is executed."""
-    rdf_content = package_eligible_notice.distilled_rdf_manifestation.object_data.encode("utf-8")
-    mets_manifestation_content = create_notice_package(in_data=package_eligible_notice, rdf_content=rdf_content)
-    package_eligible_notice.set_mets_manifestation(
-        mets_manifestation=METSManifestation(object_data=mets_manifestation_content))
-    return package_eligible_notice
+    packaged_notice = package_notice(notice=package_eligible_notice)
+    return packaged_notice
 
 
 @then('the notice have METS manifestation')