FastAPI AsyncPG

FastAPI的AsyncPG集成

叙述

首先，非常抱歉我的英语很差。如果有人能推动一个PR来纠正我所有的英语错误，我会非常高兴。无论如何，我会尽力而为。

看看fastapi生态系统，好像每个人都试图将fastapi与orms集成，但根据我的经验，使用原始SQL非常高效。

如果你稍加思考，你的实际模型层就是数据库上的模式（你可以在它上面添加抽象），但最终结果是你的数据，这些是表、列和行。

此外，SQL，这是我学到的最好的东西之一，因为它是始终存在的。

另一方面，PostgreSQL非常健壮和稳固，数千个项目依赖于它，并使用它作为它们的存储层。AsyncPG是一个从头开始编写的疯狂快速PostgreSQL驱动程序。

FastAPI看起来是一个干净且开发人员高效的Web框架方法。它与OpenAPI的集成非常好，而且让开发人员迁移变得非常容易。

集成

fastapi_asyncpg试图以惯用的方式集成fastapi和asyncpg。当配置完成后，fastapi_asyncpg向fastapi路径函数公开了两个可注入的提供者，可以使用

• db.connection：它只是从池中挑选的一个原始连接，当路径函数结束时自动释放，这主要归功于fastapi周围的DI系统。

• db.transaction：相同，但将pathfunction封装在事务中，这几乎与Django中的atomic装饰器相同。另外，db.atomic已被别名。

from fastapi import FastAPI
from fastapi import Depends
from fastapi_asyncpg import configure_asyncpg

app = FastAPI()
# we need to pass the fastapi app to make use of lifespan asgi events
db = configure_asyncpg(app, "postgresql://postgres:postgres@localhost/db")

@db.on_init
async def initialization(conn):
    # you can run your db initialization code here
    await conn.execute("SELECT 1")


@app.get("/")
async def get_content(db=Depends(db.connection)):
    rows = await db.fetch("SELECT wathever FROM tablexxx")
    return [dict(r) for r in rows]

@app.post("/")
async def mutate_something_compled(db=Depends(db.atomic))
    await db.execute()
    await db.execute()
    # if something fails, everyting is rolleback, you know all or nothing

主工厂函数上还有一个名为initialization的可调用对象。它可以像flask一样用于初始化数据库上的任何需要。在asyncpg建立连接后、应用完全启动前，会调用initialization（一些项目将此用作较差的迁移运行器，如果部署多个应用实例，这不是最佳实践）。

测试

对于测试，我们使用pytest-docker-fixtures，它需要在宿主机或你使用的任何CI上安装docker（看起来与github actions配合使用时表现正常）。

它有效，为会话创建一个容器并将其作为pytest fixture暴露出来。使用真实数据库运行测试是一种好习惯，pytest-docker-fixtures使其变得非常容易。作为额外的奖励，所有fixture都在CI上运行。我们使用Jenkins与docker配合，但travis和github actions似乎也可以工作。

需要将fixture添加到pytest插件conftest.py文件中。

在conftest.py上

pytest_plugins = [
    "pytest_docker_fixtures",
]

有了这些，我们就可以yield一个pg fixture

from pytest_docker_fixtures import images

# image params can be configured from here
images.configure(
    "postgresql", "postgres", "11.1", env={"POSTGRES_DB": "test_db"}
)

# and then on our test we have a pg container running
# ready to recreate our db
async def test_pg(pg):
    host, port = pg
    dsn = f"postgresql://postgres@{host}:{port}/test_db"
    await asyncpg.Connect(dsn=dsn)
    # let's go

有了这些，我们就可以创建自己的pytest.fixture来修补应用dsn，使其与自定义创建的容器兼容。

from .app import app, db
from async_asgi_testclient import TestClient

import pytest

pytestmark = pytest.mark.asyncio

@pytest.fixture
async def asgi_app(pg)
    host, port = pg
    dsn = f"postgresql://postgres@{host}:{port}/test_db"
    # here we patch the dsn for the db
    # con_opts: are also accessible
    db.dsn = dsn
    yield app, db

async def test_something(asgi_app):
    app, db = asgi_app
    async with db.pool.acquire() as db:
        # setup your test state

    # this context manager handlers lifespan events
    async with TestClient(app) as client:
        res = await client.request("/")
```

Anyway if the application will grow, to multiples subpackages,
and apps, we trend to build the main app as a factory, that
creates it, something like:

```python
from fastapi_asyncpg import configure_asyncpg
from apppackage import settings

import venusian

def make_asgi_app(settings):
    app = FastAPI()
    db = configure_asyncpg(settings.DSN)

    scanner = venusian.Scanner(app=app)
    venusian.scan(theapp)
    return app

然后在fixture中，我们只需要从我们的函数中实例化一个app

from .factory import make_asgi_app
from async_asgi_testclient import TestClient

import pytest

pytestmark = pytest.mark.asyncio

@pytest.fixture
async def asgi_app(pg)
    host, port = pg
    dsn = f"postgresql://postgres@{host}:{port}/test_db"
    app = make_asgi_app({"dsn": dsn})
    # ther's a pointer on the pool into app.state
    yield app

async def test_something(asgi_app):
    app = asgi_app
    pool = app.state.pool
    async with db.pool.acquire() as db:
        # setup your test state

    # this context manager handlers lifespan events
    async with TestClient(app) as client:
        res = await client.request("/")

在tests中也公开并使用了另一种方法，该方法在测试结束时向测试公开单个连接并回滚更改。我们在一个大型项目（每个模式有500张表，多个模式）中使用这种方法，并且它似乎可以加快测试创建的速度。这种方法是Databases使用的。请随意查看测试，看看是否更适合。

附加信息

有一些我日常使用的辅助函数与asyncpg配合使用，帮助我加快一些SQL操作，它们都在sql.py中，并且大多是自我文档化的。它们在测试中使用。

作者

fastapi_asyncpg由Jordi collell <jordic@gmail.com>编写。