78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
import os
|
|
from fastapi import FastAPI, Depends, Request
|
|
from fastapi.responses import RedirectResponse, HTMLResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from contextlib import asynccontextmanager
|
|
from sqlalchemy import select
|
|
|
|
from pathlib import Path
|
|
from .database import engine, async_session, Base
|
|
from .routers import auth, admin, cards
|
|
from .auth import get_current_user, hash_password
|
|
from .models import User
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_app: FastAPI):
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
# Create default admin user if not exists
|
|
async with async_session() as session:
|
|
result = await session.execute(select(User).where(User.username == "admin"))
|
|
if not result.scalar_one_or_none():
|
|
session.add(User(
|
|
username="admin",
|
|
password_hash=hash_password("admin123"),
|
|
role="admin",
|
|
))
|
|
await session.commit()
|
|
|
|
# Ensure image upload directory exists
|
|
Path("app/static/images").mkdir(parents=True, exist_ok=True)
|
|
yield
|
|
|
|
|
|
app = FastAPI(lifespan=lifespan)
|
|
|
|
app.mount("/static", StaticFiles(directory="app/static"), name="static")
|
|
app.include_router(cards.router)
|
|
app.include_router(auth.router)
|
|
app.include_router(admin.router)
|
|
|
|
templates = Jinja2Templates(directory="app/templates")
|
|
|
|
|
|
@app.middleware("http")
|
|
async def auth_middleware(request: Request, call_next):
|
|
"""Attach user to request for templates."""
|
|
token = request.cookies.get("token")
|
|
request.state.user = None
|
|
if token:
|
|
from .auth import decode_token
|
|
from sqlalchemy import select
|
|
from .database import async_session
|
|
payload = decode_token(token)
|
|
if payload:
|
|
async with async_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.username == payload["sub"])
|
|
)
|
|
request.state.user = result.scalar_one_or_none()
|
|
response = await call_next(request)
|
|
return response
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def root(request: Request):
|
|
if not request.state.user:
|
|
return RedirectResponse(url="/login")
|
|
return await cards.index(request)
|
|
|
|
|
|
@app.get("/logout")
|
|
async def logout_redirect():
|
|
resp = RedirectResponse(url="/login", status_code=302)
|
|
resp.delete_cookie("token")
|
|
return resp |