blob: 29408e38e851abe8facdd79e720e97e9d10f80f6 (
plain)
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
|
from datetime import datetime, UTC
from typing import Optional
from sqlalchemy import update
from sqlmodel import select
from starlette.authentication import AuthenticationBackend, AuthCredentials, BaseUser, UnauthenticatedUser
from starlette.requests import HTTPConnection
from db.models import Session, User
from db.session import get_session_context
class AuthBackend(AuthenticationBackend):
async def authenticate(self, request: HTTPConnection) -> Optional[tuple[AuthCredentials, BaseUser]]:
if "id" not in request.session:
return None
query = (
select(Session)
.join(User)
.where(
Session.id == request.session['id']
)
)
with get_session_context() as db_session:
session: Session = db_session.exec(query).first()
if session is not None:
db_session.execute(
update(Session).where(Session.id == session.id)
)
db_session.commit()
return (
AuthCredentials(
[
"authenticated",
*[
connection.type for connection in session.user.app_connections
]
]
if session is not None
else []
),
session.user if session is not None else UnauthenticatedUser()
)
|
