77 lines
2.6 KiB
Python
77 lines
2.6 KiB
Python
"""
|
|
Database engine/session management.
|
|
|
|
Uses SQLAlchemy so the storage backend is a config change, not a code
|
|
change. Today DB_URL points at a local SQLite file. When you're ready
|
|
to move to the MariaDB LXC, set DB_URL in .env to something like:
|
|
|
|
mysql+pymysql://user:[email protected]:3306/order_manager
|
|
|
|
and install PyMySQL (pip install pymysql). Nothing else in the app
|
|
needs to change.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy import create_engine, inspect, text
|
|
from sqlalchemy.orm import sessionmaker, Session
|
|
|
|
from app import config
|
|
from app.models import Base, Order
|
|
|
|
_engine = None
|
|
_SessionLocal = None
|
|
|
|
|
|
def get_engine():
|
|
global _engine
|
|
if _engine is None:
|
|
db_url = config.get("DB_URL", config.DEFAULT_DB_URL)
|
|
connect_args = {}
|
|
if db_url.startswith("sqlite"):
|
|
# allow use across the QThread worker and the UI thread
|
|
connect_args = {"check_same_thread": False}
|
|
_engine = create_engine(db_url, connect_args=connect_args, future=True)
|
|
return _engine
|
|
|
|
|
|
def get_session_factory():
|
|
global _SessionLocal
|
|
if _SessionLocal is None:
|
|
_SessionLocal = sessionmaker(bind=get_engine(), future=True, expire_on_commit=False)
|
|
return _SessionLocal
|
|
|
|
|
|
def init_db() -> None:
|
|
"""Create tables that don't exist yet, and patch existing ones with
|
|
any columns added to the model since they were created. Safe to call
|
|
every startup - this is a local cache rebuilt from JIRA/ShipStation,
|
|
not a system of record, so a simple additive auto-migration is
|
|
enough; we don't need a full migration framework for it."""
|
|
engine = get_engine()
|
|
Base.metadata.create_all(engine)
|
|
_add_missing_columns(engine)
|
|
|
|
|
|
def _add_missing_columns(engine) -> None:
|
|
inspector = inspect(engine)
|
|
if "orders" not in inspector.get_table_names():
|
|
return # create_all() just made it fresh - nothing to patch
|
|
|
|
existing_columns = {col["name"] for col in inspector.get_columns("orders")}
|
|
missing = [c for c in Order.__table__.columns if c.name not in existing_columns]
|
|
if not missing:
|
|
return
|
|
|
|
with engine.begin() as conn:
|
|
for column in missing:
|
|
col_type = column.type.compile(dialect=engine.dialect)
|
|
# Always added as nullable, regardless of what the model says -
|
|
# existing rows get backfilled with real values the next time
|
|
# they're touched by an import, and this avoids ALTER TABLE
|
|
# failing on a NOT NULL column with no default for old rows.
|
|
conn.execute(text(f'ALTER TABLE orders ADD COLUMN "{column.name}" {col_type}'))
|
|
|
|
|
|
def get_session() -> Session:
|
|
return get_session_factory()()
|