1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
|
from functools import wraps
from urllib.parse import urlencode
from fastapi import FastAPI, Request
from sqlalchemy import func
from sqlmodel import select
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.middleware.sessions import SessionMiddleware
from starlette.responses import RedirectResponse
from starlette.templating import Jinja2Templates
import conf
from authentication.backend import SessionAuthBackend
from db.models import User
from db.session import get_session_context
management = FastAPI()
management.add_middleware(AuthenticationMiddleware, backend=SessionAuthBackend())
management.add_middleware(SessionMiddleware, secret_key=conf.SECRET_KEY)
templates = Jinja2Templates(directory=[
'templates',
'management/templates'
])
def requires_session(func):
"""Redirects to an absolute login URL instead of Starlette's `requires(redirect=...)`,
which resolves the redirect target via `request.url_for()` against whichever router is
bound into `request.scope` -- a lookup that only reaches `authentication`'s routes when
`management` happens to be mounted under a shared root app."""
@wraps(func)
async def wrapper(request: Request, *args, **kwargs):
if not {'authenticated', 'session'}.issubset(request.auth.scopes):
next_param = urlencode({'next': str(request.url)})
return RedirectResponse(f'/auth/login/?{next_param}', status_code=303)
return await func(request, *args, **kwargs)
return wrapper
@management.get('/')
async def index(request: Request):
with get_session_context() as session:
query = select(func.count(User.id))
user_count = session.exec(query)
if user_count.first() == 0:
return RedirectResponse('/auth/register/?next=/management/dashboard/', status_code=303)
return RedirectResponse('/management/dashboard/', status_code=303)
@management.get('/dashboard/')
@requires_session
async def dashboard(request: Request):
return templates.TemplateResponse(request, 'dashboard.html')
@management.get('/users/')
@requires_session
async def users(request: Request):
with get_session_context() as session:
query = select(User)
users = session.exec(query).all()
return templates.TemplateResponse(request, 'users.html', context={
'users': users
})
|
