Merge pull request #213 from OP-TED/feature/TED-563

Feature/ted 563

Author: kaleanych

dags/old_worker_single_notice_process_orchestrator.py

@@ -1,31 +1,30 @@
+from airflow.decorators import dag
+from airflow.operators.python import get_current_context, BranchPythonOperator, PythonOperator
 from airflow.utils.trigger_rule import TriggerRule
 from pymongo import MongoClient
 
+from dags import DEFAULT_DAG_ARGUMENTS
 from dags.dags_utils import pull_dag_upstream, push_dag_downstream
 from ted_sws import config
 from ted_sws.core.model.manifestation import METSManifestation
 from ted_sws.core.model.notice import NoticeStatus
 from ted_sws.data_manager.adapters.mapping_suite_repository import MappingSuiteRepositoryMongoDB
 from ted_sws.data_manager.adapters.notice_repository import NoticeRepository
 from ted_sws.data_sampler.services.notice_xml_indexer import index_notice_by_id
+from ted_sws.event_manager.adapters.event_log_decorator import event_log
+from ted_sws.event_manager.adapters.event_logger import EventLogger
+from ted_sws.event_manager.model.event_message import NoticeEventMessage, EventMessageProcessType, EventMessageMetadata, \
+    TechnicalEventMessage
+from ted_sws.event_manager.services.logger_from_context import get_logger_from_dag_context, \
+    handle_event_message_metadata_dag_context, get_task_id_from_dag_context
 from ted_sws.notice_metadata_processor.services.metadata_normalizer import normalise_notice_by_id
-
-from airflow.decorators import dag
-from airflow.operators.python import get_current_context, BranchPythonOperator, PythonOperator
-
-from dags import DEFAULT_DAG_ARGUMENTS
 from ted_sws.notice_metadata_processor.services.notice_eligibility import notice_eligibility_checker_by_id
 from ted_sws.notice_packager.services.notice_packager import create_notice_package
+from ted_sws.notice_publisher.services.notice_publisher import publish_notice_by_id
 from ted_sws.notice_transformer.adapters.rml_mapper import RMLMapper
 from ted_sws.notice_transformer.services.notice_transformer import transform_notice_by_id
 from ted_sws.notice_validator.services.shacl_test_suite_runner import validate_notice_by_id_with_shacl_suite
 from ted_sws.notice_validator.services.sparql_test_suite_runner import validate_notice_by_id_with_sparql_suite
-from ted_sws.event_manager.adapters.event_logger import EventLogger
-from ted_sws.event_manager.adapters.event_log_decorator import event_log
-from ted_sws.event_manager.model.event_message import NoticeEventMessage, EventMessageProcessType, EventMessageMetadata, \
-    TechnicalEventMessage
-from ted_sws.event_manager.services.logger_from_context import get_logger_from_dag_context, \
-    handle_event_message_metadata_dag_context, get_task_id_from_dag_context
 from ted_sws.notice_validator.services.xpath_coverage_runner import validate_xpath_coverage_notice_by_id
 
 NOTICE_ID = "notice_id"
@@ -285,10 +284,18 @@ def _generate_mets_package(**context_args):
 
     def _check_package_integrity_by_package_structure():
         notice_id = pull_dag_upstream(NOTICE_ID)
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+        notice = notice_repository.get(reference=notice_id)
+        notice.update_status_to(NoticeStatus.ELIGIBLE_FOR_PUBLISHING)
+        notice_repository.update(notice=notice)
         push_dag_downstream(NOTICE_ID, notice_id)
 
     def _publish_notice_in_cellar():
         notice_id = pull_dag_upstream(NOTICE_ID)
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+        publish_notice_by_id(notice_id=notice_id, notice_repository=notice_repository)
         push_dag_downstream(NOTICE_ID, notice_id)
 
     def _check_notice_public_availability_in_cellar():

dags/worker_single_notice_process_orchestrator.py

@@ -19,6 +19,7 @@
 from ted_sws.notice_metadata_processor.services.metadata_normalizer import normalise_notice
 from ted_sws.notice_metadata_processor.services.notice_eligibility import notice_eligibility_checker
 from ted_sws.notice_packager.services.notice_packager import create_notice_package
+from ted_sws.notice_publisher.services.notice_publisher import publish_notice_by_id
 from ted_sws.notice_transformer.adapters.rml_mapper import RMLMapper
 from ted_sws.notice_transformer.services.notice_transformer import transform_notice
 from ted_sws.notice_validator.services.shacl_test_suite_runner import validate_notice_with_shacl_suite
@@ -200,7 +201,6 @@ def _validate_transformed_rdf_manifestation(**context_args):
         validate_xpath_coverage_notice(notice=notice, mapping_suite=mapping_suite, mongodb_client=mongodb_client)
         push_dag_downstream(NOTICE_OBJECT, notice)
         context = get_current_context()
-
         handle_event_message_metadata_dag_context(event_message, context)
         event_message.notice_id = notice_id
         event_message.domain_action = get_task_id_from_dag_context(context)
