blob: c920ad55dc6fb74138a956a3f7a60c1e3f584b77 (
plain)
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
from sqlalchemy import create_engine
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
def regular() -> None:
e = create_engine("sqlite://")
# EXPECTED_TYPE: Engine
reveal_type(e)
with e.connect() as conn:
# EXPECTED_TYPE: Connection
reveal_type(conn)
result = conn.execute(text("select * from table"))
# EXPECTED_TYPE: CursorResult[Any]
reveal_type(result)
with e.begin() as conn:
# EXPECTED_TYPE: Connection
reveal_type(conn)
result = conn.execute(text("select * from table"))
# EXPECTED_TYPE: CursorResult[Any]
reveal_type(result)
async def asyncio() -> None:
e = create_async_engine("sqlite://")
# EXPECTED_TYPE: AsyncEngine
reveal_type(e)
async with e.connect() as conn:
# EXPECTED_TYPE: AsyncConnection
reveal_type(conn)
result = await conn.execute(text("select * from table"))
# EXPECTED_TYPE: CursorResult[Any]
reveal_type(result)
# stream with direct await
async_result = await conn.stream(text("select * from table"))
# EXPECTED_TYPE: AsyncResult[Any]
reveal_type(async_result)
# stream with context manager
async with conn.stream(
text("select * from table")
) as ctx_async_result:
# EXPECTED_TYPE: AsyncResult[Any]
reveal_type(ctx_async_result)
# stream_scalars with direct await
async_scalar_result = await conn.stream_scalars(
text("select * from table")
)
# EXPECTED_TYPE: AsyncScalarResult[Any]
reveal_type(async_scalar_result)
# stream_scalars with context manager
async with conn.stream_scalars(
text("select * from table")
) as ctx_async_scalar_result:
# EXPECTED_TYPE: AsyncScalarResult[Any]
reveal_type(ctx_async_scalar_result)
async with e.begin() as conn:
# EXPECTED_TYPE: AsyncConnection
reveal_type(conn)
result = await conn.execute(text("select * from table"))
# EXPECTED_TYPE: CursorResult[Any]
reveal_type(result)
|
