Quick API

Using ormar in conjunction with fastapi is easy.

All you need to do other than connect to the database at startup is replace the pydantic model with the ormar model.

Here you can find a very simple sample application code.

!!! Warning This example assumes that you have already created a database. If this is not the case, visit the Database Initialization section.

!!!tip The following examples (all parts) should be placed in one file.

It's divided into subsections for clarity.

!!!Note If you want to learn more about how to use the ormar model in fastapi requests and responses, check out the responses and requests documentation.

quick start

!!!Note that you can find the complete quick start script under Examples in the github repository.

Import and initialization

Use the FastAPI lifecycle to define startup and shutdown programs and use them in your applications.

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

Model definition

Define the ormar model with the appropriate fields.

These models will be used instead of pydantic models.

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)

!!!Tip You can read more about defining models in the Models section.

Fastapi endpoint definition

Define the endpoints you want, noting how the ormar model is used as response_model and request parameters.

@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 how ormar model methods like save() are directly available after fastapi initializes for you.

!!!Note that you can return directly to a model (or list of models)- fastapi will jsonize it for you

test application

Run fastapi

If you want to run this script and use fastapi swagger first install uvicorn

pip install uvicorn

And start fastapi.

uvicorn <filename_without_extension>:app --reload

You can now navigate to your browser (fastapi address is 127.0.0.1:8000/docs by default) and use the api.

!!!info You can read more about running fastapi in the fastapi documentation.

Testing with pytest

Here is an example test that will prove everything works as expected.

Be sure to create tables first. If you use pytest, you can use fixtures.

@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

!!!Tip If you want to see more test cases and how to test ormar/fastapi, see the test directory in the github repository

!!!info You can read more about testing fastapi in the fastapi documentation.