Merge branch 'main' into feature/TED-4

Author: Dragos0000

.github/workflows/unit-tests-hermes.yml

@@ -27,8 +27,8 @@ jobs:
       - name: Get Saxon
         run: make init-saxon
 
-      - name: Start staging infra
-        run: make start-project-services
+#      - name: Start staging infra
+#        run: make start-project-services
 
       - name: Run unit tests
         run: make test-all

MANIFEST.in

@@ -1 +1 @@
-recursive-include ted_sws *.rq *.json *.jinja2 *.csv
+recursive-include ted_sws *.rq *.json *.jinja2 *.csv *.rml.ttl

Makefile

@@ -97,20 +97,20 @@ build-airflow: guard-ENVIRONMENT create-env-airflow build-externals
 	@ docker-compose -p ${ENVIRONMENT} --file ./infra/airflow/docker-compose.yaml --env-file ${ENV_FILE} up -d --force-recreate
 
 start-airflow: build-externals
-	@ echo -e "$(BUILD_PRINT)Starting Airflow servies $(END_BUILD_PRINT)"
+	@ echo -e "$(BUILD_PRINT)Starting Airflow services $(END_BUILD_PRINT)"
 	@ docker-compose -p ${ENVIRONMENT} --file ./infra/airflow/docker-compose.yaml --env-file ${ENV_FILE} up -d
 
 stop-airflow:
-	@ echo -e "$(BUILD_PRINT)Stoping Airflow services $(END_BUILD_PRINT)"
+	@ echo -e "$(BUILD_PRINT)Stopping Airflow services $(END_BUILD_PRINT)"
 	@ docker-compose -p ${ENVIRONMENT} --file ./infra/airflow/docker-compose.yaml --env-file ${ENV_FILE} down
 
 #	------------------------
 start-allegro-graph: build-externals
-	@ echo -e "$(BUILD_PRINT)Starting Allegro-Graph servies $(END_BUILD_PRINT)"
+	@ echo -e "$(BUILD_PRINT)Starting Allegro-Graph services $(END_BUILD_PRINT)"
 	@ docker-compose -p ${ENVIRONMENT} --file ./infra/allegro-graph/docker-compose.yml --env-file ${ENV_FILE} up -d
 
 stop-allegro-graph:
-	@ echo -e "$(BUILD_PRINT)Stoping Allegro-Graph services $(END_BUILD_PRINT)"
+	@ echo -e "$(BUILD_PRINT)Stopping Allegro-Graph services $(END_BUILD_PRINT)"
 	@ docker-compose -p ${ENVIRONMENT} --file ./infra/allegro-graph/docker-compose.yml --env-file ${ENV_FILE} down
 
 #	------------------------
@@ -281,4 +281,27 @@ refresh-mapping-files:
 #
 #stop-silk-service:
 #	@ echo -e "Stop silk service"
-#	@ cd infra/silk/ && docker-compose down
+#	@ cd infra/silk/ && docker-compose down
+
+
+#-----------------------------------------------------------------------------
+# API Service commands
+#-----------------------------------------------------------------------------
+build-all-apis: build-id_manager-api
+
+start-all-apis: start-id_manager-api
+
+stop-all-apis: stop-id_manager-api
+
+build-id_manager-api:
+	@ echo -e "$(BUILD_PRINT) Build id_manager API service $(END_BUILD_PRINT)"
+	@ docker-compose -p common --file infra/api/docker-compose.yml --env-file ${ENV_FILE} build --no-cache --force-rm
+	@ docker-compose -p common --file infra/api/docker-compose.yml --env-file ${ENV_FILE} up -d --force-recreate
+
+start-id_manager-api:
+	@ echo -e "$(BUILD_PRINT)Starting id_manager API service $(END_BUILD_PRINT)"
+	@ docker-compose -p common --file infra/api/docker-compose.yml --env-file ${ENV_FILE} up -d
+
+stop-id_manager-api:
+	@ echo -e "$(BUILD_PRINT)Stopping id_manager API service $(END_BUILD_PRINT)"
+	@ docker-compose -p common --file infra/api/docker-compose.yml --env-file ${ENV_FILE} down

README.md

@@ -85,8 +85,7 @@ to get the Usage Help:
 ```bash
 Usage: resources_injector [OPTIONS] [MAPPING_SUITE_ID]
 
-  Injects the requested resources from Conceptual Mappings into the
-  MappingSuite
+  Injects the requested resources from Conceptual Mappings into the MappingSuite
 
 Options:
   -i, --opt-conceptual-mappings-file TEXT     Use to overwrite default INPUT
@@ -96,6 +95,27 @@ Options:
   --help                                      Show this message and exit.
 ```
 
