refactor: factorise la création d'engine et normalise solde_montant

- connection: création de l'engine centralisée dans _build_engine (source
  unique de config) ; init_db réutilise reset_connection + get_engine au lieu
  de dupliquer create_engine
- service: _normalize_amount garantit qu'un montant non numérique issu de
  l'extraction (string, dict…) n'entre jamais en base dans une colonne Float ;
  appliqué à solde_montant
- tests: couverture de _normalize_amount et du solde sous forme de string

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-11 10:15:09 +02:00
parent 635a094591
commit 4606d6785b
3 changed files with 60 additions and 29 deletions

View File

@@ -36,23 +36,27 @@ def get_db_path() -> Path:
return DEFAULT_DB_PATH
def _build_engine(db_path: Path):
"""Create a SQLAlchemy engine for the given SQLite path.
Single source of truth for engine configuration.
"""
# Create parent directory if needed
db_path.parent.mkdir(parents=True, exist_ok=True)
return create_engine(
f"sqlite:///{db_path}",
echo=False, # Set to True for SQL debugging
connect_args={"check_same_thread": False}, # Required for FastAPI
)
def get_engine(db_path: Path | None = None):
"""Get or create SQLAlchemy engine (singleton pattern)."""
global _engine
if _engine is None:
if db_path is None:
db_path = get_db_path()
# Create parent directory if needed
db_path.parent.mkdir(parents=True, exist_ok=True)
# Create engine with SQLite
_engine = create_engine(
f"sqlite:///{db_path}",
echo=False, # Set to True for SQL debugging
connect_args={"check_same_thread": False}, # Required for FastAPI
)
_engine = _build_engine(db_path or get_db_path())
return _engine
@@ -87,25 +91,13 @@ def init_db(db_path: Path | None = None) -> Path:
if db_path is None:
db_path = get_db_path()
# Reset globals to use new path
global _engine, _SessionLocal
_engine = None
_SessionLocal = None
# Create parent directory
db_path.parent.mkdir(parents=True, exist_ok=True)
# Create engine and tables
engine = create_engine(
f"sqlite:///{db_path}", echo=False, connect_args={"check_same_thread": False}
)
# Reset globals so the engine/session factory rebuild against db_path
reset_connection()
# Build the engine (reuses the shared configuration) and create tables
engine = get_engine(db_path)
Base.metadata.create_all(bind=engine)
# Update globals
_engine = engine
_SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Seed predefined tags if the table is empty
_seed_tags_if_empty(engine)