Add documentation about how to use the async tools (session, etc) · Issue #626 · fastapi/sqlmodel · GitHub

Privileged issue

I'm @tiangolo or he asked me directly to create an issue here.

Issue Content

Add documentation about how to use the async tools (session, etc).

Funding

You can sponsor this specific effort via a Polar.sh pledge below
We receive the pledge once the issue is completed & verified

First of all we have to import from sqlalchemy which I dont like. It is a bit confusing. Hence I would like to import it from sqlmodel itself.

Like how we have

with Session(engine) as session:
    session.exec()

we should have

async with AsyncSession(engine) as session:
    await session.exec()

so it becomes easier for us to make sessions.

from collections.abc import AsyncGenerator
from typing import Annotated, Callable

from fastapi import Depends
from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, Session

from core.config import settings

##############

# region 'Async Session'
engine = create_async_engine(
    settings.SQLALCHEMY_DATABASE_URI_ASYNC,
    future=True,
    echo=settings.LOCAL_DEV,
    hide_parameters=not settings.LOCAL_DEV,
    connect_args={
        # https://www.postgresql.org/docs/current/runtime-config.html
        "server_settings": {
            "application_name": f"{settings.PROJECT_NAME} {settings.VERSION} async",
            "jit": "off",
        },
    },
)

AsyncSessionFactory = sessionmaker(
    bind=engine,
    autoflush=False,
    expire_on_commit=False,
    class_=AsyncSession,
)


async def get_db() -> AsyncGenerator:
    yield AsyncSessionFactory


Session = Annotated[AsyncSession, Depends(get_db)]
# endregion


##############
@router.post(...)
async def some_function(
    session: Session)
    async with session() as db:  # noqa
         ...
        stmt = select(Model1)
        items = await db.execute(stmt)
        data = items.all()
        return data

asyncpg, works fine, except:

asyncpg cant execute sql-files with multistatements
asyncpg not shows params statements

Added Documentation for usage with async tools. Fixes fastapi#626

A full working example of the following is here: https://github.com/deshetti/sqlmodel-async-example
Opened a PR to add documentation to the docs: #633

from contextlib import asynccontextmanager
from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel import Field, SQLModel

# Initialize FastAPI application
app = FastAPI()


# Define User model for SQLModel
class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: int


# Define UserCreate model for Pydantic validation
# For id field to not show up on the OpenAPI spec
class UserCreate(BaseModel):
    name: str
    age: int


# Database connection string
DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost/sampledb"

# Create an asynchronous engine for the database
engine = create_async_engine(
    DATABASE_URL,
    echo=True,
    future=True,
    pool_size=20,
    max_overflow=20,
    pool_recycle=3600,
)


# Ayschronous Context manager for handling database sessions
@asynccontextmanager
async def get_session() -> AsyncSession:
    async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    async with async_session() as session:
        yield session


# Function to create a new user in the database
async def create_user(user: User) -> User:
    async with get_session() as session:
        session.add(user)
        await session.commit()
        await session.refresh(user)
    return user


# Event handler for startup event of FastAPI application
@app.on_event("startup")
async def on_startup():
    async with engine.begin() as conn:
        # For SQLModel, this will create the tables (but won't drop existing ones)
        await conn.run_sync(SQLModel.metadata.create_all)


# Endpoint to create a new user
@app.post("/users/", response_model=User)
async def create_user_endpoint(user: UserCreate):
    db_user = User(**user.dict())
    result = await create_user(db_user)
    return result


# Main entry point of the application
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)

First of all we have to import from sqlalchemy which I dont like. It is a bit confusing. Hence I would like to import it from sqlmodel itself.

Like how we have
with Session(engine) as session:
    session.exec()
we should have
async with AsyncSession(engine) as session:
    await session.exec()
so it becomes easier for us to make sessions.

@tiangolo Can I work on this? Add a sqlmodel version of asyncsession as well as create_async_engine

First of all we have to import from sqlalchemy which I dont like. It is a bit confusing. Hence I would like to import it from sqlmodel itself.

