Feat/build args

pyproject.toml

@@ -38,15 +38,16 @@ websocket-client = "^1.7.0"
 sqlalchemy = "^1.0.0"
 fastapi-utils = "^0.2.1"
 pytz = "^2023.3.post1"
-opentelemetry-distro = "^0.46b0"
-opentelemetry-exporter-otlp = "^1.25.0"
-opentelemetry-instrumentation-requests = "^0.46b0"
-opentelemetry-instrumentation-fastapi = "^0.46b0"
-opentelemetry-instrumentation-sqlalchemy = "^0.46b0"
+opentelemetry-distro = ">0.46b0"
+opentelemetry-exporter-otlp = ">1.25.0"
+opentelemetry-instrumentation-requests = ">0.46b0"
+opentelemetry-instrumentation-fastapi = ">0.46b0"
+opentelemetry-instrumentation-sqlalchemy = ">0.46b0"
 ruamel-yaml = "^0.18.6"
 unidecode = "^1.3.8"
 deepdiff = "^7.0.1"
 setuptools = ">72.0.0"
+fastapi-lifespan-manager = "^0.1.4"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.4.4"

src/config/main.py

@@ -3,6 +3,7 @@
 import logging
 import sys
 from os import getenv
+from typing import List
 
 from deepdiff import DeepDiff
 from opentelemetry import trace
@@ -25,7 +26,8 @@
 from config.optel import OpTel
 from config.server import Server
 from const import URL_CONFIG_FILE
-from utils import edit_config, load_config, logo, str2bool, title
+from database.usage_points import DatabaseUsagePoints
+from utils import barcode_message, edit_config, load_config, logo, str2bool, title
 
 locale.setlocale(locale.LC_ALL, "fr_FR.UTF-8")
 
@@ -88,7 +90,6 @@ def __init__(self):
         self.load_logging()
         self.setup_tracing()
         logo(VERSION)
-        self.display()
 
         comments = None
         for key in self.config.config:
@@ -99,6 +100,7 @@ def __init__(self):
 
         self.check_config()
         if self.dev:
+            barcode_message("DEV MODE")
             exemple_file = "config.example.yaml"
             edit_config(data=self.default, file=exemple_file, comments=comments, wipe=True)
             edit_config(
@@ -108,6 +110,19 @@ def __init__(self):
                 wipe=True,
             )
             title([f"Generate {exemple_file}", f" => {exemple_file} generated"])
+        self.display()
+        self.clean_database()
+
+    def clean_database(self):
+        """Clean database."""
+        title("Nettoyage de la base de données...")
+        usage_point_list: List[UsagePointId] = []
+        if self.myelectricaldata.usage_point_config is not None:
+            for upi, _ in self.myelectricaldata.usage_point_config.items():
+                usage_point_list.append(upi)
+        for usage_point in DatabaseUsagePoints().get_all():
+            if usage_point.usage_point_id not in usage_point_list:
+                DatabaseUsagePoints(usage_point.usage_point_id).delete()
 
     def check_config(self):
         """Check current config file."""
@@ -290,5 +305,5 @@ def tracing_fastapi(self, app):
             logging.debug("[OpenTelemetry] FastAPI loaded")
             FastAPIInstrumentor.instrument_app(app)
 
-
-APP_CONFIG = Config()
+if __name__ == "config.main":
+    APP_CONFIG = Config()

src/database/usage_points.py

@@ -1,6 +1,7 @@
 """Manage UsagePoints table in database."""
 
 from datetime import datetime, timedelta
+from typing import List
 
 from sqlalchemy import delete, select, update
 from sqlalchemy.orm import scoped_session
@@ -72,7 +73,7 @@ def __init__(self, usage_point_id=None):
         self.session: scoped_session = DB.session()
         self.usage_point_config = None
 
-    def get_all(self):
+    def get_all(self) -> List[UsagePoints]:
         """Get all data from usage point table."""
         query = select(UsagePoints)
         data = self.session.scalars(query).all()

src/external_services/myelectricaldata/tempo.py

@@ -20,6 +20,7 @@ class Tempo:
     def __init__(self):
         self.url = URL
         self.valid_date = datetime.combine(datetime.now(tz=TIMEZONE) + relativedelta(days=1), datetime.min.time())
