快速API

将 ormar 与 fastapi 结合使用非常简单。

除了在启动时连接到数据库之外，您需要做的所有其他事情就是用 ormar 模型替换 pydantic 模型。

在这里您可以找到一个非常简单的示例应用程序代码。

！！！警告此示例假设您已经创建了一个数据库。如果不是这种情况，请访问数据库初始化部分。

!!!tip 以下示例（所有部分）应放在一个文件中。

It's divided into subsections for clarity.

!!!note 如果您想了解有关如何在 fastapi 请求和响应中使用 ormar 模型的更多信息，请查看响应和请求文档。

快速入门

!!!note 请注意，您可以在 github 存储库的示例下找到完整的快速启动脚本。

导入和初始化

使用 FastAPI 生命周期定义启动和关闭程序，并在应用程序中使用。

from typing import List, Optional

import databases
import sqlalchemy
from fastapi import FastAPI

import ormar

from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    if not config.database.is_connected:
        await config.database.connect()

    yield

    if config.database.is_connected:
        await config.database.disconnect()


base_ormar_config = ormar.OrmarConfig(
    metadata=sqlalchemy.MetaData(),
    database=databases.Database("sqlite:///test.db"),
)
app = FastAPI(lifespan=lifespan(base_ormar_config))

Info

型号定义

使用适当的字段定义 ormar 模型。

将使用这些模型来代替 pydantic 模型。

base_ormar_config = OrmarConfig(
    metadata = metadata
    database = database
)


class Category(ormar.Model):
    ormar_config = base_ormar_config.copy(tablename="categories")

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)


class Item(ormar.Model):
    ormar_config = base_ormar_config.copy()

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)
    category: Optional[Category] = ormar.ForeignKey(Category, nullable=True)

!!!提示您可以在模型部分阅读有关定义模型的更多信息。

Fastapi端点定义

定义您所需的端点，注意 ormar 模型如何用作response_model 和请求参数。

@app.get("/items/", response_model=List[Item])
async def get_items():
    items = await Item.objects.select_related("category").all()
    return items


@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    await item.save()
    return item


@app.post("/categories/", response_model=Category)
async def create_category(category: Category):
    await category.save()
    return category


@app.put("/items/{item_id}")
async def get_item(item_id: int, item: Item):
    item_db = await Item.objects.get(pk=item_id)
    return await item_db.update(**item.model_dump())


@app.delete("/items/{item_id}")
async def delete_item(item_id: int, item: Item = None):
    if item:
        return {"deleted_rows": await item.delete()}
    item_db = await Item.objects.get(pk=item_id)
    return {"deleted_rows": await item_db.delete()}

!!!note 请注意，在 fastapi 为您初始化后，像 save() 这样的 ormar 模型方法如何直接可用。

!!!note 请注意，您可以直接返回一个模型（或模型列表） - fastapi 将为您将其 jsonize

测试应用程序

运行fastapi

如果你想运行这个脚本并使用 fastapi swagger 首先安装 uvicorn

pip 安装 uvicorn

并启动 fastapi。

uvicorn <filename_without_extension>:app --reload

现在您可以导航到浏览器（默认情况下 fastapi 地址为 127.0.0.1:8000/docs）并使用 api。

!!!info 您可以在 fastapi 文档中阅读有关运行 fastapi 的更多信息。

使用 pytest 进行测试

这里有一个示例测试，将证明一切都按预期工作。

请务必先创建表。如果您使用 pytest，则可以使用固定装置。

@pytest.fixture(autouse=True, scope="module")
def create_test_database():
    engine = sqlalchemy.create_engine(DATABASE_URL)
    metadata.create_all(engine)
    yield
    metadata.drop_all(engine)



# here is a sample test to check the working of the ormar with fastapi

from starlette.testclient import TestClient

def test_all_endpoints():
    # note that TestClient is only sync, don't use asyns here
    client = TestClient(app)
    # note that you need to connect to database manually
    # or use client as contextmanager during tests
    with client as client:
        response = client.post("/categories/", json={"name": "test cat"})
        category = response.json()
        response = client.post(
            "/items/", json={"name": "test", "id": 1, "category": category}
        )
        item = Item(**response.json())
        assert item.pk is not None

        response = client.get("/items/")
        items = [Item(**item) for item in response.json()]
        assert items[0] == item

        item.name = "New name"
        response = client.put(f"/items/{item.pk}", json=item.model_dump())
        assert response.json() == item.model_dump()

        response = client.get("/items/")
        items = [Item(**item) for item in response.json()]
        assert items[0].name == "New name"

        response = client.delete(f"/items/{item.pk}", json=item.model_dump())
        assert response.json().get("deleted_rows", "__UNDEFINED__") != "__UNDEFINED__"
        response = client.get("/items/")
        items = response.json()
        assert len(items) == 0

!!!提示如果您想查看更多测试用例以及如何测试 ormar/fastapi，请参阅 github 存储库中的测试目录

!!!info 您可以在 fastapi 文档中阅读有关测试 fastapi 的更多信息。