Coverage for tests / test_asyncio.py: 100%

1import asyncio 9 ctx1jklmnopqr

2from collections.abc import Mapping 9 ctx1jklmnopqr

3from typing import Any 9 ctx1jklmnopqr

4 

5import pytest 9 ctx1jklmnopqr

6from sqlalchemy import update 9 ctx1jklmnopqr

7from sqlalchemy.engine import CursorResult, ScalarResult 9 ctx1jklmnopqr

8from sqlalchemy.ext.asyncio import create_async_engine 9 ctx1jklmnopqr

9from sqlmodel import Field, SQLModel, col, select 9 ctx1jklmnopqr

10from sqlmodel.ext.asyncio.session import AsyncSession 9 ctx1jklmnopqr

11from typing_extensions import Unpack, assert_type 9 ctx1jklmnopqr

12 

13 

14@pytest.mark.parametrize("execution_options", [{}, {"populate_existing": True}]) 9 ctx1jklmnopqr

15def test_async_session(execution_options: Mapping[str, Any]) -> None: 9 ctx1jklmnopqr

16 class Hero(SQLModel, table=True): 9 ctx1abcdefghi

17 id: int | None = Field(default=None, primary_key=True) 9 ctx1abcdefghi

18 name: str 9 ctx1abcdefghi

19 

20 async def run() -> None: 9 ctx1abcdefghi

21 engine = create_async_engine("sqlite+aiosqlite://") 9 ctx1abcdefghi

22 try: 9 ctx1abcdefghi

23 async with engine.begin() as connection: 9 ctx1abcdefghi

24 await connection.run_sync(SQLModel.metadata.create_all) 9 ctx1abcdefghi

25 async with AsyncSession(engine) as session: 9 ctx1abcdefghi

26 session.add(Hero(name="Deadpond")) 9 ctx1abcdefghi

27 await session.commit() 9 ctx1abcdefghi

28 

29 statement = select(Hero) 9 ctx1abcdefghi

30 result = await session.exec( 9 ctx1abcdefghi

31 statement, execution_options=execution_options 

32 ) 

33 assert_type(result, ScalarResult[Hero]) 9 ctx1abcdefghi

34 assert result.one().name == "Deadpond" 9 ctx1abcdefghi

35 

36 with pytest.warns(DeprecationWarning, match="session.exec"): 9 ctx1abcdefghi

37 scalars = await session.scalars(statement) 9 ctx1abcdefghi

38 assert_type(scalars, ScalarResult[Hero]) 9 ctx1abcdefghi

39 assert scalars.one().name == "Deadpond" 9 ctx1abcdefghi

40 

41 row = (await session.exec(select(col(Hero.id), Hero.name))).one() 9 ctx1abcdefghi

42 assert_type(row, tuple[int | None, str]) 9 ctx1abcdefghi

43 assert row == (1, "Deadpond") 9 ctx1abcdefghi

44 

45 updated = await session.exec(update(Hero).values(name="Rusty-Man")) 9 ctx1abcdefghi

46 assert_type(updated, CursorResult[Unpack[tuple[Any, ...]]]) 9 ctx1abcdefghi

47 assert updated.rowcount == 1 9 ctx1abcdefghi

48 await session.commit() 9 ctx1abcdefghi

49 assert (await session.exec(select(Hero.name))).one() == "Rusty-Man" 9 ctx1abcdefghi

50 

51 with pytest.warns(DeprecationWarning, match="session.exec"): 9 ctx1abcdefghi

52 result = await session.execute(select(Hero)) # ty: ignore[deprecated] 9 ctx1abcdefghi

53 assert result.scalars().one().name == "Rusty-Man" 9 ctx1abcdefghi

54 finally: 

55 await engine.dispose() 9 ctx1abcdefghi

56 

57 asyncio.run(run()) 9 ctx1abcdefghi