Complete worked example
python
import asyncio
from pathlib import Path
from tempfile import TemporaryDirectory
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
class Base(DeclarativeBase):
pass
class Role(Base):
__tablename__ = "roles"
id: Mapped[int] = mapped_column(primary_key=True)
score: Mapped[int]
async def read_owned(scores, requested):
with TemporaryDirectory() as folder:
engine = create_async_engine("sqlite+aiosqlite:///" + str(Path(folder) / "owned.sqlite3"),
connect_args={"autocommit": False})
maker = async_sessionmaker(engine, expire_on_commit=False)
owners = []
try:
async with engine.begin() as connection:
await connection.run_sync(Base.metadata.create_all)
async with maker() as session, session.begin():
session.add_all(Role(id=i, score=value) for i, value in enumerate(scores, 1))
async def one(wanted):
async with maker() as session:
owners.append(session)
role = await session.get(Role, wanted)
await session.commit()
return None if role is None else role.score
tasks = [asyncio.create_task(one(wanted)) for wanted in requested]
try:
values = await asyncio.gather(*tasks)
except BaseException:
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
raise
return {"values": values, "sessions": len({id(owner) for owner in owners})}
finally:
await engine.dispose()
def inspect_example(scores, requested):
return asyncio.run(read_owned(scores, requested))- Reasoning
- The session factory uses expire_on_commit=False. Each task loads a Role, commits its read unit and returns an ordinary value without attempting an implicit attribute reload afterward. Requested IDs [2, 1, 99] over scores [4, 9] yield values [9, 4, None] from three distinct Session owners. The synchronous inspect_example wrapper runs a fresh event loop for this computer lab. Inside an already asynchronous application, await the asynchronous function instead of nesting asyncio.run in a running loop. For SQLAlchemy 2.1, install the asyncio extra so the required greenlet support is included alongside the chosen async driver.