+#### CMD: rml_modules_injector
+Injects the requested RML modules from Conceptual Mappings into the MappingSuite.
+
+Use:
+```bash
+rml_modules_injector --help
+```
+to get the Usage Help:
+```bash
+Usage: rml_modules_injector [OPTIONS] [MAPPING_SUITE_ID]
+
+  Injects the requested RML modules from Conceptual Mappings into the MappingSuite
+
+Options:
+  -i, --opt-conceptual-mappings-file TEXT     Use to overwrite default INPUT
+  -o, --opt-output-folder TEXT                Use to overwrite default OUTPUT
+  -r, --opt-rml-modules-folder TEXT
+  -m, --opt-mappings-folder TEXT
+  --help                                      Show this message and exit.
+```
+
 #### CMD: metadata_generator
 Generates metadata.json file from Conceptual Mappings file data.
 
@@ -189,6 +209,7 @@ By default, successively runs the following commands:
 ```bash
 - normalisation_resource_generator
 - resources_injector
+- rml_modules_injector
 - metadata_generator
 - yarrrml2rml_converter
 - sparql_generator
@@ -207,7 +228,7 @@ Usage: mapping_suite_processor [OPTIONS] MAPPING_SUITE_ID
   yarrrml2rml_converter - sparql_generator
 
 Options:
-  -c, --opt-commands [normalisation_resource_generator|resources_injector|metadata_generator|yarrrml2rml_converter|sparql_generator]
+  -c, --opt-commands [normalisation_resource_generator|resources_injector|rml_modules_injector|metadata_generator|yarrrml2rml_converter|sparql_generator]
   -m, --opt-mappings-folder TEXT
   --help                                      Show this message and exit.
 ```