+        self.display_nb_day = 10
         self.nb_check_day = 31
         self.total_tempo_days = {
             "red": 22,
@@ -44,7 +45,7 @@ def run(self):
             query_response = Query(endpoint=target).get()
             if query_response.status_code == CODE_200_SUCCESS:
                 try:
-                    response_json = json.loads(query_response.text)
+                    response_json: dict = json.loads(query_response.text)
                     for date, color in response_json.items():
                         date_obj = datetime.strptime(date, "%Y-%m-%d").replace(tzinfo=TIMEZONE)
                         DatabaseTempo().set(date_obj, color)
@@ -57,11 +58,10 @@ def run(self):
                         "description": "Erreur lors de la récupération de données Tempo.",
                     }
                 return response
-            else:
-                return {
-                    "error": True,
-                    "description": json.loads(query_response.text)["detail"],
-                }
+            return {
+                "error": True,
+                "description": json.loads(query_response.text)["detail"],
+            }
 
     def get(self):
         """Retrieves tempo data from the database.
@@ -110,8 +110,13 @@ def fetch(self):
                 else:
                     logging.info(" => Toutes les données sont déjà en cache.")
             if "error" not in result:
-                for key, value in result.items():
-                    logging.info(f"{key}: {value}")
+                if len(result) > 0:
+                    i = 0
+                    for key, value in result.items():
+                        if i < self.display_nb_day:
+                            logging.info(f"{key}: {value}")
+                            i += 1
+                    logging.info("...")
             else:
                 logging.error(result)
                 return "OK"
@@ -140,6 +145,7 @@ def calc_day(self):
             for day in current_tempo_day:
                 result[day.color.lower()] -= 1
             DatabaseTempo().set_config("days", result)
+            logging.info(" => OK")
             return result
 
     def fetch_day(self):

src/main.py

@@ -1,5 +1,6 @@
 """Main module of the application."""
 
+from contextlib import asynccontextmanager
 from os import listdir
 from pathlib import Path
 
@@ -10,26 +11,38 @@
 from uvicorn.config import LOGGING_CONFIG
 
 from config.main import APP_CONFIG
-from database.usage_points import DatabaseUsagePoints
 from models.jobs import Job
 from routers import account, action, data, html, info
-from utils import get_version, title
+from utils import get_version
 
-usage_point_list = []
-if APP_CONFIG.myelectricaldata.usage_point_config is not None:
-    for upi, _ in APP_CONFIG.myelectricaldata.usage_point_config.items():
-        usage_point_list.append(upi)
 
-title("Nettoyage de la base de données...")
-for usage_point in DatabaseUsagePoints().get_all():
-    if usage_point.usage_point_id not in usage_point_list:
-        DatabaseUsagePoints(usage_point.usage_point_id).delete()
+#######################################################################################################################
+# JOBS
+@repeat_every(seconds=APP_CONFIG.server.cycle, wait_first=False)
+def job_boot():
+    """Bootstap jobs."""
+    Job().boot()
+
+
+@repeat_every(seconds=3600, wait_first=True)
+def job_home_assistant():
+    """Home Assistant Ecowatt."""
+    Job().export_home_assistant(target="ecowatt")
 
-swagger_configuration = {
-    "operationsSorter": "method",
-    "tagsSorter": "alpha",
-    "deepLinking": True,
-}
+
+@repeat_every(seconds=600, wait_first=False)
+def job_gateway_status():
+    """Gateway status check."""
+    Job().get_gateway_status()
+
+
+@asynccontextmanager
+async def bootstrap(app: FastAPI):  # pylint: disable=unused-argument
+    """Bootstap jobs."""
+    await job_boot()
+    await job_home_assistant()
+    await job_gateway_status()
+    yield
 
 
 APP = FastAPI(
@@ -49,6 +62,7 @@
         "tagsSorter": "alpha",
         "deepLinking": True,
     },
+    lifespan=bootstrap,
 )
 
 #######################################################################################################################
@@ -67,30 +81,6 @@
 APP.include_router(action.ROUTER)
 APP.include_router(account.ROUTER)
 
-
-#######################################################################################################################
-# JOB TASKS
-@APP.on_event("startup")
-@repeat_every(seconds=APP_CONFIG.server.cycle, wait_first=False)
-def import_job():
-    """Perform the import job."""
-    Job().boot()
-
-
-@APP.on_event("startup")
-@repeat_every(seconds=3600, wait_first=True)
-def home_assistant_export():
-    """Perform the home assistant export job."""
-    Job().export_home_assistant(target="ecowatt")
-
-
-@APP.on_event("startup")
-@repeat_every(seconds=600, wait_first=False)
-def gateway_status():
-    """Perform gateway status."""
-    Job().get_gateway_status()
-
-
 #######################################################################################################################
 # FastAPI opentelemetry configuration
 APP_CONFIG.tracing_fastapi(APP)
@@ -103,19 +93,23 @@ def gateway_status():
     log_config["formatters"]["access"]["datefmt"] = APP_CONFIG.logging.log_format_date
     log_config["formatters"]["default"]["fmt"] = APP_CONFIG.logging.log_format
     log_config["formatters"]["default"]["datefmt"] = APP_CONFIG.logging.log_format_date
-    uvicorn_params = {}
-    uvicorn_params["log_config"] = log_config
-    uvicorn_params["host"] = APP_CONFIG.server.cidr
-    uvicorn_params["port"] = APP_CONFIG.server.port
-    uvicorn_params["reload"] = True
-    uvicorn_params["reload_dirs"] = [APP_CONFIG.application_path]
-    uvicorn_params["reload_includes"] = [APP_CONFIG.application_path]
-    uvicorn_params["reload_excludes"] = [".venv", ".git/*", ".idea/*", ".vscode/*", ".py[cod]"]
+    uvicorn_params = {
+        "reload": False,
+        "log_config": log_config,
+        "host": APP_CONFIG.server.cidr,
+        "port": APP_CONFIG.server.port,
+        "log_level": "error",
+        "reload_dirs": None,
+        "reload_includes": None,
+        "reload_excludes": None,
+    }
     if APP_CONFIG.logging.log_http:
         uvicorn_params["log_level"] = "info"
-    else:
-        uvicorn_params["log_level"] = "error"
-    uvicorn_params = {**uvicorn_params, **APP_CONFIG.ssl_config.__dict__}
+    if APP_CONFIG.dev:
+        uvicorn_params["reload"] = True
+        uvicorn_params["reload_dirs"] = [APP_CONFIG.application_path]
+        uvicorn_params["reload_includes"] = [APP_CONFIG.application_path]
+        uvicorn_params["reload_excludes"] = [".venv", ".git/*", ".idea/*", ".vscode/*", ".py[cod]"]
 
-    APP_CONFIG.display()
+    uvicorn_params = {**uvicorn_params, **APP_CONFIG.ssl_config.__dict__}
     uvicorn.run("main:APP", **uvicorn_params)

src/models/jobs.py

@@ -40,12 +40,12 @@ def __init__(self, usage_point_id=None):
 
     def boot(self):
         """Boots the import job."""
-        if APP_CONFIG.dev or APP_CONFIG.logging.debug:
+        if APP_CONFIG.dev:
             logging.warning("=> Import job disable")
         else:
             self.job_import_data()
 
-    def job_import_data(self, wait=True, target=None):  # noqa: PLR0912, PLR0915, C901
+    def job_import_data(self, wait=True, target=None):  # noqa: PLR0912, C901
         """Import data from the API."""
         if DB.lock_status():
             return {"status": False, "notif": "Importation déjà en cours..."}
@@ -59,9 +59,6 @@ def job_import_data(self, wait=True, target=None):  # noqa: PLR0912, PLR0915, C9
                 time.sleep(1)
                 i = i - 1
 
-        if target == "gateway_status" or target is None:
-            self.get_gateway_status()
-
         # ######################################################################################################
         # FETCH TEMPO DATA
         if target == "tempo" or target is None:

src/models/query.py

@@ -4,8 +4,6 @@
 
 import requests
 
-from database.config import DatabaseConfig
-from utils import str2bool
 from config.main import APP_CONFIG
 
 

src/utils.py

@@ -322,6 +322,13 @@ def get_version():
     return VERSION
 
 
+def barcode_message(message):
+    """Barcode message."""
+    art = text2art(message)
+    for line in art.splitlines():
+        logging.info(f'{decor("barcode1")}{line: ^93}{decor("barcode1", reverse=True)}')
+
+
 def logo(version):
     """Print the logo of MyElectricalData with the version number.