FastAPI Examples¶

from tortoise import fields, models


class Users(models.Model):
    """
    The User model
    """

    id = fields.IntField(primary_key=True)
    #: This is a username
    username = fields.CharField(max_length=20, unique=True)
    name = fields.CharField(max_length=50, null=True)
    family_name = fields.CharField(max_length=50, null=True)
    category = fields.CharField(max_length=30, default="misc")
    password_hash = fields.CharField(max_length=128, null=True)
    created_at = fields.DatetimeField(auto_now_add=True)
    modified_at = fields.DatetimeField(auto_now=True)

    def full_name(self) -> str:
        """
        Returns the best name
        """
        if self.name or self.family_name:
            return f"{self.name or ''} {self.family_name or ''}".strip()
        return self.username

    class PydanticMeta:
        computed = ["full_name"]
        exclude = ["password_hash"]

# mypy: no-disallow-untyped-decorators
# pylint: disable=E0611,E0401
import multiprocessing
from collections.abc import AsyncGenerator
from concurrent.futures import ProcessPoolExecutor
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path

import anyio
import pytest
from asgi_lifespan import LifespanManager
from httpx import ASGITransport, AsyncClient

from tortoise.contrib.test import truncate_all_models
from tortoise.fields.data import JSON_LOADS
from tortoise.timezone import UTC, localtime

try:
    from config import register_orm
    from main import app
    from main_custom_timezone import app as app_east
    from models import Users
    from schemas import User_Pydantic
except ImportError:
    if (cwd := Path.cwd()) == (parent := Path(__file__).parent):
        dirpath = "."
    else:
        dirpath = str(parent.relative_to(cwd))
    print(f"You may need to explicitly declare python path:\n\nexport PYTHONPATH={dirpath}\n")
    raise

ClientManagerType = AsyncGenerator[AsyncClient, None]


@pytest.fixture(scope="module")
def anyio_backend() -> str:
    return "asyncio"


@asynccontextmanager
async def client_manager(app, base_url="http://test", **kw) -> ClientManagerType:
    async with LifespanManager(app):
        transport = ASGITransport(app=app)
        async with AsyncClient(transport=transport, base_url=base_url, **kw) as c:
            yield c


@pytest.fixture(scope="module")
async def client() -> ClientManagerType:
    async with client_manager(app) as c:
        await truncate_all_models()
        yield c


@pytest.fixture(scope="module")
async def client_east() -> ClientManagerType:
    # app_east uses _enable_global_fallback=False, so we need to explicitly
    # enter the context from app.state to make it current for tests
    async with client_manager(app_east) as c:
        ctx = app_east.state._tortoise_context
        with ctx:  # Enter context to make it current via contextvar
            await truncate_all_models()
            yield c


class UserTester:
    async def create_user(self, async_client: AsyncClient) -> Users:
        response = await async_client.post("/users", json={"username": "admin"})
        assert response.status_code == 200, response.text
        data = response.json()
        assert data["username"] == "admin"
        assert "id" in data
        user_id = data["id"]

        user_obj = await Users.get(id=user_id)
        assert user_obj.id == user_id
        return user_obj

    async def user_list(self, async_client: AsyncClient) -> tuple[datetime, Users, User_Pydantic]:
        utc_now = datetime.now(UTC)
        user_obj = await Users.create(username="test")
        response = await async_client.get("/users")
        assert response.status_code == 200, response.text
        data = response.json()
        assert isinstance(data, list)
        item = await User_Pydantic.from_tortoise_orm(user_obj)
        item_dict = JSON_LOADS(item.model_dump_json())
        api_item = next((x for x in data if x["id"] == user_obj.id), None)
        assert api_item is not None, f"User {user_obj.id} not found in response"
        for key, value in item_dict.items():
            assert key in api_item, f"Key {key!r} missing from API response"
            if key in ("created_at", "modified_at"):
                # Compare as datetime objects to handle timezone format differences
                # (Pydantic normalizes to UTC, FastAPI preserves original timezone)
                # Replace trailing 'Z' with '+00:00' for fromisoformat() compatibility
                a = datetime.fromisoformat(api_item[key].replace("Z", "+00:00"))
                b = datetime.fromisoformat(value.replace("Z", "+00:00"))
                assert a == b, f"Datetime mismatch on {key!r}: {api_item[key]} != {value}"
            else:
                assert api_item[key] == value, f"Mismatch on {key!r}"
        return utc_now, user_obj, item


class TestUser(UserTester):
    @pytest.mark.anyio
    async def test_create_user(self, client: AsyncClient) -> None:  # nosec
        await self.create_user(client)

    @pytest.mark.anyio
    async def test_user_list(self, client: AsyncClient) -> None:  # nosec
        await self.user_list(client)


@pytest.mark.anyio
async def test_404(client: AsyncClient) -> None:
    response = await client.get("/404")
    assert response.status_code == 404, response.text
    data = response.json()
    assert isinstance(data["detail"], str)


@pytest.mark.anyio
async def test_422(client: AsyncClient) -> None:
    response = await client.get("/422")
    assert response.status_code == 422, response.text
    data = response.json()
    assert isinstance(data["detail"], list)
    assert isinstance(data["detail"][0], dict)


class TestUserEast(UserTester):
    timezone = "Asia/Shanghai"
    delta_hours = 8

    @pytest.mark.anyio
    async def test_create_user_east(self, client_east: AsyncClient) -> None:  # nosec
        user_obj = await self.create_user(client_east)
        created_at = user_obj.created_at

        # Verify time zone
        asia_now = localtime(timezone=self.timezone)
        assert created_at.hour - asia_now.hour == 0

        # UTC timezone
        utc_now = localtime(timezone="UTC")
        assert (created_at.hour - utc_now.hour) in [self.delta_hours, self.delta_hours - 24]

    @pytest.mark.anyio
    async def test_user_list(self, client_east: AsyncClient) -> None:  # nosec
        time, user_obj, item = await self.user_list(client_east)
        created_at = user_obj.created_at
        assert (created_at.hour - time.hour) in [self.delta_hours, self.delta_hours - 24]
        assert item.model_dump()["created_at"].hour == created_at.hour


@pytest.mark.anyio
async def test_404_east(client_east: AsyncClient) -> None:
    response = await client_east.get("/404")
    assert response.status_code == 404, response.text
    data = response.json()
    assert isinstance(data["detail"], str)


@pytest.mark.anyio
async def test_422_east(client_east: AsyncClient) -> None:
    response = await client_east.get("/422")
    assert response.status_code == 422, response.text
    data = response.json()
    assert isinstance(data["detail"], list)
    assert isinstance(data["detail"][0], dict)


def query_without_app(pk: int) -> int:
    async def runner() -> bool:
        async with register_orm():
            return await Users.filter(id__gt=pk).count()

    return anyio.run(runner)


def test_query_without_app():
    multiprocessing.set_start_method("spawn")
    with ProcessPoolExecutor(max_workers=1) as executor:
        future = executor.submit(query_without_app, 0)
        assert future.result() >= 0

# pylint: disable=E0611,E0401
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from config import register_orm
from fastapi import FastAPI
from routers import router as users_router

from tortoise.contrib.fastapi import tortoise_exception_handlers


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    async with register_orm(app):
        # db connected
        yield
        # app teardown
    # db connections closed


app = FastAPI(
    title="Tortoise ORM FastAPI example",
    lifespan=lifespan,
    exception_handlers=tortoise_exception_handlers(),
)
app.include_router(users_router, prefix="")