@@ -259,7 +280,7 @@ Options:
 ##### Start local API server
 To start the API server:
 ```bash
-id-manager-api-start-server
+api-id_manager-start-server
 ```
 Output:
 ```bash
@@ -269,11 +290,11 @@ See http://localhost:8000/api/v1/docs for API usage.
 ```
 Use:
 ```bash
-api-start-server --help
+api-id_manager-start-server --help
 ```
 to get the cli command Usage Help:
 ```bash
-Usage: id-manager-api-start-server [OPTIONS]
+Usage: api-id_manager-start-server [OPTIONS]
 
 Options:
   -h, --host TEXT

dags/__init__.py

@@ -3,14 +3,13 @@
 DEFAULT_DAG_ARGUMENTS = {
     "owner": "airflow",
     "depends_on_past": False,
-    "start_date": datetime.now(),
+    "start_date": datetime.now()-timedelta(minutes=2),
     "email": ["info@meaningfy.ws"],
     "email_on_failure": False,
     "email_on_retry": False,
     "retries": 0,
     "retry_delay": timedelta(minutes=3600),
-    "schedule_interval": "@once",
-    "max_active_runs": 128,
-    "concurrency": 128,
-    "execution_timeout": timedelta(hours=24),
+    "max_active_runs": 15,
+    "concurrency": 15,
+    "execution_timeout": timedelta(days=10),
 }

dags/fetch_notices_for_date_range.py

@@ -0,0 +1,61 @@
+from datetime import datetime
+
+from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+from dateutil import rrule
+
+from dags import DEFAULT_DAG_ARGUMENTS
+from airflow.decorators import dag, task
+from airflow.operators.python import get_current_context
+
+from dags.fetch_notices_per_day_worker import DATE_WILD_CARD_KEY
+
+START_DATE_KEY = "start_date"
+END_DATE_KEY = "end_date"
+
+
+def generate_daily_dates(start_date: str, end_date: str) -> list:
+    """
+        Given a date range returns all daily dates in that range
+    :param start_date:
+    :param end_date:
+    :return:
+    """
+    return [dt.strftime('%Y%m%d')
+            for dt in rrule.rrule(rrule.DAILY,
+                                  dtstart=datetime.strptime(start_date, '%Y%m%d'),
+                                  until=datetime.strptime(end_date, '%Y%m%d'))]
+
+
+def generate_wild_card_by_date(date: str) -> str:
+    """
+        Method to build a wildcard_date from a date string
+    :param date:
+    :return:
+    """
+    return f"{date}*"
+
+
+@dag(default_args=DEFAULT_DAG_ARGUMENTS, schedule_interval=None, tags=['master', 'fetch_notices_for_date_range'])
+def fetch_notices_for_date_range():
+    @task
+    def trigger_fetch_notices_workers_for_date_range():
+        context = get_current_context()
+        dag_conf = context["dag_run"].conf
+
+        if START_DATE_KEY not in dag_conf.keys():
+            raise "Config key [start_date] is not present in dag context"
+        if END_DATE_KEY not in dag_conf.keys():
+            raise "Config key [end_date] is not present in dag context"
+
+        for generated_date in generate_daily_dates(dag_conf[START_DATE_KEY], dag_conf[END_DATE_KEY]):
+            wildcard_date = generate_wild_card_by_date(date=generated_date)
+            TriggerDagRunOperator(
+                task_id=f'trigger_fetch_notices_per_day_worker_dag_{wildcard_date[:-1]}',
+                trigger_dag_id="fetch_notices_per_day_worker",
+                conf={DATE_WILD_CARD_KEY: wildcard_date}
+            ).execute(context=context)
+
+    trigger_fetch_notices_workers_for_date_range()
+
+
+dag = fetch_notices_for_date_range()

dags/fetch_notices_per_day_worker.py

@@ -0,0 +1,59 @@
+import datetime
+import time
+from random import randint
+
+from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+
+from dags import DEFAULT_DAG_ARGUMENTS
+from airflow.decorators import dag, task
+from airflow.operators.python import get_current_context
+from pymongo import MongoClient
+
+from dags.index_and_normalise_notice_worker import NOTICE_ID
+from ted_sws import config
+from ted_sws.core.model.notice import NoticeStatus
+from ted_sws.data_manager.adapters.notice_repository import NoticeRepository
+
+from ted_sws.notice_fetcher.adapters.ted_api import TedAPIAdapter, TedRequestAPI
+from ted_sws.notice_fetcher.services.notice_fetcher import NoticeFetcher
+
+DATE_WILD_CARD_KEY = "date_wild_card"
+
+
+@dag(default_args=DEFAULT_DAG_ARGUMENTS, schedule_interval=None, tags=['worker', 'fetch_notices_per_day'])
+def fetch_notices_per_day_worker():
+    @task
+    def fetch_notices_and_trigger_index_and_normalise_notice_worker():
+        context = get_current_context()
+        dag_conf = context["dag_run"].conf
+
+        if DATE_WILD_CARD_KEY not in dag_conf.keys():
+            raise "Config key [date] is not present in dag context"
+
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_ids = NoticeFetcher(notice_repository=NoticeRepository(mongodb_client=mongodb_client),
+                                   ted_api_adapter=TedAPIAdapter(
+                                       request_api=TedRequestAPI())).fetch_notices_by_date_wild_card(
+            wildcard_date=dag_conf[DATE_WILD_CARD_KEY])  # "20220203*"
+        for notice_id in notice_ids:
+            restart_dag_operator = True
+            while restart_dag_operator:
+                restart_dag_operator = False
+                try:
+                    time.sleep(randint(10, 500) / 1000)
+                    TriggerDagRunOperator(
+                        task_id=f'trigger_index_and_normalise_notice_worker_dag_{notice_id}',
+                        trigger_dag_id="index_and_normalise_notice_worker",
+                        trigger_run_id=notice_id,
+                        conf={NOTICE_ID: notice_id}
+                    ).execute(context=context)
+                except Exception as e:
+
+                    restart_dag_operator = True
+                    print("trigger dag operator restarted !!!")
+                    print("EXCEPTION message: ", e)
+
+    fetch_notices_and_trigger_index_and_normalise_notice_worker()
+
+
+dag = fetch_notices_per_day_worker()

dags/index_and_normalise_notice_worker.py

@@ -0,0 +1,58 @@
+from pymongo import MongoClient
+
+from dags.dags_utils import pull_dag_upstream, push_dag_downstream
+from ted_sws import config
+
+from ted_sws.data_manager.adapters.notice_repository import NoticeRepository
+from ted_sws.data_sampler.services.notice_xml_indexer import index_notice
+from ted_sws.metadata_normaliser.services.metadata_normalizer import normalise_notice_by_id
+
+from airflow.decorators import dag, task
+from airflow.operators.python import get_current_context
+
+from dags import DEFAULT_DAG_ARGUMENTS
+
+NOTICE_ID = "notice_id"
+
+
+@dag(default_args=DEFAULT_DAG_ARGUMENTS,
+     schedule_interval=None,
+     tags=['worker', 'index_and_normalise_notice'])
+def index_and_normalise_notice_worker():
+    """
+
+    :return:
+    """
+
+    @task
+    def index_notice_step():
+        """
+
+        :return:
+        """
+        context = get_current_context()
+        dag_params = context["dag_run"].conf
+        notice_id = dag_params[NOTICE_ID]
+        push_dag_downstream(key=NOTICE_ID, value=notice_id)
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+        notice = notice_repository.get(reference=notice_id)
+        indexed_notice = index_notice(notice=notice)
+        notice_repository.update(notice=indexed_notice)
+
+    @task
+    def normalise_notice_metadata_step():
+        """
+
+        :return:
+        """
+        notice_id = pull_dag_upstream(NOTICE_ID)
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+        normalised_notice = normalise_notice_by_id(notice_id=notice_id, notice_repository=notice_repository)
+        notice_repository.update(notice=normalised_notice)
+
+    index_notice_step() >> normalise_notice_metadata_step()
+
+
+dag = index_and_normalise_notice_worker()

dags/load_mapping_suite_in_mongodb.py

@@ -18,7 +18,10 @@
 FINISH_LOADING_MAPPING_SUITE_TASK_ID = "finish_loading_mapping_suite"
 CHECK_IF_LOAD_TEST_DATA_TASK_ID = "check_if_load_test_data"
 
-@dag(default_args=DEFAULT_DAG_ARGUMENTS, tags=['fetch', 'mapping-suite', 'github'])
+
+@dag(default_args=DEFAULT_DAG_ARGUMENTS,
+     schedule_interval=None,
+     tags=['fetch', 'mapping-suite', 'github'])
 def load_mapping_suite_in_mongodb():
     @task
     def fetch_mapping_suite_package_from_github_into_mongodb():