如何无需SSL证书文件，用asyncpg连接Azure Postgres数据库

asyncpg不支持在连接URL中添加sslmode=require这类psycopg2风格的参数，但可以通过SQLAlchemy create_async_engine的connect_args参数直接传递asyncpg的SSL配置，实现无证书的强制SSL连接。

具体实现 #

直接在创建异步引擎时，通过connect_args传入SSL配置：

方式1：简洁写法（启用默认SSL模式）

from sqlmodel import SQLModel
from sqlalchemy.ext.asyncio import create_async_engine

# 注意URL不要加sslmode参数
DATABASE_URL = "postgresql+asyncpg://user:password@your-azure-postgres-host:5432/dbname"

engine = create_async_engine(
    DATABASE_URL,
    connect_args={"ssl": True},  # 启用SSL，asyncpg会自动使用合适的加密模式
    echo=True  # 可选，用于调试SQL语句
)

# 初始化数据库（可选）
async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

方式2：明确指定SSL模式（更贴合Azure要求）

如果需要更精准的控制，可以明确指定sslmode为require：

engine = create_async_engine(
    DATABASE_URL,
    connect_args={"ssl": {"sslmode": "require"}},
    echo=True
)

在FastAPI中使用的完整示例 #

搭配FastAPI的数据库依赖，实现可复用的数据库会话：

from fastapi import FastAPI, Depends
from sqlmodel import Session, SQLModel
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

DATABASE_URL = "postgresql+asyncpg://user:password@your-azure-postgres-host:5432/dbname"

engine = create_async_engine(DATABASE_URL, connect_args={"ssl": True})

async def get_session() -> AsyncSession:
    async with AsyncSession(engine) as session:
        yield session

app = FastAPI()

@app.get("/")
async def root(session: AsyncSession = Depends(get_session)):
    # 此处可执行数据库操作
    return {"message": "已通过无证书SSL连接Azure Postgres"}

原理说明 #

Azure Postgres允许无证书的SSL连接，只要客户端发起SSL加密请求即可建立安全连接。asyncpg的ssl=True或ssl={"sslmode": "require"}配置会强制客户端与服务器建立SSL加密通道，无需本地存储根证书文件，刚好匹配Azure的这一要求。

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