How to get working async tests for mini async example? #11445
-
First Check
Commit to Help
Example Code# async0.py
"""
An end-to-end minimal example of using async fastapi, sqlalchemy, pydantic
(no httpx and pytest).
Run app in one of two ways:
uvicorn async0:app --port 8000 --reload
python async0.py
Post an item:
curl -X POST -H "Content-Type: application/json" \
-d '{"title":"foo"}' http://127.0.0.1:8000/items
Get list of all items:
curl -X GET http://127.0.0.1:8000/items/
Check databse:
sqlite3 async0.db ".tables"
sqlite3 async0.db "SELECT * FROM items;"
"""
from contextlib import asynccontextmanager
from typing import AsyncGenerator, List
import uvicorn
from fastapi import Depends, FastAPI
from pydantic import BaseModel
from sqlalchemy import select, Column, String, Integer
from sqlalchemy.ext.asyncio import (
async_sessionmaker,
create_async_engine,
AsyncEngine,
AsyncSession,
)
from sqlalchemy.orm import declarative_base
# database
DATABASE_URL = "sqlite+aiosqlite:///./async0.db"
engine: AsyncEngine = create_async_engine(DATABASE_URL, future=True)
Base = declarative_base()
class ItemBase(Base): # ORM
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
class ItemModel(BaseModel): # Pydantic
title: str
class Config:
from_attributes = True
async def create_db_and_tables():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
# Create an async session maker
async_sessionmaker = async_sessionmaker(
bind=engine, class_=AsyncSession, expire_on_commit=False
)
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with async_sessionmaker() as session:
try:
yield session
finally:
await session.close()
# fastapi
@asynccontextmanager
async def lifespan(app: FastAPI):
# Before app starts
await create_db_and_tables()
try:
yield
finally:
# After the app stops
await engine.dispose()
app = FastAPI(lifespan=lifespan)
@app.get("/")
async def root():
return {"message": "Tomato"}
@app.post("/items", response_model=ItemModel)
async def create_item(item: ItemModel, db: AsyncSession = Depends(get_db)) -> ItemModel:
item_base = ItemBase(title=item.title)
db.add(item_base)
await db.commit()
await db.refresh(item_base)
return item_base
@app.get("/items", response_model=List[ItemModel])
async def read_items(
skip: int = 0, limit: int = 10, db: AsyncSession = Depends(get_db)
) -> List[ItemModel]:
result = await db.execute(select(ItemBase).offset(skip).limit(limit))
items = result.scalars().all()
return items
if __name__ == "__main__":
uvicorn.run("async0:app", port=8000, reload=True)DescriptionI want to get the shown working minimal async example (which combines FastAPI, Pydantic, SQLite) tested in any async way, with Pytest and, say, Httpx. But I have not found any example code for this combination and ChatGPT and Groq will generate code like be low, but it seems to always fail with this message: # async0_test.py
"""
Run with
pytest async0_test.py
"""
import httpx
import pytest
import pytest_asyncio
from async0 import app, ItemModel, engine, Base, select, ItemBase
# Configure pytest to use the asyncio event loop.
@pytest_asyncio.fixture(scope="session")
def event_loop():
import asyncio
loop = asyncio.get_event_loop()
yield loop
loop.close()
@pytest_asyncio.fixture(scope="session")
async def test_app():
# Drop and recreate the database schema
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
# Use httpx to create a client that can interact with the FastAPI app
async with httpx.AsyncClient(app=app, base_url="http://test") as client:
yield client
# Optionally clear data after tests
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest.mark.asyncio
async def test_create_item(test_app):
response = await test_app.post("/items", json={"title": "foo"})
assert response.status_code == 200
data = response.json()
assert data["title"] == "foo"
@pytest.mark.asyncio
async def test_read_items(test_app):
await test_app.post("/items", json={"title": "bar"})
await test_app.post("/items", json={"title": "baz"})
response = await test_app.get("/items")
assert response.status_code == 200
items = response.json()
assert len(items) == 3
assert items[0]["title"] == "foo"
assert items[1]["title"] == "bar"
assert items[2]["title"] == "baz"Operating SystemmacOS Operating System DetailsmacOS 13.6.4 FastAPI Version0.104.1 Pydantic Version2.4.2 Python Version3.11.5 Additional ContextSome more version numbers: SQLite 3.41.2 |
Beta Was this translation helpful? Give feedback.
Replies: 1 comment 6 replies
-
|
The traceback you posted is for |
Beta Was this translation helpful? Give feedback.
-
|
Appologies, the test code is the real one I've used, but the first line is wrong (since the normal pytest convention ist to start with a test_ prefix). I've corrected it now. |
Beta Was this translation helpful? Give feedback.
-
|
It's not true. The function signatures are different. In the traceback it's |
Beta Was this translation helpful? Give feedback.
-
|
Correct, indeed. Sorry, but I had forgotten a file renaming (and was then using |
Beta Was this translation helpful? Give feedback.
-
|
It's a known issue of pytest-asyncio (pytest-dev/pytest-asyncio#706). It's recommended to use version 0.21.1 or less of |
Beta Was this translation helpful? Give feedback.
-
|
That works, thanks a lot! |
Beta Was this translation helpful? Give feedback.
-
|
@deeplook can you share you github repo with sqlalchemy 2.0 with fastapi |
Beta Was this translation helpful? Give feedback.
It's a known issue of pytest-asyncio (pytest-dev/pytest-asyncio#706). It's recommended to use version 0.21.1 or less of
pytest-asyncioto avoid this problem