d092c19b41
- app/models/: Printer, Filament, PrintJob, Todo, PricingConfig with full SQLAlchemy 2.x mapped columns and relationships - app/routers/: list + get-by-id endpoints for all five resources - app/db.py: SQLAlchemy engine + SessionLocal + get_db dependency - alembic/: env.py reads MYSQL_* env vars; 0001_initial_schema mirrors db.sql - alembic.ini: script_location configured, URL injected at runtime Since db.sql already created the tables, stamp the DB before running future migrations: alembic stamp head Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
34 lines
765 B
Python
34 lines
765 B
Python
import os
|
|
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
|
|
def _db_url() -> str:
|
|
return (
|
|
f"mysql+pymysql://{os.environ['MYSQL_USER']}:{os.environ['MYSQL_PASSWORD']}"
|
|
f"@{os.environ['MYSQL_HOST']}:{os.environ.get('MYSQL_PORT', 3306)}"
|
|
f"/{os.environ['MYSQL_DATABASE']}"
|
|
)
|
|
|
|
|
|
engine = create_engine(_db_url(), pool_pre_ping=True)
|
|
SessionLocal = sessionmaker(bind=engine)
|
|
|
|
|
|
def get_db():
|
|
db: Session = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def check_db() -> tuple[bool, str]:
|
|
try:
|
|
with engine.connect() as conn:
|
|
conn.execute(text("SELECT 1"))
|
|
return True, "ok"
|
|
except Exception as e:
|
|
return False, str(e)
|