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
import os
from concurrent.futures import ProcessPoolExecutor
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path
from typing import AsyncGenerator, Tuple

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

from tortoise.contrib.test import MEMORY_SQLITE
from tortoise.fields.data import JSON_LOADS

os.environ["DB_URL"] = MEMORY_SQLITE
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:
    app.state.testing = True
    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:
        yield c


@pytest.fixture(scope="module")
async def client_east() -> ClientManagerType:
    async with client_manager(app_east) as c:
        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(pytz.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)
        assert JSON_LOADS(item.model_dump_json()) in data
        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)


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_tz = pytz.timezone(self.timezone)
        asia_now = datetime.now(pytz.utc).astimezone(asia_tz)
        assert created_at.hour - asia_now.hour == 0

        # UTC timezone
        utc_tz = pytz.timezone("UTC")
        utc_now = datetime.now(pytz.utc).astimezone(utc_tz)
        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


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
import os
from contextlib import asynccontextmanager
from typing import AsyncGenerator

from fastapi import FastAPI
from routers import router as users_router

from examples.fastapi.config import register_orm
from tortoise import Tortoise, generate_config
from tortoise.contrib.fastapi import RegisterTortoise


@asynccontextmanager
async def lifespan_test(app: FastAPI) -> AsyncGenerator[None, None]:
    config = generate_config(
        os.getenv("TORTOISE_TEST_DB", "sqlite://:memory:"),
        app_modules={"models": ["models"]},
        testing=True,
        connection_label="models",
    )
    async with RegisterTortoise(
        app=app,
        config=config,
        generate_schemas=True,
        add_exception_handlers=True,
        _create_db=True,
    ):
        # db connected
        yield
        # app teardown
    # db connections closed
    await Tortoise._drop_databases()


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    if getattr(app.state, "testing", None):
        async with lifespan_test(app) as _:
            yield
    else:
        # app startup
        async with register_orm(app):
            # db connected
            yield
            # app teardown
        # db connections closed


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