ilia
31c656d66f
CI / security-scan (pull_request) Failing after 28s
CI / lint-and-test (pull_request) Failing after 31s
CI / dependency-scan (pull_request) Successful in 22s
CI / docker-build-test (pull_request) Failing after 34s
CI / workflow-summary (pull_request) Successful in 4s
Switch HouseWatcherClient to public STOCK Act JSON (legacy housestockwatcher.com is down), add httpx/scikit-learn to pyproject, wire SMTP settings through Settings, fix get_session() as a context manager, and use savepoints in TradeLoader so one bad row does not roll back the batch.
43 lines
1.0 KiB
Python
43 lines
1.0 KiB
Python
"""
|
|
Database layer: engine, session factory, and base model.
|
|
"""
|
|
|
|
from collections.abc import Generator
|
|
from contextlib import contextmanager
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
|
|
|
|
from pote.config import settings
|
|
|
|
# Create engine
|
|
engine = create_engine(
|
|
settings.database_url,
|
|
echo=settings.log_level == "DEBUG",
|
|
connect_args={"check_same_thread": False} if "sqlite" in settings.database_url else {},
|
|
)
|
|
|
|
# Session factory
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
"""Base class for all models."""
|
|
|
|
pass
|
|
|
|
|
|
@contextmanager
|
|
def get_session() -> Generator[Session, None, None]:
|
|
"""Get a database session (context manager or FastAPI-style dependency)."""
|
|
session = SessionLocal()
|
|
try:
|
|
yield session
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def init_db() -> None:
|
|
"""Create all tables. Use Alembic migrations in production."""
|
|
Base.metadata.create_all(bind=engine)
|