74 lines
2.3 KiB
Python
74 lines
2.3 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 .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()
|
||
|
|
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
|