Function asyncio

test/typing/plain_files/ext/asyncio/engines.py:22–95

()

async def asyncio() -> None:
    e = create_async_engine("sqlite://")

    assert_type(e, AsyncEngine)

    async with e.connect() as conn:
        assert_type(conn, AsyncConnection)

        result = await conn.execute(text("select * from table"))

        assert_type(result, CursorResult[Unpack[tuple[Any, ...]]])

        # stream with direct await
        async_result = await conn.stream(text("select * from table"))

        assert_type(async_result, AsyncResult[Unpack[tuple[Any, ...]]])

        # stream with context manager
        async with conn.stream(
            text("select * from table")
        ) as ctx_async_result:
            assert_type(ctx_async_result, AsyncResult[Unpack[tuple[Any, ...]]])

        # stream_scalars with direct await
        async_scalar_result = await conn.stream_scalars(
            text("select * from table")
        )

        assert_type(async_scalar_result, AsyncScalarResult[Any])

        # stream_scalars with context manager
        async with conn.stream_scalars(
            text("select * from table")
        ) as ctx_async_scalar_result:
            assert_type(ctx_async_scalar_result, AsyncScalarResult[Any])

    async with e.begin() as conn:
        assert_type(conn, AsyncConnection)

        result = await conn.execute(text("select * from table"))

        assert_type(result, CursorResult[Unpack[tuple[Any, ...]]])

        await conn.run_sync(work_sync, 1)

        # EXPECTED_MYPY: Missing positional argument "foo" in call to "run_sync" of "AsyncConnection"
        await conn.run_sync(work_sync)

    ce = select(1).compile(e)
    ce.statement
    cc = select(1).compile(conn)
    cc.statement

    async with e.connect() as conn:
        metadata = MetaData()

        await conn.run_sync(metadata.create_all)
        await conn.run_sync(metadata.reflect)