@@ -261,10 +261,18 @@ def _generate_mets_package(**context_args):
 
     def _check_package_integrity_by_package_structure():
         notice_id = pull_dag_upstream(NOTICE_ID)
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+        notice = notice_repository.get(reference=notice_id)
+        notice.update_status_to(NoticeStatus.ELIGIBLE_FOR_PUBLISHING)
+        notice_repository.update(notice=notice)
         push_dag_downstream(NOTICE_ID, notice_id)
 
     def _publish_notice_in_cellar():
         notice_id = pull_dag_upstream(NOTICE_ID)
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+        publish_notice_by_id(notice_id=notice_id, notice_repository=notice_repository)
         push_dag_downstream(NOTICE_ID, notice_id)
 
     def _check_notice_public_availability_in_cellar():

infra/sftp/docker-compose.yml

@@ -11,7 +11,7 @@ services:
       - SFTP_PORT=${SFTP_PORT}
     command: ${SFTP_USER}:${SFTP_PASSWORD}:::upload
     ports:
-      - ${SFTP_PORT:-2235}:${SFTP_PORT}
+      - ${SFTP_PORT:-2235}:22
     networks:
       - sftp-net
       - proxy-net

ted_sws/notice_publisher/adapters/notice_publisher.py

@@ -1,25 +1,28 @@
 import pysftp
 
-from ted_sws.notice_publisher.adapters.notice_publisher_abc import NoticePublisherABC
 from ted_sws import config
+from ted_sws.notice_publisher.adapters.sftp_publisher_abc import SFTPPublisherABC
 
 
-class SFTPNoticePublisher(NoticePublisherABC):
+class SFTPPublisher(SFTPPublisherABC):
+    """
+
+    """
+
     connection: pysftp.Connection = None
     default_host = config.SFTP_HOST
     default_user = config.SFTP_USER
     default_pass = config.SFTP_PASSWORD
     default_port = config.SFTP_PORT
 
-    def __init__(self, hostname=default_host, username=default_user, password=default_pass, port=default_port,
-                 remote_path=None):
+    def __init__(self, hostname=default_host, username=default_user, password=default_pass, port=default_port):
         """Constructor Method"""
         # Set connection object to None (initial value)
         self.hostname = hostname
         self.username = username
         self.password = password
-        self.port = port or self.default_port
-        self.remote_path = remote_path
+        self.port = port
+        self.is_connected = False
 
     def connect(self):
         """Connects to the sftp server and returns the sftp connection object"""
@@ -39,30 +42,31 @@ def connect(self):
         except Exception as err:
             raise Exception(err)
 
+        self.is_connected = True
+
     def disconnect(self):
         """Closes the sftp connection"""
-        self.connection.close()
+        if self.is_connected:
+            self.connection.close()
+            self.is_connected = False
 
-    def publish(self, source_path, remote_path=None) -> bool:
-        return self.upload(source_path, remote_path)
+    def __del__(self):
+        self.disconnect()
 
-    def upload(self, source_path, remote_path=None) -> bool:
+    def publish(self, source_path, remote_path) -> bool:
         """
-       Uploads the notice's METS manifestation to the sftp server remote path.
+        Publish file_content to the sftp server remote path.
        """
-        if remote_path is None:
-            remote_path = self.remote_path
-
-        if remote_path is None:
-            raise ValueError("No remote path specified.")
-
         try:
             self.connection.put(source_path, remote_path)
             return True
         except Exception as err:
             raise Exception(err)
 
     def remove(self, remote_path) -> bool:
+        """
+
+        """
         try:
             self.connection.unlink(remote_path)
             return True

ted_sws/notice_publisher/adapters/notice_publisher_abc.py

@@ -0,0 +1,21 @@
+import abc
+
+
+class SFTPPublisherABC(abc.ABC):
+
+    @abc.abstractmethod
+    def connect(self):
+        """
+
+        """
+
+    @abc.abstractmethod
+    def publish(self, source_path, remote_path):
+        """
+
+        """
+
+    def disconnect(self):
+        """
+
+        """

ted_sws/notice_publisher/adapters/sftp_notice_publisher.py

@@ -1,80 +1,47 @@
 import base64
+import pathlib
 import tempfile
-from pathlib import Path
-from typing import Union
 
 from ted_sws import config
-from ted_sws.core.model.notice import Notice, NoticeStatus, UnsupportedStatusTransition
+from ted_sws.core.model.notice import Notice, NoticeStatus
 from ted_sws.data_manager.adapters.notice_repository import NoticeRepositoryABC
