Test orchestration dbt

backend/bloom/usecase/create_kpler_ais_messages.py

@@ -28,7 +28,6 @@ async def run(self, dump_path: str):
         orm_data = []
 
         try:
-            process_start=datetime.now(timezone.utc)
             current_datetime=None
             position_count= None
 

backend/dbt_trawlwatch/package-lock.yml

@@ -1,5 +1,5 @@
 packages:
-- package: dbt-labs/dbt_utils
-  name: dbt_utils
-  version: 1.3.0
-sha1_hash: 8067dd74fb58d3d05a437ab1975d5eeeaf3d8bea
+  - name: dbt_utils
+    package: dbt-labs/dbt_utils
+    version: 1.3.0
+sha1_hash: 226ae69cdfbc9367e2aa2c472b01f99dbce11de0

backend/load_spire_data_from_api.py

@@ -1,6 +1,6 @@
 import argparse
 import json
-from datetime import datetime, timezone, timedelta
+from datetime import datetime, timezone
 from pathlib import Path
 from time import perf_counter
 
@@ -22,7 +22,6 @@ def run(dump_path: str) -> None:
 
     orm_data = []
     try:
-        process_start=datetime.now(timezone.utc)
         current_datetime=None
         position_count= None
         with db.session() as session:
@@ -48,7 +47,7 @@ def run(dump_path: str) -> None:
                         position_count=len(raw_vessels)
                         if dump_path is not None:
                             try:
-                                now =current_datetime.strftime("%Y-%m-%dT%H:%M:%S")
+                                now = current_datetime.strftime("%Y-%m-%dT%H:%M:%S")
                                 dump_file = Path(args.dump_path, f"spire_{now}").with_suffix(".json")
                                 with dump_file.open("wt") as handle:
                                     json.dump(raw_vessels, handle)
@@ -71,11 +70,6 @@ def run(dump_path: str) -> None:
                         session.commit()
                         raise(e)
                     session.commit()
-                    if current_datetime != None:
-                        TaskExecutionRepository.set_duration(session,
-                                                            "load_spire_data_from_api",
-                                                            current_datetime,
-                                                            datetime.now(timezone.utc)-process_start)
                     if position_count != None:
                         TaskExecutionRepository.set_position_count(session,
                                                         "load_spire_data_from_api",

clevercloud/cron.json

@@ -1,3 +1,4 @@
 [
-    "*/15 * * * * $ROOT/cron.sh"
+    "*/15 * * * * $ROOT/cron_spire_endpoint.sh",
+    "*/15 * * * * $ROOT/cron_kpler_endpoint.sh"
 ]

cron_spire_endpoint.sh

@@ -0,0 +1,10 @@
+#! /bin/bash -l
+
+source ${APP_HOME}/backend/.venv/bin/activate
+python ${APP_HOME}/backend/load_spire_data_from_api.py &&\
+cd ${APP_HOME}/backend/dbt_trawlwatch &&\
+dbt deps &&\
+dbt run --select observ_spire_ais_data_retrievals &&\
+dbt run --select itm_vessel_last_raw_position &&\
+dbt run --select mart_dim_vessels__last_positions
+deactivate