BlackSheep Examples¶

import tortoise
from tortoise import fields, models
from tortoise.contrib.pydantic import pydantic_model_creator

if tortoise.__version__ >= "0.20":
    raise RuntimeError("blacksheep not support pydantic V2, use tortoise-orm<0.20 instead!")


class Users(models.Model):
    id = fields.UUIDField(pk=True)
    username = fields.CharField(max_length=63)

    def __str__(self) -> str:
        return f"User {self.id}: {self.username}"


UserPydanticOut = pydantic_model_creator(Users, name="UserOut")
UserPydanticIn = pydantic_model_creator(Users, name="UserIn", exclude_readonly=True)

# pylint: disable=E0401,E0611
from typing import Union
from uuid import UUID

from blacksheep import Response
from blacksheep.server import Application
from blacksheep.server.openapi.v3 import Info, OpenAPIHandler
from blacksheep.server.responses import created, no_content, ok
from models import UserPydanticIn, UserPydanticOut, Users

from tortoise.contrib.blacksheep import register_tortoise

app = Application()
register_tortoise(
    app,
    db_url="sqlite://:memory:",
    modules={"models": ["models"]},
    generate_schemas=True,
    add_exception_handlers=True,
)


docs = OpenAPIHandler(info=Info(title="Tortoise ORM BlackSheep example", version="0.0.1"))
docs.bind_app(app)


@app.router.get("/")
async def users_list() -> Union[UserPydanticOut]:
    return ok(await UserPydanticOut.from_queryset(Users.all()))


@app.router.post("/")
async def users_create(user: UserPydanticIn) -> UserPydanticOut:
    user = await Users.create(**user.dict(exclude_unset=True))
    return created(await UserPydanticOut.from_tortoise_orm(user))


@app.router.patch("/{id}")
async def users_patch(id: UUID, user: UserPydanticIn) -> UserPydanticOut:
    await Users.filter(id=id).update(**user.dict(exclude_unset=True))
    return ok(await UserPydanticOut.from_tortoise_orm(await Users.get(id=id)))


@app.router.put("/{id}")
async def users_put(id: UUID, user: UserPydanticIn) -> UserPydanticOut:
    await Users.filter(id=id).update(**user.model_dump())
    return ok(await UserPydanticOut.from_tortoise_orm(await Users.get(id=id)))


@app.router.delete("/{id}")
async def users_delete(id: UUID) -> Response:
    await Users.filter(id=id).delete()
    return no_content()