Add FastAPI app skeleton: settings, db, models, security, auth

- settings.py: pydantic-settings config for the two mount points
  (MUSIC_DATA_DIR, ALEMBIC_CONFIG_DIR), OIDC client config, encryption key path
- db.py: SQLite engine with WAL mode, plain-SQL schema migration runner
  (schema/*.sql, tracked by schema_version -- not the alembic migration tool)
- models.py: SQLAlchemy ORM models matching schema/0001_init.sql
- security/crypto.py: Fernet encrypt/decrypt for the secrets table
- security/oidc.py + routers/auth.py: Pocket ID OIDC login/callback/logout
- security/deps.py: require_auth dependency (redirect-to-login on no session)
- main.py: app factory, lifespan (init_db + beets WAL enable), session
  middleware, minimal dashboard route

Verified boots end-to-end via TestClient: unauthenticated GET / redirects to
/auth/login, static files serve, and all 13 schema tables get created on
first run.
This commit is contained in:
andrew
2026-07-08 13:24:18 -06:00
parent 68cb007e4c
commit bd56c8153f
18 changed files with 595 additions and 0 deletions
+53
View File
@@ -0,0 +1,53 @@
import time
from fastapi import APIRouter, Request
from fastapi.responses import RedirectResponse
from sqlalchemy import select
from app.db import SessionLocal
from app.models import User
from app.security.oidc import oauth
router = APIRouter(prefix="/auth", tags=["auth"])
@router.get("/login")
async def login(request: Request):
redirect_uri = request.url_for("auth_callback")
return await oauth.pocketid.authorize_redirect(request, redirect_uri)
@router.get("/callback", name="auth_callback")
async def auth_callback(request: Request):
token = await oauth.pocketid.authorize_access_token(request)
userinfo = token.get("userinfo") or await oauth.pocketid.userinfo(token=token)
sub = userinfo["sub"]
email = userinfo.get("email")
name = userinfo.get("name")
now = time.time()
with SessionLocal() as db:
user = db.execute(select(User).where(User.oidc_sub == sub)).scalar_one_or_none()
if user is None:
user = User(oidc_sub=sub, email=email, name=name, last_login_at=now)
db.add(user)
else:
user.email = email
user.name = name
user.last_login_at = now
db.commit()
request.session["user"] = {"sub": sub, "email": email, "name": name}
return RedirectResponse(url="/")
@router.get("/logout")
async def logout(request: Request):
request.session.clear()
end_session_endpoint = None
metadata = await oauth.pocketid.load_server_metadata()
end_session_endpoint = metadata.get("end_session_endpoint")
if end_session_endpoint:
return RedirectResponse(url=end_session_endpoint)
return RedirectResponse(url="/")