使用psycopg3异步Pool+SQLAlchemy/SQLModel/FastAPI遇连接串错误求助

错误原因 #

你遇到的missing "=" after "postgresql+psycopg://postgres:postgres@127.0.0.1:5432/foo" in connection info string错误，本质是psycopg3在解析连接字符串时，发现格式不符合语法要求。最常见的触发场景是：

• 错误地将连接池参数（如pool_size=5）直接附加到连接串末尾，却没有用?作为前缀，导致解析器将这些参数识别为连接串的一部分，找不到key=value格式的分隔符；
• 连接串中存在未转义的特殊字符（不过你的案例中密码为postgres，暂不考虑这种可能）。

另外需要明确：SQLAlchemy 2.0+对psycopg3的支持使用postgresql+psycopg://方言格式是正确的，这个部分没有问题。

正确的异步连接池搭建方案 #

连接池的配置不需要嵌入连接字符串，而是通过SQLAlchemy的异步引擎参数来设置。以下是完整的FastAPI+SQLModel+psycopg3异步连接池实现：

1. 确认兼容依赖版本 #

确保你的requirements.txt包含以下版本（或更高的兼容版本）：

psycopg[binary]>=3.1.0
sqlalchemy>=2.0.0
sqlmodel>=0.0.14
fastapi>=0.100.0
uvicorn>=0.23.2

2. 核心代码实现 #

from fastapi import FastAPI
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import AsyncAdaptedQueuePool

# 正确的连接字符串格式
DATABASE_URL = "postgresql+psycopg://postgres:postgres@127.0.0.1:5432/foo"

# 初始化带连接池的异步引擎
engine = create_async_engine(
    DATABASE_URL,
    echo=True,  # 可选：开启SQL日志，方便调试
    poolclass=AsyncAdaptedQueuePool,  # SQLAlchemy官方异步连接池实现
    pool_size=5,  # 连接池常驻连接数
    max_overflow=10,  # 超出常驻数的临时连接上限
    pool_recycle=300,  # 闲置连接自动回收时间（秒），避免数据库主动断开
    pool_pre_ping=True  # 获取连接前自动检查可用性，避免使用失效连接
)

# 异步初始化数据库表
async def init_db():
    async with engine.begin() as conn:
        # await conn.run_sync(SQLModel.metadata.drop_all)  # 可选：删除现有表
        await conn.run_sync(SQLModel.metadata.create_all)

# 依赖注入：获取异步数据库会话
async def get_session() -> AsyncSession:
    async with AsyncSession(engine) as session:
        yield session

# FastAPI应用实例
app = FastAPI(title="Async PostgreSQL with Connection Pool")

# 启动时初始化数据库
@app.on_event("startup")
async def startup():
    await init_db()

# 测试路由
@app.get("/")
async def health_check():
    return {"status": "ok", "message": "Async connection pool setup completed"}

3. 关键注意事项 #

• 必须使用create_async_engine而非同步的create_engine，才能适配FastAPI的异步请求处理；
• 所有连接池参数通过create_async_engine的关键字参数传入，不要尝试拼接到连接字符串中；
• 如果数据库密码包含特殊字符（如@、#、&），需要用URL编码替换（例如%40代替@），否则会导致连接串解析失败；
• SQLModel的异步会话完全依赖SQLAlchemy的异步引擎，确保两者版本兼容。

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