pyr0ball
a90c3fc172
Wraps the Task 2 CRUD layer in a FastAPI app (schemas.py + main.py) with POST/GET/GET-by-id/PATCH routes for items, matching the exact paths and status codes the Discord bot's HTTP client will depend on. Also fixes a latent bug in db.get_engine: sqlite:///:memory: without a StaticPool gives each new session a fresh, empty database, which broke as soon as more than one session shared an engine (the API's per-request session pattern). Tasks 1-2 never hit this because their tests used a single session per engine.
28 lines
852 B
Python
28 lines
852 B
Python
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker, DeclarativeBase
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
def get_engine(db_url: str):
|
|
if db_url.startswith("sqlite"):
|
|
connect_args = {"check_same_thread": False}
|
|
# In-memory SQLite is per-connection; without a shared StaticPool,
|
|
# each new session would see a fresh, empty database.
|
|
if ":memory:" in db_url:
|
|
return create_engine(
|
|
db_url, connect_args=connect_args, poolclass=StaticPool
|
|
)
|
|
return create_engine(db_url, connect_args=connect_args)
|
|
return create_engine(db_url)
|
|
|
|
|
|
def make_session_factory(engine):
|
|
return sessionmaker(bind=engine, autoflush=False, autocommit=False)
|
|
|
|
|
|
def init_db(engine):
|
|
Base.metadata.create_all(bind=engine)
|