A R R R Associates
53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
from __future__ import annotations
|
|
from datetime import datetime, timedelta, timezone
|
|
from fastapi import Request, Depends
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db.deps import get_common_db
|
|
from app.modules.core.iam.models import User
|
|
from app.modules.core.tenancy.models import Branch
|
|
from app.modules.core.tenancy.settings_models import BranchSettings
|
|
|
|
SESSION_USER_ID_KEY = "user_id"
|
|
SESSION_LOGIN_AT_KEY = "login_at"
|
|
|
|
def _now_utc() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
def get_current_user(request: Request, db: Session = Depends(get_common_db)) -> User | None:
|
|
user_id = request.session.get(SESSION_USER_ID_KEY)
|
|
if not user_id:
|
|
return None
|
|
|
|
user = db.execute(select(User).where(User.id == int(user_id))).scalar_one_or_none()
|
|
if not user or not user.is_active or not getattr(user, "allow_login", True) or getattr(user, "is_locked", False) or getattr(user, "deleted_at", None) is not None:
|
|
return None
|
|
|
|
# Enforce session duration from BranchSettings
|
|
login_at = request.session.get(SESSION_LOGIN_AT_KEY)
|
|
if login_at:
|
|
try:
|
|
login_at_dt = datetime.fromisoformat(login_at)
|
|
except Exception:
|
|
login_at_dt = None
|
|
else:
|
|
login_at_dt = None
|
|
|
|
bs = db.execute(select(BranchSettings).where(BranchSettings.branch_id == user.branch_id)).scalar_one_or_none()
|
|
max_minutes = bs.session_duration_minutes if bs else 480
|
|
|
|
if login_at_dt:
|
|
if _now_utc() - login_at_dt > timedelta(minutes=max_minutes):
|
|
# expire session
|
|
request.session.pop(SESSION_USER_ID_KEY, None)
|
|
request.session.pop(SESSION_LOGIN_AT_KEY, None)
|
|
return None
|
|
|
|
return user
|
|
|
|
def require_login(user: User | None = Depends(get_current_user)) -> User:
|
|
if not user:
|
|
raise PermissionError("Not authenticated")
|
|
return user
|