50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
from typing import Annotated
|
||
|
|
from uuid import UUID
|
||
|
|
|
||
|
|
from fastapi import Depends, Header
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
|
|
||
|
|
from core.auth.models import CurrentUser
|
||
|
|
from core.db import get_db
|
||
|
|
from services.base.supabase import supabase_service
|
||
|
|
from v1.feedback.repository import FeedbackRepository
|
||
|
|
from v1.feedback.service import FeedbackService
|
||
|
|
|
||
|
|
|
||
|
|
async def get_optional_user(
|
||
|
|
authorization: str | None = Header(default=None),
|
||
|
|
) -> CurrentUser | None:
|
||
|
|
if not authorization:
|
||
|
|
return None
|
||
|
|
|
||
|
|
scheme, _, token = authorization.partition(" ")
|
||
|
|
if scheme.lower() != "bearer" or not token:
|
||
|
|
return None
|
||
|
|
|
||
|
|
try:
|
||
|
|
client = supabase_service.get_client()
|
||
|
|
response = await asyncio.to_thread(client.auth.get_user, token)
|
||
|
|
user = getattr(response, "user", None)
|
||
|
|
user_id = getattr(user, "id", None)
|
||
|
|
if not isinstance(user_id, str) or not user_id:
|
||
|
|
return None
|
||
|
|
return CurrentUser(
|
||
|
|
id=UUID(user_id),
|
||
|
|
email=getattr(user, "email", None),
|
||
|
|
role=getattr(user, "role", None),
|
||
|
|
)
|
||
|
|
except Exception:
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def get_feedback_service(
|
||
|
|
session: Annotated[AsyncSession, Depends(get_db)],
|
||
|
|
) -> FeedbackService:
|
||
|
|
return FeedbackService(
|
||
|
|
repository=FeedbackRepository(session=session),
|
||
|
|
storage=supabase_service,
|
||
|
|
)
|