57 lines
1.2 KiB
Python
57 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import Depends, Request, Response
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.app.db import get_db
|
|
from backend.app import models
|
|
|
|
DEVICE_COOKIE_NAME = "opc_device_token"
|
|
|
|
|
|
def _issue_new_device_user(db: Session, response: Response) -> models.User:
|
|
token = secrets.token_hex(32)
|
|
user = models.User(
|
|
device_token=token,
|
|
created_at=datetime.now(timezone.utc),
|
|
last_seen_at=datetime.now(timezone.utc),
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
response.set_cookie(
|
|
key=DEVICE_COOKIE_NAME,
|
|
value=token,
|
|
httponly=True,
|
|
secure=False,
|
|
samesite="Lax",
|
|
max_age=60 * 60 * 24 * 365,
|
|
)
|
|
return user
|
|
|
|
|
|
def get_current_user(
|
|
request: Request,
|
|
response: Response,
|
|
db: Session = Depends(get_db),
|
|
) -> models.User:
|
|
token = request.cookies.get(DEVICE_COOKIE_NAME)
|
|
if not token:
|
|
return _issue_new_device_user(db, response)
|
|
|
|
user = (
|
|
db.query(models.User)
|
|
.filter(models.User.device_token == token)
|
|
.first()
|
|
)
|
|
if not user:
|
|
return _issue_new_device_user(db, response)
|
|
|
|
user.last_seen_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
return user
|
|
|