Async SQLModel FastAPI Application

[UncleDrew92]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from fastapi import FastAPI, HTTPException
from sqlmodel import SQLModel, create_engine, select, Session
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel.ext.asyncio.engine import create_async_engine
from typing import List
import asyncio

# Define your database models using SQLModel
class Item(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    description: str | None = None

# Create the FastAPI app
app = FastAPI()

# Define the async database URL
DATABASE_URL = "sqlite+aiosqlite:///./test.db"

# Create async database engine
engine = create_async_engine(DATABASE_URL, echo=True)

# Create the database tables
@app.on_event("startup")
async def on_startup():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

# Dependency to get async session
async def get_session() -> AsyncSession:
    async with AsyncSession(engine) as session:
        yield session

# Create an item
@app.post("/items/", response_model=Item)
async def create_item(item: Item, session: AsyncSession = Depends(get_session)):
    session.add(item)
    await session.commit()
    await session.refresh(item)
    return item

# Read all items
@app.get("/items/", response_model=List[Item])
async def read_items(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(Item))
    items = result.scalars().all()
    return items

# Read a single item by ID
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int, session: AsyncSession = Depends(get_session)):
    item = await session.get(Item, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

Description

I am curious when there is a plan to fill out the advanced section of the documentation that shows how to "use async and await with the async session?"

I am curious how to setup tests and avoid MissingGreenlet errors when there are instances of lazy-loading that is tries to get run synchronously.

https://sqlmodel.tiangolo.com/advanced/

Operating System

Windows

Operating System Details

No response

SQLModel Version

0.0.22

Python Version

3.12

Additional Context

No response

[1st]

I'm using SQL Alchemy with async in my project. But problem is that I need to call .awaitable_attrs attribute each time, to load a related object from DB. Problematic part here is that it doesn't provide typing data, and mypy can't work with a property by knowing its type (the name way as if I call without .awaitable_attrs in the middle.

Will be good to get some info from @tiangolo about timeline for this feature. As it seems guys are stuck a bit on this project and async support is a bit frozen. So I will live with the workaround that I made to use await before each .awaitable_attrs attribute.