I am trying to create ETags with my subrahp.
I want to implement logic in all my subgraph so we check ETag and dont fetch if nothing has changed. This is good both for server, and client. But ut seems like Apollo Router is stripping my ETag, bevause when i curl it directly to the subgraph i get the ETag, but thru Apollo - no. Is it anything wrong in the schema? Or is it Apollo that is bad configured.
# schema.py
import logging
import strawberry
from strawberry.types import Info
from typing import Optional, List
from aiodataloader import DataLoader
import asyncpg
from strawberry.federation.schema_directives import Key, External
from datetime import datetime
import hashlib # For strong ETag generation
from strawberry import SchemaDirectiveVisitor
from graphql.language.ast import FieldNode
from fastapi import HTTPException
# Configure logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
def generate_strong_etag(data: str) -> str:
"""Generates a strong ETag using SHA-256."""
return hashlib.sha256(data.encode('utf-8')).hexdigest()
@strawberry.directive(locations=[strawberry.directive.DirectiveLocation.FIELD_DEFINITION])
class ETag:
"""
Strawberry directive to add ETag support and caching.
"""
pass
class ETagVisitor(SchemaDirectiveVisitor):
async def resolve(self, next_, source, info: Info, args, kwargs):
"""
Middleware-like function to handle ETag logic.
"""
# 1. Get the X-User-Id header.
user_id_header = info.context["request"].state.user_id if "request" in info.context and hasattr(info.context["request"].state, 'user_id') else None
if user_id_header is None:
# No user ID, proceed without ETag (or raise an error, depending on requirements).
logger.warning("No X-User-Id header found. ETag cannot be generated.")
return await next_(source, info, *args, **kwargs)
try:
user_id = int(user_id_header) # Validate
except ValueError:
logger.error(f"Invalid X-User-Id header: {user_id_header}")
raise strawberry.exceptions.StrawberryGraphQLError("Invalid user ID provided.")
# 2. Generate the ETag.
etag_data = f"{user_id}:{info.field_name}" # Include field name!
etag = generate_strong_etag(etag_data)
logger.debug(f"Generated ETag: {etag} for user: {user_id}, field: {info.field_name}")
# 3. Check the "If-None-Match" header.
if_none_match = info.context["request"].headers.get("If-None-Match")
if if_none_match == etag:
# 304 Not Modified. Return *nothing*. The schema will handle this correctly.
logger.debug(f"ETag matched If-None-Match. Returning 304 Not Modified.")
raise HTTPException(status_code=304) # Use FastAPI's HTTPException
# 4. Call the resolver and get the result.
result = await next_(source, info, *args, **kwargs)
# 5. Store ETag for middleware.
if result is not None:
info.context["request"].state.set_etag = etag # Use request.state
return result
def visit_field_definition(self, field, object_type):
original_resolver = field.resolve
async def resolve_with_etag(source, info: Info, *args, **kwargs):
return await self.resolve(original_resolver, source, info, args, kwargs)
field.resolve = resolve_with_etag
return field
@strawberry.federation.type(keys=["userId"], extend=True)
class UserInfo:
userId: int = strawberry.federation.field(external=True)
last_updated: Optional[datetime] = None
@classmethod
async def resolve_reference(cls, info: Info, userId: int):
db_pool = info.context["db"]
async with db_pool.acquire() as conn:
query = "SELECT userid, last_updated FROM userinfo WHERE userid = $1"
row = await conn.fetchrow(query, userId)
if row:
return UserInfo(userId=row["userid"], last_updated=row["last_updated"])
else:
return None
class UserByIdLoader(DataLoader):
def __init__(self, db_pool: asyncpg.Pool):
super().__init__(batch_load_fn=self.batch_load_fn)
self.db_pool = db_pool
async def batch_load_fn(self, userIds: List[int]):
async with self.db_pool.acquire() as conn:
query = "SELECT userid FROM userinfo WHERE userid = ANY($1::int[])"
rows = await conn.fetch(query, userIds)
user_map = {row["userid"]: row["userid"] for row in rows}
return [user_map.get(userId) for userId in userIds]
@strawberry.federation.type(keys=["userId"], extend=True)
class User:
userId: int = strawberry.federation.field(external=True)
@classmethod
async def resolve_reference(cls, info: Info, userId: int) -> Optional["UserInfo"]:
return await UserInfo.resolve_reference(info, userId)
@strawberry.type
class Query:
@strawberry.field(directives=[ETag()])
async def get_user_by_header(self, info: Info) -> Optional[UserInfo]:
db_pool = info.context["db"]
user_id_header = info.context["request"].state.user_id
if user_id_header is None:
return None
try:
user_id = int(user_id_header)
except ValueError:
logger.error(f"Invalid X-User-Id header: {user_id_header}")
raise strawberry.exceptions.StrawberryGraphQLError("Invalid User ID.")
async with db_pool.acquire() as conn:
query = "SELECT userid, last_updated FROM userinfo WHERE userid = $1" # Example: Fetch last_updated
row = await conn.fetchrow(query, user_id)
if not row:
return None
return UserInfo(userId=row["userid"], last_updated=row["last_updated"])
schema = strawberry.federation.Schema(
query=Query,
types=[UserInfo, User],
enable_federation_2=True,
directives=[ETag()]
)