FastApi 纯 SQLAlchemy 异步试用

今天整理邮件看到 SQLAlchemy 发布了 1.4 版本,其中针对异步的相关功能完全开发了出来,包括 core 和 ORM 的异步能力。虽然文档中说异步功能还处于 alpha 阶段,API随时可能变动,但并不影响我们来尝鲜测试。

FastApi 组件的 ORM 模块中,使用了两个组件 SQLAlchemy 和 databases(encode 开源的异步链接库)。这下我们可以把 databases 去掉了。

之前我分享了一个 FastApi 的使用demo代码(https://github.com/pylixm/docker-fastapi-demo),我们在它上面修改。

去除异步链接库databases,使用纯SQLAlchemy构建异步回话 session。

# https://github.com/pylixm/docker-fastapi-demo/blob/e6719e0de878fdfb7cd1dc655bb22614876e059a/core/db.py#L22
# async
# mysql+aiomysql://user:password@host:port/dbname
async_engine = create_async_engine(
    SQLALCHEMY_ASYNC_DATABASE_URL, echo=True,
)

async_session = sessionmaker(
        async_engine, expire_on_commit=False, class_=AsyncSession
    )

为实现 session 回话的重复使用,我们把它挂到app 实例上。

# https://github.com/pylixm/docker-fastapi-demo/blob/e6719e0de878fdfb7cd1dc655bb22614876e059a/main.py#L28

    @app.on_event("startup")
    async def startup():
        app.state.se = async_session()

View 视图中直接调用 session 回话来操作数据库。

# https://github.com/pylixm/docker-fastapi-demo/blob/e6719e0de878fdfb7cd1dc655bb22614876e059a/service/views.py#L11

@router.get('/')
async def index(request: Request):
    print(posts.select())
    # post_list = await database.fetch_all(query=posts.select())
    post_list = await request.app.state.se.execute(posts.select())
    return templates.TemplateResponse(
        'index.html',
        {
            'request': request,
            'post_list': post_list
        }
    )

完整代码,可见GitHub:https://github.com/pylixm/docker-fastapi-demo/tree/sqlalchemy_async

这样,我们就把 databases 组件去除了,ORM 层只用 SQLAlchemy 来实现。我们在后期的开发和维护中,只关心 SQLAlchemy 技术栈便可以了。