FastAPI + Strawberry GraphQL: проблема с установкой файлов cookie в ответе (объект ответа не имеет атрибута) ⇐ Python

[Anonymous]

Я использую FastAPI с Strawberry GraphQL и пытаюсь реализовать мутацию, которая устанавливает в ответе access_token иrefresh_token как файлы cookie. Однако я столкнулся со следующей ошибкой:
Объект «Ответ» не имеет атрибута «cookie»

Код: Выделить всё

from typing import List, Optional, Any
from contextlib import asynccontextmanager
from redis import Redis
import strawberry
from strawberry.fastapi import BaseContext, GraphQLRouter
from fastapi import Request, Response, Depends
from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession
from src.infrastructure.database.connection import async_session
from src.infrastructure.repositories.user_repository import SQLAlchemyUserRepository
from src.infrastructure.repositories.otp_repository import SQLAlchemyOTPRepository
from src.application.usecases.IEmailUseCase import EmailServiceUseCase
from src.core.config import settings
from src.application.usecases.IUserRegister import UserRegistrationServiceUseCase
from src.infrastructure.config.reddis_config import RedisConfig
from src.__lib.UserRole import UserRole
import logging

# Setting up logger
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

# Custom error for registration
class RegistrationError(Exception):
pass

# Context manager for database session and Redis client
@asynccontextmanager
async def get_session() -> AsyncSession:
async with async_session() as session:
yield session

@asynccontextmanager
async def get_redis_client() -> Redis:
redis_config = RedisConfig()
redis_client = redis_config.get_client()
yield redis_client
redis_client.close()

# Define GraphQL types
@strawberry.type
class RegistrationStatus:
status: str
message: str
email: Optional[str] = None
created_at: Optional[str] = None
attempts_remaining: Optional[int] = None
expires_in: Optional[int] = None

@strawberry.type
class RegistrationResponse:
message: str
status: RegistrationStatus

@strawberry.type
class User:
id: int
email: str
is_active: bool
is_verified: bool
bio: Optional[str] = None
profile_image_url: Optional[str] = None
role: str

@strawberry.type
class VerificationResponse:
user: User
access_token: str
refresh_token: str

@strawberry.input
class UserCreateInput:
email: str
password: str
bio: Optional[str] = None
profile_image_url: Optional[str] = None

# Custom context with session and Redis client
class CustomContext(BaseContext):
def __init__(self, request: Request):
super().__init__()
self.request: Request = request
self.session: Optional[AsyncSession] = None
self.redis_client: Optional[Redis] = None
self.response: Response = Response()

async def __aenter__(self):
async with get_session() as session:
self.session = session
async with get_redis_client() as redis_client:
self.redis_client = redis_client
return self

async def __aexit__(self, exc_type, exc_val, exc_tb):
self.session = None
self.redis_client = None

# GraphQL Query and Mutation definitions
@strawberry.type
class Query:
@strawberry.field
async def users(self, info, skip: int = 0, limit: int = 100) -> List[User]:
async with get_session() as session:
repository = SQLAlchemyUserRepository(session)
users = await repository.list_users(skip=skip, limit=limit)
return [
User(
id=user.id,
email=str(user.email),
is_active=user.is_active,
bio=user.bio,
profile_image_url=user.profileImageURL,
role=user.role.value,
)
for user in users
]

@strawberry.type
class Mutation:
@strawberry.mutation
async def verify_registration(self, info, email: str, otp: str) ->  VerificationResponse:
context: CustomContext = info.context

try:
async with get_redis_client() as redis_client, get_session() as session:
registration_service = UserRegistrationServiceUseCase(
user_repository=SQLAlchemyUserRepository(session),
otp_repository=SQLAlchemyOTPRepository(session),
email_service=EmailServiceUseCase(),
redis_client=redis_client,
)

logger.info(f"Verifying OTP for {email}")

# Verify OTP and retrieve tokens
user, access_token, refresh_token = await registration_service.verify_otp(email, otp)

logger.info(f"OTP verified for {email}, issuing access token")

# Set cookies in the response
context.response.set_cookie(
key="access_token",
value=f"Bearer {access_token}",
httponly=True,
max_age=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,  # Convert minutes to seconds
)
context.response.set_cookie(
key="refresh_token",
value=f"Bearer {refresh_token}",
httponly=True,
max_age=settings.REFRESH_TOKEN_EXPIRE_DAYS * 24 * 60 * 60,  # Days to seconds
)

# Debugging cookie issue
logger.debug(f"Cookies set in response: {context.response.headers.get('Set-Cookie')}")

return VerificationResponse(
user=User(
id=user.id,
email=str(user.email),
is_active=user.is_active,
bio=user.bio,
profile_image_url=user.profileImageURL,
role=user.role.value,
is_verified=user.is_verified,
),
access_token=access_token,
refresh_token=refresh_token,
)
except RegistrationError as e:
logger.error(f"OTP verification failed for {email}: {str(e)}")
raise RegistrationError(f"OTP verification failed: {str(e)}")

# Schema and context setup
schema = strawberry.Schema(query=Query, mutation=Mutation)

async def get_context_with_response(request: Request, response: Response) -> CustomContext:
context = CustomContext(request)
context.response = response
await context.__aenter__()
return context

graphql_app = GraphQLRouter(schema, context_getter=get_context_with_response)

в ответ я получаю access_token
{
"data": null,
"errors": [{
"message": "Объект 'Ответ' не имеет атрибута 'cookies'",
"locations": [
{
"line": 2,"столбец": 3
}
],
"путь": [
"verifyRegistration"
]
}
]


Подробнее здесь: https://stackoverflow.com/questions/792 ... nse-object