-from ted_sws.notice_packager.adapters.archiver import ARCHIVE_DEFAULT_FORMAT
-from ted_sws.notice_publisher.adapters.notice_publisher import NoticePublisherFactory
-from ted_sws.notice_publisher.adapters.notice_publisher_abc import NoticePublisherABC
-
-
-class NoticePublishBuilder:
-    def __init__(self, notice: Union[Notice, str], notice_repository: NoticeRepositoryABC,
-                 notice_publisher: NoticePublisherABC, remote_path: str = None):
-        self.notice_repository = notice_repository
-        if isinstance(notice, str):
-            notice_id = notice
-            notice: Notice = self.notice_repository.get(reference=notice_id)
-            if notice is None:
-                raise Exception(f"Notice {notice_id} could not be found.")
-
-        self.notice = notice
-        self.check_publish_eligibility()
-
-        self.notice_publisher = notice_publisher
-        self.remote_path = remote_path
-
-    def check_publish_eligibility(self):
-        if self.notice.status != NoticeStatus.ELIGIBLE_FOR_PUBLISHING:
-            raise UnsupportedStatusTransition(
-                f"Notice {self.notice.ted_id} is not Eligible for Publishing. Status: {self.notice.status}")
-
-    def package_content(self) -> bytes:
-        mets_manifestation = self.notice.mets_manifestation
-        if not mets_manifestation or not mets_manifestation.object_data:
-            raise ValueError("Notice does not have a METS manifestation to be published.")
-
-        package_content = base64.b64decode(bytes(mets_manifestation.object_data, encoding='utf-8'), validate=True)
-        return package_content
-
-    def build_remote_path(self) -> str:
-        remote_path = self.remote_path
-        if remote_path is None:
-            remote_path = f"{config.SFTP_PATH}/{self.notice.ted_id}"
-            if ARCHIVE_DEFAULT_FORMAT == "zip":
-                remote_path += '.zip'
-        return remote_path
-
-    def publish(self):
-        source_file = tempfile.NamedTemporaryFile()
-        source_file.write(self.package_content())
-
-        try:
-            if self.notice_publisher.publish(source_path=Path(source_file.name), remote_path=self.build_remote_path()):
-                self.notice.update_status_to(NoticeStatus.PUBLISHED)
-                self.notice_repository.update(self.notice)
-        except Exception as e:
-            raise Exception(f"Notice {self.notice.ted_id} could not be published: " + str(e))
-
-        return self.notice.status == NoticeStatus.PUBLISHED
-
-
-def publish_notice(notice: Union[Notice, str], notice_publisher: NoticePublisherABC,
-                   notice_repository: NoticeRepositoryABC, remote_path=None) -> bool:
-    publish_builder = NoticePublishBuilder(notice, notice_repository, notice_publisher, remote_path)
-    return publish_builder.publish()
-
-
-def publish_single_notice(notice: Union[Notice, str], notice_repository: NoticeRepositoryABC, hostname, username,
-                          password, port=None, remote_path=None) -> bool:
-    notice_publisher: NoticePublisherABC = NoticePublisherFactory.get_publisher(hostname=hostname, username=username,
-                                                                                password=password, port=port,
-                                                                                remote_path=remote_path)
-    notice_publisher.connect()
-    notice_published: bool = publish_notice(notice, notice_publisher, notice_repository, remote_path)
-    notice_publisher.disconnect()
-
-    return notice_published
+from ted_sws.notice_publisher.adapters.sftp_notice_publisher import SFTPPublisher
+from ted_sws.notice_publisher.adapters.sftp_publisher_abc import SFTPPublisherABC
+
+
+def publish_notice(notice: Notice, publisher: SFTPPublisherABC = SFTPPublisher(),
+                   remote_folder_path=config.SFTP_PATH) -> bool:
+    """
+        This function publishes the METS manifestation for a Notice in Cellar.
+    """
+
+    mets_manifestation = notice.mets_manifestation
+    if not mets_manifestation or not mets_manifestation.object_data:
+        raise ValueError("Notice does not have a METS manifestation to be published.")
+
+    package_content = base64.b64decode(bytes(mets_manifestation.object_data, encoding='utf-8'), validate=True)
+    remote_notice_path = f"{remote_folder_path}/{notice.ted_id}.zip"
+    source_file = tempfile.NamedTemporaryFile()
+    source_file.write(package_content)
+    try:
+        publisher.connect()
+        if publisher.publish(source_path=pathlib.Path(source_file.name),
+                             remote_path=remote_notice_path):
+            notice.update_status_to(NoticeStatus.PUBLISHED)
+        publisher.disconnect()
+    except Exception as e:
+        raise Exception(f"Notice {notice.ted_id} could not be published: " + str(e))
+
+    return notice.status == NoticeStatus.PUBLISHED
+
+
+def publish_notice_by_id(notice_id: str, notice_repository: NoticeRepositoryABC,
+                         publisher: SFTPPublisherABC = SFTPPublisher(), remote_folder_path=config.SFTP_PATH) -> bool:
+    """
+        This function publishes the METS manifestation of a Notice, based on notice_id, in Cellar.
+    """
+    notice = notice_repository.get(reference=notice_id)
+    result = publish_notice(notice=notice, publisher=publisher, remote_folder_path=remote_folder_path)
+    if result:
+        notice_repository.update(notice=notice)
+    return result