Like how we have
with Session(engine) as session:
    session.exec()
we should have
async with AsyncSession(engine) as session:
    await session.exec()
so it becomes easier for us to make sessions.

It looks like sqlmodel has already implemented its own asyncsession but is just not importable from the root directory. Currently I believe you need to import it through
from sqlmodel.ext.asyncio.session import AsyncSession

If we want to adhere to sqlalchemy model import structure it should be
from sqlmodel.ext.asyncio import AsyncSession

However, according to @alvynabranches the proposed solution is
from sqlmodel import AsyncSession

@tiangolo what do you think?

@tiangolo when do you plan to add a fully-functional async support to this library?

We are using the the SQL Alchemy at the moment and recently rewrote our code to use async. But problem is in the async m2m and m2o relations, that we need to use .awaitable_attrs that our code a bit ugly. Look on this:

docs = await folder.awaitable_attrs.documents
docs_to_authors = {doc.id: await doc.awaitable_attrs.authors for doc in docs}

Do you plan to get rid of such awaitable_attrs syntax in your version and have something like this?

docs = await folder.documents
docs_to_authors = {doc.id: await doc.authors for doc in docs}

so that under the hood it will know that "documents" and "authors" are actually awaitable props and we need to use await keyword for them. Otherwise mypy needs to catch such a problems and notify us that we forgot to add await keyword for such m2m and m2o relations when trying to load them from DB.

@tiangolo when do you plan to add a fully-functional async support to this library?

We are using the the SQL Alchemy at the moment and recently rewrote our code to use async. But problem is in the async m2m and m2o relations, that we need to use .awaitable_attrs that our code a bit ugly. Look on this:
docs = await folder.awaitable_attrs.documents
docs_to_authors = {doc.id: await doc.awaitable_attrs.authors for doc in docs}
Do you plan to get rid of such awaitable_attrs syntax in your version and have something like this?
docs = await folder.documents
docs_to_authors = {doc.id: await doc.authors for doc in docs}
so that under the hood it will know that "documents" and "authors" are actually awaitable props and we need to use await keyword for them. Otherwise mypy needs to catch such a problems and notify us that we forgot to add await keyword for such m2m and m2o relations when trying to load them from DB.

If you want to fetch all related data, I suggest you to try selectinload

