# ============================================================ # config.py — Configuration & connexion SQL Server # Data Sentinel | COYAUD Anthony | 2026 # ============================================================ import os import pyodbc from contextlib import contextmanager def _build_connection_string() -> str: """ En déploiement (variables d'environnement présentes), on se connecte en authentification SQL. Sinon, on garde la connexion Windows locale par défaut (poste de dev). """ server = os.getenv("DB_SERVER") if server: driver = os.getenv("DB_DRIVER", "ODBC Driver 18 for SQL Server") port = os.getenv("DB_PORT", "1433") return ( f"Driver={{{driver}}};" f"Server={server},{port};" f"Database={os.getenv('DB_NAME', 'DataSentinel')};" f"UID={os.getenv('DB_USER')};" f"PWD={os.getenv('DB_PASSWORD')};" "Encrypt=yes;TrustServerCertificate=yes;" ) return ( "Driver={ODBC Driver 17 for SQL Server};" "Server=LaptopCA\\SQLEXPRESS;" "Database=DataSentinel;" "Trusted_Connection=yes;" ) class Config: # Chaîne de connexion SQL Server (env en prod, Windows en local) DB_CONNECTION_STRING = _build_connection_string() # Paramètres API API_TITLE = "Data Sentinel API" API_VERSION = "1.0.0" API_DESCRIPTION = "API de monitoring de la qualité des données — XEFI" # Sécurité JWT SECRET_KEY = os.getenv("JWT_SECRET", "data-sentinel-secret-change-in-prod") ALGORITHM = os.getenv("JWT_ALGORITHM", "HS256") TOKEN_EXPIRE_MINUTES = int(os.getenv("JWT_EXPIRE_MINUTES", "60")) # ---------------------------------------------------------------- # Connexion — retourne un curseur pyodbc via context manager # ---------------------------------------------------------------- def get_connection() -> pyodbc.Connection: """Ouvre une connexion SQL Server et la retourne.""" return pyodbc.connect(Config.DB_CONNECTION_STRING) @contextmanager def get_cursor(): """ Context manager : ouvre connexion + curseur, commit ou rollback, ferme proprement. Usage : with get_cursor() as cursor: cursor.execute("SELECT ...") rows = cursor.fetchall() """ conn = get_connection() cursor = conn.cursor() try: yield cursor conn.commit() except Exception: conn.rollback() raise finally: cursor.close() conn.close()