improve logging

dags/__init__.py

@@ -0,0 +1,43 @@
+import logging
+import logging.config
+
+LOGGING_CONFIG = {
+    'version': 1,
+    'disable_existing_loggers': False,
+    'formatters': {
+        'default': {
+            'format': '[%(asctime)s] %(levelname)s in %(name)s: %(message)s',
+        },
+        'verbose': {
+            'format': '[%(asctime)s] %(levelname)s [%(name)s:%(lineno)d] %(message)s',
+        },
+    },
+    'handlers': {
+        'console': {
+            'class': 'logging.StreamHandler',
+            'formatter': 'default',
+        },
+        # Exemple fichier
+        'file': {
+            'class': 'logging.FileHandler',
+            'filename': 'app.log',
+            'formatter': 'verbose',
+        },
+    },
+    'root': {
+        'handlers': ['console', 'file'],
+        'level': 'INFO',
+    },
+    'loggers': {
+        'myapp': {
+            'handlers': ['console'],
+            'level': 'DEBUG',
+            'propagate': False,
+        },
+    }
+}
+
+def setup_logging():
+    logging.config.dictConfig(LOGGING_CONFIG)
+
+setup_logging()

dags/common.py

@@ -8,6 +8,8 @@ from datetime import datetime, timedelta
 from typing import Generator
 from contextlib import contextmanager
 
+logger = logging.getLogger(__name__)
+
 # Load environment variables from .env file
 load_dotenv()
 
@@ -63,7 +65,7 @@ def check_db_connection():
     try:
         with get_session() as session:
             session.execute(text("SELECT 1"))
-            logging.info("Database connection is successful.")
+            logger.info("Database connection is successful.")
     except Exception as e:
-        logging.error(f"Database connection failed: {e}")
+        logger.error(f"Database connection failed: {e}")
         raise

dags/new_transaction.py

@@ -16,6 +16,8 @@ from http.client import (
 )
 from pathlib import Path
 
+logger = logging.getLogger(__name__)
+
 # Load environment variables from .env file
 load_dotenv(Path(__file__).parent.parent / ".env")
 
@@ -39,7 +41,7 @@ def _pull_transaction(ti, prefix: str = ''):
     # If status code is 429 or 500, wait for a few seconds and retry
     if response.status_code in [TOO_MANY_REQUESTS, INTERNAL_SERVER_ERROR]:
         waiting_time = 15
-        logging.warning(f"Rate limit exceeded. Retrying in {waiting_time} seconds...")
+        logger.warning(f"Rate limit exceeded. Retrying in {waiting_time} seconds...")
         time.sleep(waiting_time)
         response = get_current_transaction()
 
@@ -55,7 +57,7 @@ def _pull_transaction(ti, prefix: str = ''):
     # Push the transaction dictionary to XCom
     ti.xcom_push(key=f"{prefix}_transaction_dict", value=transaction_dict)
     
-    logging.info(f"Fetched data: {transaction_dict}")
+    logger.info(f"Fetched data: {transaction_dict}")
 
 @task(task_id="push_transaction")
 def _push_transaction(ti, prefix: str = ''):
@@ -93,7 +95,7 @@ def _push_transaction(ti, prefix: str = ''):
 
     # Check if the transaction dictionary is empty
     if not transaction_dict:
-        logging.error("No transaction data found.")
+        logger.error("No transaction data found.")
         return
     
     # Call the fraud detection pipeline with the transaction dictionary
@@ -116,7 +118,7 @@ def _push_transaction(ti, prefix: str = ''):
     # Load the JSON data
     data = api_response.json()
 
-    logging.info(f"Fraud detection response: {data}")
+    logger.info(f"Fraud detection response: {data}")
 
 # Copy default_args from the original code
 dag_args = default_args.copy()

dags/refresh_views.py

@@ -5,6 +5,8 @@ from datetime import datetime, timedelta
 import logging
 from sqlalchemy import text
 
+logger = logging.getLogger(__name__)
+
 @task(task_id="refresh_views")
 def _refresh_views():
     """
@@ -17,7 +19,7 @@ def _refresh_views():
         
         for view in list_views:
             view_name = view[0]
-            logging.info(f"Refreshing materialized view: {view_name}")
+            logger.info(f"Refreshing materialized view: {view_name}")
             session.execute(text(f"REFRESH MATERIALIZED VIEW CONCURRENTLY public.{view_name} WITH DATA;"))
         
             session.commit()