fixed lint errors

Author: qcdyx

api/src/database/database.py

@@ -10,7 +10,6 @@
 from database_gen.sqlacodegen_models import Base, Feed, Gtfsfeed, Gtfsrealtimefeed, Gbfsfeed
 from sqlalchemy.orm import sessionmaker
 import logging
-from typing import Final
 
 lock = threading.Lock()
 
@@ -76,7 +75,7 @@ class Database:
 
     def __new__(cls, *args, **kwargs):
         if not isinstance(cls.instance, cls):
-            with lock:
+            with cls.lock:
                 if not isinstance(cls.instance, cls):
                     cls.instance = object.__new__(cls)
         return cls.instance

api/tests/integration/test_database.py

@@ -3,7 +3,6 @@
 
 import pytest
 from sqlalchemy.orm import Query
-import os
 
 from database.database import Database, generate_unique_id
 from database_gen.sqlacodegen_models import Feature, Gtfsdataset

functions-python/batch_process_dataset/src/main.py

@@ -239,7 +239,7 @@ def create_dataset(self, dataset_file: DatasetFile):
                     session.add(latest_dataset)
                 session.add(new_dataset)
 
-                db.refresh_materialized_view(t_feedsearch.name)  # todo: ????????
+                db.refresh_materialized_view(session, t_feedsearch.name)
                 session.commit()
                 logging.info(f"[{self.feed_stable_id}] Dataset created successfully.")
         except Exception as e:
@@ -310,49 +310,52 @@ def process_dataset(cloud_event: CloudEvent):
     stable_id = "UNKNOWN"
     execution_id = "UNKNOWN"
     bucket_name = os.getenv("DATASETS_BUCKET_NANE")
-    start_db_session(os.getenv("FEEDS_DATABASE_URL"))
-    maximum_executions = os.getenv("MAXIMUM_EXECUTIONS", 1)
-    public_hosted_datasets_url = os.getenv("PUBLIC_HOSTED_DATASETS_URL")
-    trace_service = None
-    dataset_file: DatasetFile = None
-    error_message = None
+    db = Database(database_url=os.getenv("FEEDS_DATABASE_URL"))
     try:
-        #  Extract  data from message
-        data = base64.b64decode(cloud_event.data["message"]["data"]).decode()
-        json_payload = json.loads(data)
-        logging.info(
-            f"[{json_payload['feed_stable_id']}] JSON Payload: {json.dumps(json_payload)}"
-        )
-        stable_id = json_payload["feed_stable_id"]
-        execution_id = json_payload["execution_id"]
-        trace_service = DatasetTraceService()
-
-        trace = trace_service.get_by_execution_and_stable_ids(execution_id, stable_id)
-        logging.info(f"[{stable_id}] Dataset trace: {trace}")
-        executions = len(trace) if trace else 0
-        logging.info(
-            f"[{stable_id}] Dataset executed times={executions}/{maximum_executions} "
-            f"in execution=[{execution_id}] "
-        )
+        with db.start_db_session():
+            maximum_executions = os.getenv("MAXIMUM_EXECUTIONS", 1)
+            public_hosted_datasets_url = os.getenv("PUBLIC_HOSTED_DATASETS_URL")
+            trace_service = None
+            dataset_file: DatasetFile = None
+            error_message = None
+            #  Extract  data from message
+            data = base64.b64decode(cloud_event.data["message"]["data"]).decode()
+            json_payload = json.loads(data)
+            logging.info(
+                f"[{json_payload['feed_stable_id']}] JSON Payload: {json.dumps(json_payload)}"
+            )
+            stable_id = json_payload["feed_stable_id"]
+            execution_id = json_payload["execution_id"]
+            trace_service = DatasetTraceService()
 
-        if executions > 0:
-            if executions >= maximum_executions:
-                error_message = f"[{stable_id}] Function already executed maximum times in execution: [{execution_id}]"
-                logging.error(error_message)
-                return error_message
-
-        processor = DatasetProcessor(
-            json_payload["producer_url"],
-            json_payload["feed_id"],
-            stable_id,
-            execution_id,
-            json_payload["dataset_hash"],
-            bucket_name,
-            int(json_payload["authentication_type"]),
-            json_payload["api_key_parameter_name"],
-            public_hosted_datasets_url,
-        )
-        dataset_file = processor.process()
+            trace = trace_service.get_by_execution_and_stable_ids(
+                execution_id, stable_id
+            )
+            logging.info(f"[{stable_id}] Dataset trace: {trace}")
+            executions = len(trace) if trace else 0
+            logging.info(
+                f"[{stable_id}] Dataset executed times={executions}/{maximum_executions} "
+                f"in execution=[{execution_id}] "
+            )
+
+            if executions > 0:
+                if executions >= maximum_executions:
+                    error_message = f"[{stable_id}] Function already executed maximum times in execution: [{execution_id}]"
+                    logging.error(error_message)
+                    return error_message
+
+            processor = DatasetProcessor(
+                json_payload["producer_url"],
+                json_payload["feed_id"],
+                stable_id,
+                execution_id,
+                json_payload["dataset_hash"],
+                bucket_name,
+                int(json_payload["authentication_type"]),
+                json_payload["api_key_parameter_name"],
+                public_hosted_datasets_url,
+            )
+            dataset_file = processor.process()
     except Exception as e:
         logging.error(e)
         error_message = f"[{stable_id}] Error execution: [{execution_id}] error: [{e}]"

functions-python/helpers/database.py

@@ -17,15 +17,12 @@
 from contextlib import contextmanager
 import os
 import threading
-from typing import Final, Optional, TYPE_CHECKING
+from typing import Final, Optional
 
 from sqlalchemy import create_engine, text
 from sqlalchemy.orm import sessionmaker
 import logging
 
-if TYPE_CHECKING:
-    from sqlalchemy.engine import Engine
-
 DB_REUSE_SESSION: Final[str] = "DB_REUSE_SESSION"
 
 
@@ -55,7 +52,7 @@ class Database:
 
     def __new__(cls, *args, **kwargs):
         if not isinstance(cls.instance, cls):
-            with lock:
+            with cls.lock:
                 if not isinstance(cls.instance, cls):
                     cls.instance = object.__new__(cls)
         return cls.instance