Fastapi. Как использовать dependency внутри функции или создать свою функцию на основе dependency
Проблематика такова. Пишу приложение на fastapi. Дополнительно установлен fastapi-users. Сконфигурирован на хранение access_token в бд. Дополнительно определены зависимости current_user и current_superuser, используемые в большинстве эндпоинтов, как ограничитель доступа. БэкендБ стратегия, зависимости:
....
from typing import TYPE_CHECKING
from fastapi_users import FastAPIUsers
from fastapi_users.authentication import (
AuthenticationBackend,
BearerTransport,
)
from fastapi import Depends
from fastapi_users.authentication.strategy.db import AccessTokenDatabase, DatabaseStrategy
from sqlalchemy import Integer
from app1.core.settings import settings
from app1.core.auth.access_token import get_access_token_db
from app1.core.auth.user_manager import get_user_manager
if TYPE_CHECKING:
from app1.core.models import AccessToken
# Strategy
def get_database_strategy(
access_token_db: AccessTokenDatabase["AccessToken"] = Depends(get_access_token_db),
) -> DatabaseStrategy:
return DatabaseStrategy(
access_token_db,
lifetime_seconds=settings.access_token.ACCESS_TOKEN_LIFETIME,
)
# Backend
bearer_transport = BearerTransport(
tokenUrl=settings.auth.TRANSPORT_TOKEN_URL
)
authentication_backend = AuthenticationBackend(
name="access-token-db",
transport=bearer_transport,
get_strategy=get_database_strategy,
)
# FastAPI Users
fastapi_users = FastAPIUsers["User", Integer](
get_user_manager,
[authentication_backend],
)
current_user = fastapi_users.current_user(
active=True,
verified=True,
)
current_superuser = fastapi_users.current_user(
active=True,
verified=True,
superuser=True,
)
Далее использование в эндпоинтах.
@router.get(
"/{id}/",
dependencies=[Depends(current_user),],
status_code=status.HTTP_200_OK,
response_model=ProductRead,
)
async def get_one(
id: int,
session: AsyncSession = Depends(DBConfigurer.session_getter)
):
...
Но тут решил побаловаться с сессиями, но добавлять их не только для аутентифицированных пользователей, а для всех подряд.
# SESSION DATA ################################
class SessionData(BaseModel):
user: Annotated[Optional[Dict[str, Any]], None]
data: Annotated[Dict[str, Any], {}]
Сам эндпоинт примерно:
@router.post(
"/create_session",
)
async def create_session(
response: Response,
user: Any = Depends(current_user),
session: AsyncSession = Depends(DBConfigurer.session_getter),
):
user_dict: Dict[str, Any] = user.to_dict() if user else {}
user_session_uuid = uuid4()
data = fsc.SessionData(user=user_dict)
await fsc.backend.create(user_session_uuid, data)
fsc.cookie.attach_to_response(response, user_session_uuid)
user_email = user.email if user and hasattr(user, "email") else "Anonimous"
return f"created session for {user_email!r}"
Проблема в том, что current_user, если пользователь не авторизован, еще на уровне Depends выкидывает HTTPException со статусом 401. Подскажите, как бы использовать этот обычный dependency current_user от fastapi_users, чтобы не выкидывать ошибку в случае, если пользователь не авторизован, а возвращать None. То ли превратить как-то dependency в обычную функцию и обернуть try/except, то ли как-то модернизировать саму dependency. Все, что мог соорудить, это напрямую написать самому ручками и получить совсем не приемлемый костыль. Может кто подскажет новичку что ппроще. Заранее благодарен