-
Notifications
You must be signed in to change notification settings - Fork 2
/
db_test.py
40 lines (31 loc) · 1.14 KB
/
db_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
"""数据库连接测试"""
import asyncio
from sqlalchemy import select, insert
from db.models import Demo
from db import async_session_local
async def test_select():
async with async_session_local() as session:
async with session.begin():
_exec = select(Demo).where(Demo.id == 1)
result: Demo = (await session.execute(_exec)).scalars().first()
print(dict(result))
async def test_select_by_pk(pk):
async with async_session_local() as session:
async with session.begin():
if (result := await session.get(Demo, pk)) is not None:
print(dict(result))
async def test_insert(name='插入测试'):
async with async_session_local() as session:
async with session.begin():
_exec = insert(Demo).values(name=name)
await session.execute(_exec)
await session.commit()
async def main():
"""
使用`asyncio.run`结束时抛出`RuntimeError: Event loop is closed`
是Windows平台上常见且可以忽略的异常
"""
await test_select()
await test_select_by_pk(2)
await test_insert()
asyncio.run(main())