基于FastAPI+Strawberry的Cloud SQL异步会话池配置问题

错误直接原因 #

你遇到的'Connection' object is not callable错误，本质是代码里误将AsyncSession实例当作函数调用了——比如写成session()而非正确的await session.get(...)/await session.execute(...)这类会话方法调用。结合Cloud SQL连接场景，先从正确的异步会话池配置、请求级会话实现入手，再修正查询代码。

1. 正确配置Cloud SQL异步连接引擎 #

使用asyncpg驱动适配PostgreSQL异步场景，Cloud SQL推荐用Unix套接字连接（比TCP更稳定），同时配置合理的会话池参数避免连接超时或耗尽：

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

# Cloud SQL连接字符串格式（替换占位符为你的实际信息）
DATABASE_URL = "postgresql+asyncpg://<DB_USER>:<DB_PASSWORD>@/<DB_NAME>?host=/cloudsql/<PROJECT_ID>:<REGION>:<INSTANCE_NAME>"

# 配置异步引擎与会话池
engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,          # 常驻连接数
    max_overflow=20,       # 临时扩容连接数
    pool_recycle=300,      # 自动回收闲置连接（Cloud SQL默认超时300秒）
    pool_pre_ping=True,    # 预检测失效连接
    echo=False             # 生产环境关闭SQL日志
)

# 创建AsyncSession工厂
AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False  # 提交后不自动失效实例，适配GraphQL场景
)

2. 实现请求级会话依赖（FastAPI+Strawberry通用） #

通过FastAPI的依赖注入机制，为每个请求生成独立会话，请求结束自动关闭：

from fastapi import Depends

async def get_db_session() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

FastAPI接口中使用 #

from fastapi import FastAPI
from your_models import User

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int, session: AsyncSession = Depends(get_db_session)):
    # 正确调用方式：使用会话方法，而非直接调用session()
    user = await session.get(User, user_id)
    return {"id": user.id, "name": user.name} if user else {"error": "User not found"}

Strawberry GraphQL中使用 #

通过上下文注入会话，确保GraphQL查询也能获取请求级会话：

import strawberry
from strawberry.fastapi import GraphQLRouter
from your_models import User as DBUser
from your_graphql_types import User as GraphQLUser

# 定义上下文生成器
async def get_context(session: AsyncSession = Depends(get_db_session)):
    return {"db_session": session}

@strawberry.type
class Query:
    @strawberry.field
    async def user(self, user_id: int, info: strawberry.Info) -> GraphQLUser | None:
        session = info.context["db_session"]
        # 避免写成session()，调用正确的会话方法
        db_user = await session.get(DBUser, user_id)
        return GraphQLUser(id=db_user.id, name=db_user.name) if db_user else None

schema = strawberry.Schema(query=Query)
graphql_app = GraphQLRouter(schema, context_getter=get_context)

app.include_router(graphql_app, prefix="/graphql")

3. Cloud SQL额外检查项 #

• 本地测试时必须启动Cloud SQL Proxy：cloud-sql-proxy <PROJECT_ID>:<REGION>:<INSTANCE_NAME>
• 确认Cloud SQL实例的IP白名单/VPC peering配置允许当前环境连接
• 检查数据库用户权限、密码是否正确，实例的最大连接数是否大于会话池配置的pool_size + max_overflow

内容的提问来源于stack exchange，提问作者Sebastian