src/api/dependencies.py

from typing import Annotated

from fastapi import Depends, Header, Path
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session

from src.api.errors import error_response
from src.config import get_settings
from src.models.auth import OrganizationRole
from src.services.auth import AuthService, AuthServiceError
from src.services.db import User, get_session
from src.services.organizations import OrganizationService

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login", auto_error=False)

DbSessionDep = Annotated[Session, Depends(get_session)]
AccessTokenDep = Annotated[str | None, Depends(oauth2_scheme)]


def service_error_to_response(exc: AuthServiceError):
    return error_response(status_code=exc.status_code, code=exc.code, message=exc.message)


def get_optional_current_user(db: DbSessionDep, token: AccessTokenDep) -> User | None:
    if not token:
        return None
    try:
        return AuthService(db, get_settings()).authenticate_access_token(token)
    except AuthServiceError:
        return None


def require_current_user(db: DbSessionDep, token: AccessTokenDep) -> User:
    if not token:
        raise AuthServiceError(401, "unauthorized", "Authentication required.")
    return AuthService(db, get_settings()).authenticate_access_token(token)


CurrentUserDep = Annotated[User, Depends(require_current_user)]
OptionalCurrentUserDep = Annotated[User | None, Depends(get_optional_current_user)]


def require_organization_membership(
    org_id: Annotated[str, Path()],
    user: CurrentUserDep,
    db: DbSessionDep,
):
    return OrganizationService(db).require_membership(user.id, org_id)


def require_organization_manager(
    org_id: Annotated[str, Path()],
    user: CurrentUserDep,
    db: DbSessionDep,
):
    return OrganizationService(db).require_manager(user.id, org_id)


OrganizationMembershipDep = Annotated[object, Depends(require_organization_membership)]
OrganizationManagerDep = Annotated[object, Depends(require_organization_manager)]
GuestSessionHeaderDep = Annotated[str | None, Header(alias="X-Guest-Session-Id")]


def role_allows_organization_management(role: OrganizationRole | str) -> bool:
    return str(role) == OrganizationRole.OWNER.value