from sqlalchemy.orm import selectinload
...
(await session.exec(select(User).options(selectinload(User.roles))).all()

When we were using SQLAlchemy with Pydantic, we introduced a mix-in to handle the fetching of awaitable_attrs.

class ModelInDBMixin(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    created_at: Optional[dt.datetime] = None
    updated_at: Optional[dt.datetime] = None

    @classmethod
    async def from_data(cls, obj: Base, **kwargs):
        fields = [
            field
            for field in dir(obj)
            if not field.startswith("_")
            and field != "awaitable_attrs"
            and field in cls.model_fields.keys()
        ]
        data = await asyncio.gather(
            *[get_attr_or_none(obj, field) for field in fields]
        )
        data_dict = {k: v for (k, v) in zip(fields, data) if v is not None}
        data_dict.update(kwargs)
        try:
            return cls.model_validate(data_dict)
        except Exception as e:
            raise e

Usage looking like User.from_data(db_obj) just in the same way you'd call from_orm in the original api.

Now, with SQLModel, we're monkeypatching sqlmodel_validate:

import asyncio
from collections.abc import Mapping
from typing import Any

import sqlmodel._compact  # type: ignore
from pydantic_core import PydanticUndefined as Undefined
from sqlmodel._compat import (ObjectWithUpdateWrapper, _TSQLModel,
                              is_table_model_class, partial_init)


async def get_attr_or_none(obj: Any, attr: str) -> Any:
    try:
        # make sure attr is not a hybrid_method
        if hasattr(getattr(obj, attr), "__call__"):
            return None
        return await getattr(obj.awaitable_attrs, attr)
    except Exception:
        return None

def sqlmodel_validate(
    cls: type[_TSQLModel],
    obj: Any,
    *,
    strict: bool | None = None,
    from_attributes: bool | None = None,
    context: Mapping[str, Any] | None = None,
    update: Mapping[str, Any] | None = None,
) -> _TSQLModel:
    asyncio_loop = asyncio.get_event_loop()
    if not is_table_model_class(cls):
        new_obj: _TSQLModel = cls.__new__(cls)
    else:
        # If table, create the new instance normally to make SQLAlchemy create
        # the _sa_instance_state attribute
        with partial_init():
            new_obj = cls()
    
    # SQLModel Override to get class SQLAlchemy __dict__ attributes and
    # set them back in after creating the object
    old_dict = new_obj.__dict__.copy()
    use_obj = obj
    
    # Handle awaitable attributes if they exist
    if hasattr(obj, 'awaitable_attrs'):
        # Get all model fields that might be awaitable
        fields = [
            field for field in cls.model_fields.keys()
            if not field.startswith("_")
        ]
        # Fetch all awaitable attributes
        awaited_values = asyncio_loop.run_until_complete(
            asyncio.gather(*[
                get_attr_or_none(obj, field) 
                for field in fields
            ])
        )
        # Create a dictionary of non-None awaited values
        awaited_dict = {
            k: v for k, v in zip(fields, awaited_values)
            if v is not None
        }
        # If obj is a dict, update it with awaited values
        if isinstance(obj, dict):
            use_obj = {**obj, **awaited_dict}
        else:
            # Create a new object that combines original and awaited attributes
            use_obj = ObjectWithUpdateWrapper(obj=obj, update=awaited_dict)
    
    # Apply any explicit updates
    if isinstance(use_obj, dict) and update:
        use_obj = {**use_obj, **update}
    elif update:
        use_obj = ObjectWithUpdateWrapper(obj=use_obj, update=update)

    cls.__pydantic_validator__.validate_python(
        use_obj,
        strict=strict,
        from_attributes=from_attributes,
        context=context,
        self_instance=new_obj,
    )
    
    # Capture fields set to restore it later
    fields_set = new_obj.__pydantic_fields_set__.copy()
    if not is_table_model_class(cls):
        # If not table, normal Pydantic code, set __dict__
        new_obj.__dict__ = {**old_dict, **new_obj.__dict__}
    else:
        # Do not set __dict__, instead use setattr to trigger SQLAlchemy
        # instrumentation
        for key, value in {**old_dict, **new_obj.__dict__}.items():
            setattr(new_obj, key, value)
    
    # Restore fields set
    object.__setattr__(new_obj, "__pydantic_fields_set__", fields_set)
    
    # Get and set any relationship objects
    if is_table_model_class(cls):
        for key in new_obj.__sqlmodel_relationships__:
            value = getattr(use_obj, key, Undefined)
            if value is not Undefined:
                setattr(new_obj, key, value)
    
    return new_obj


sqlmodel._compat.sqlmodel_validate = sqlmodel_validate

Here's a working example showing how to use async tools like AsyncSession with SQLModel:

from sqlmodel import SQLModel, Field
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
import asyncio

# Define your model
class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str

# Setup async engine and sessionmaker
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

# Create database tables
async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

# Insert a record using async session
async def create_hero():
    async with async_session() as session:
        async with session.begin():
            hero = Hero(name="Deadpond", secret_name="Dive Wilson")
            session.add(hero)

# Run setup and example
async def main():
    await init_db()
    await create_hero()

asyncio.run(main())

@tiangolo Any update on this issue?

For me, AsyncSession had to be imported from sqlmodel.ext.asyncio.session import AsyncSession everywhere and not from sqlalchemy.ext.asyncio.session import AsyncSession.

Otherwise, I saw the error _AsyncGeneratorContextManager' object has no attribute 'exec'.

Here is my working code:

from sqlmodel.ext.asyncio.session import AsyncSession

async def get_async_session() -> AsyncSession:
    async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    async with async_session() as session:
        yield session

from sqlmodel.ext.asyncio.session import AsyncSession

@app.get("/get_heros")
async def get_heros(session: AsyncSession = Depends(get_async_session)):
    query = select(Hero)
    result = await session.exec(query)
    heros = result.all()
    return [Hero(name=hero.name, id=hero.id) for hero in heros]

I needed to specify poolclass for connecting with TURSO

engine = create_async_engine(
    f"sqlite+aiolibsql://{TURSO_DATABASE_URL.strip()}?secure=true",
    connect_args={"auth_token": f"{quote_plus(TURSO_AUTH_TOKEN).strip()}"},
    echo=True,
    future=True,
    poolclass=AsyncAdaptedQueuePool,
)


# Async session factory
AsyncSessionLocal = async_sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

Here's a migration guide for moving SQLModel from sync to async covering the essential setup and common pitfalls:
https://gist.github.com/rmaceissoft/bda3b6519334a34a9982b00ba02285dd

I might be wrong here but the deprecation message
in sqlmodel for session.execute() is a bit confusing.

It suggests using:

heroes = await session.exec(select(Hero)).all()

But that actually raises:

AttributeError: 'coroutine' object has no attribute 'all'

That’s because session.exec() is async you need to await it before calling .all().
So the correct version should be:

heroes = (await session.exec(select(Hero))).all()

Might be worth updating the deprecation message to avoid confusion.

in sqlmodel 0.0.27
sqlmodel/ext/asyncio/session.py ln 112

Oh, you are right. This is a mistake in deprecation message. PR is welcome!

Hi @YuriiMotov can you take a took at this PR #1643

I'm working on a comprehensive Async tutorial for SQLModel, covering , , and integration with FastAPI. I'll also fix the deprecation warning typo in . PR incoming.

You should consider using methods such as unique to avoid type-checking errors. For example: https://github.com/AndreuCodina/python-template/blob/main/src/python_template/api/workflows/products/discontinue_product/discontinue_product_workflow.py

docs: add async/await tutorial and examples (fixes fastapi#626)

Intent to claim. I have developed a comprehensive async tools guide with examples for SQLModel using our Swarm AI engine. Will submit PR shortly.

All, thanks a lot for the engagement here 🙏

We've discussed this internally and decided that we prefer to write the first draft of this new documentation section ourselves. We are now tracking this on our internal TODO list. We'll go ahead and close this issue, as it will otherwise continue to draw external contributions, which has intensified a lot over the past year due to LLM usage.

Again, thanks to everyone who contributed, we'll be taking all comments & contributions into account when we publish the first draft for this.

Add documentation about how to use the async tools (session, etc) · Issue #626 · fastapi/sqlmodel · GitHub

Repository navigation

Add documentation about how to use the async tools (session, etc) #626

Privileged issue

Issue Content

Funding

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Add documentation about how to use the async tools (session, etc) · Issue #626 · fastapi/sqlmodel · GitHub

Repository navigation

Add documentation about how to use the async tools (session, etc) #626

Description

Privileged issue

Issue Content

Funding

Activity

alvynabranches commented on Jul 31, 2023

MatsiukMykola commented on Aug 7, 2023

deshetti commented on Aug 8, 2023

PookieBuns commented on Aug 9, 2023

PookieBuns commented on Aug 9, 2023

1st commented on May 28, 2024

ryanrain2016 commented on Jun 27, 2024

tonyf commented on Apr 13, 2025

prerna-pramod-gh commented on Apr 14, 2025

darhsu commented on Sep 18, 2025

XChikuX commented on Sep 25, 2025

rmaceissoft commented on Sep 29, 2025

DanielLeviLucas commented on Oct 20, 2025

YuriiMotov commented on Oct 20, 2025

DanielLeviLucas commented on Oct 20, 2025

YuriiMotov commented on Oct 20, 2025

DanielLeviLucas commented on Oct 21, 2025

DanielLeviLucas commented on Nov 11, 2025

cscguochang commented on Feb 1, 2026

AndreuCodina commented on Feb 1, 2026

Adraca commented on Mar 4, 2026

svlandeg commented on May 13, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions