94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
"""
|
|
db.py — Module centralisé d'accès à la base de données MariaDB.
|
|
|
|
Ce module est le SEUL point de contact avec PyMySQL dans l'application.
|
|
Il expose un gestionnaire de connexion/curseur réutilisable par tous les
|
|
modèles de la couche models/, avec :
|
|
- gestion des erreurs et rollback automatique,
|
|
- reconnexion en cas de perte de connexion (ping avant chaque emprunt),
|
|
- curseurs en mode dictionnaire (résultats accessibles par nom de colonne).
|
|
|
|
Toutes les requêtes exécutées via ce module DOIVENT utiliser des paramètres
|
|
préparés (placeholders %s) — jamais de concaténation de chaînes — afin
|
|
d'éliminer tout risque d'injection SQL (Bloc 1 - C4, C8 ; OWASP A03:2021).
|
|
"""
|
|
|
|
import logging
|
|
from contextlib import contextmanager
|
|
|
|
import pymysql
|
|
import pymysql.cursors
|
|
|
|
logger = logging.getLogger("gesthub.db")
|
|
|
|
|
|
class Database:
|
|
"""Petite couche d'abstraction au-dessus de PyMySQL (pas d'ORM)."""
|
|
|
|
def __init__(self, host, port, user, password, database):
|
|
self.host = host
|
|
self.port = port
|
|
self.user = user
|
|
self.password = password
|
|
self.database = database
|
|
|
|
def _connect(self):
|
|
return pymysql.connect(
|
|
host=self.host,
|
|
port=self.port,
|
|
user=self.user,
|
|
password=self.password,
|
|
database=self.database,
|
|
cursorclass=pymysql.cursors.DictCursor,
|
|
autocommit=False,
|
|
charset="utf8mb4",
|
|
)
|
|
|
|
@contextmanager
|
|
def cursor(self, commit=False):
|
|
"""Context manager fournissant un curseur et gérant transaction/erreurs.
|
|
|
|
Usage :
|
|
with db.cursor(commit=True) as cur:
|
|
cur.execute("INSERT INTO annonces (titre) VALUES (%s)", (titre,))
|
|
"""
|
|
conn = self._connect()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
yield cur
|
|
if commit:
|
|
conn.commit()
|
|
except pymysql.MySQLError:
|
|
conn.rollback()
|
|
logger.exception("Erreur SQL — transaction annulée (rollback)")
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
def fetch_all(self, query, params=None):
|
|
with self.cursor() as cur:
|
|
cur.execute(query, params or ())
|
|
return cur.fetchall()
|
|
|
|
def fetch_one(self, query, params=None):
|
|
with self.cursor() as cur:
|
|
cur.execute(query, params or ())
|
|
return cur.fetchone()
|
|
|
|
def execute(self, query, params=None):
|
|
"""INSERT/UPDATE/DELETE avec commit automatique. Retourne lastrowid."""
|
|
with self.cursor(commit=True) as cur:
|
|
cur.execute(query, params or ())
|
|
return cur.lastrowid
|
|
|
|
|
|
def init_db(config):
|
|
"""Fabrique une instance Database à partir d'un objet de configuration."""
|
|
return Database(
|
|
host=config.DB_HOST,
|
|
port=config.DB_PORT,
|
|
user=config.DB_USER,
|
|
password=config.DB_PASSWORD,
|
|
database=config.DB_NAME,
|
|
)
|