库备份工具DBTwin、SQL开发工具等

当使用两个不同的异步会话来测试FastAPI应用程序与数据库的连接时可能会出现以下错误

此外在测试中我们通常将会话包装在一个准备数据库进行测试的fixture中测试完成后所有内容都会被清理。

Database

postgresqlasyncpg://user:passwordhost:5432/dbname

engine

create_async_engine(DATABASE_URL,

echoTrue,

FastAPI()​​app.post(/lamps/{lamp_id}/on)async

def

session.execute(select(Lamp).where(Lamp.id

lamp_id))

pytest_asyncio.fixture(scopefunction,

nametest_session)

conn.run_sync(Base.metadata.create_all)

​yield

conn.run_sync(Base.metadata.drop_all)

​await

Lamp()test_session.add(lamp)await

​logging.error(New

async_client.post(f/lamps/{lamp.id}/on)assert

response.status_code

test_session.execute(select(Lamp).where(Lamp.id

lamp.id))new_lamp

results.scalar_one_or_none()logging.error(Updated

status:

这是一个常规的Pytest它在一个fixture中获取到数据库的会话。

在返回会话之前此fixture会先创建所有的表格使用之后它们会被删除。

fixture的会话而在主代码中我们使用来自app/database.py文件的会话。

尽管我们使用相同的引擎但是生成的会话是不同的。

这一点很重要。

应从数据库返回status

在测试中我首先在数据库中创建一个对象。

这是通过来自测试的会话进行的常规INSERT。

我们称其为Session

然后我对/lamps/1/on端点执行一个POST请求。

这是打开灯的操作。

为了使示例更短我没有使用fixture。

一旦请求开始工作将创建一个新的数据库会话。

我们称其为Session

2。

使用这个会话我从数据库加载所需的对象。

我将状态输出到日志。

它是off。

之后我更新了这个状态并在数据库中保存了更新。

数据库收到了一个请求

BEGIN

请注意也存在COMMIT命令。

尽管事务是隐式的但其结果在其他会话中在COMMIT之后立即可用。

接下来我使用refresh发出请求从数据库获取更新后的对象。

我输出状态。

现在它的值是on。

在测试中我从Session

1发出常规请求以获取修改过的对象。

但在状态字段中我看到的值是off。

同时根据所有日志最后一个SELECT请求被执行并返回了status

结果是为获取新对象而做的请求并没有更新当前的对象而是找到并使用了现有的对象。

一开始我使用ORM添加了一个灯泡对象。

我在另一个会话中更改了它。

当更改完成时当前会话对此更改一无所知。

并且在Session

依赖性覆盖

为了使用相同的会话您可以简单地使用我在测试中创建的那个覆盖应用程序中的会话。

这很简单。

async

test_sessionapp.dependency_overrides[get_session]

_override_get_db

app.dependency_overrides[get_session]

_override_get_db

async_client.post(f/lamps/{lamp.id}/on)

assert

test_session.execute(select(Lamp).where(Lamp.id

1))

但是如果应用程序使用多个会话这是可能的那么这可能不是最佳方法。

此外如果在被测试的函数中没有调用commit或rollback那么这也将无济于事。

第二种解决方案是最简单和最有逻辑的。

我们不应该创建一个新的请求来获取一个对象。

为了更新处理端点请求后立即调用刷新就足够了。

在内部它调用expires这导致保存的实例不用于新的请求并且数据重新填充。

这种解决方案是最有逻辑的也最容易理解。

await

之后你不需要再试图加载new_lamp对象只需检查相同的lamp。

def

async_client.post(f/lamps/{lamp.id}/on)

assert

但是如果我们更改了很多对象最好调用expire_all。

然后所有实例都将从数据库中读取一致性不会被破坏。

def

async_client.post(f/lamps/{lamp.id}/on)

assert

test_session.execute(select(Lamp).where(Lamp.id

1))

实际上使用会话终止的最后一种方法也调用了expire_all但会话可以进一步使用。

当读取新数据时我们将获得最新的对象。

await

async_client.post(f/lamps/{lamp.id}/on)

assert

test_session.execute(select(Lamp).where(Lamp.id

1))

expire_all但它明确地回滚了事务。

如果需要执行事务commit()

也会执行

expire_all。

但在这个例子中既不需要回滚也不需要提交因为测试中的事务已经完成应用程序中的事务不会影响来自测试的会话。

实际上此功能仅在SQLAlchemy

ORM的异步模式中在事务中工作。

然而在代码中我确实向数据库发出请求以获取新对象的行为看起来似乎不合逻辑如果它仍然返回一个缓存的对象而不是从数据库强制接收的对象。

当调试代码时这有点令人困惑。

但当正确使用时这就是它应有的样子。

在异步模式下使用SQLAlchemy

ORM您必须并行跟踪事务和线程中的会话。

如果所有这些看起来太复杂那么使用SQLAlchemy

作者Aleksei

squids.cn云数据库RDS迁移工具DBMotion云备份DBTwin等数据库生态工具。