Aller au contenu

API rapide pour

Il est très simple d'utiliser Ormar en combinaison avec Fastapi.

Tout ce que vous devez faire, en plus de vous connecter à la base de données au démarrage, est de remplacer le modèle pydantique par le modèle ormar.

Ici, vous pouvez trouver un exemple très simple de code d'application.

!!! AVERTISSEMENT Cet exemple suppose que vous avez créé une base de données. si ce n'est pas le cas, visitez la section Initialisation de base de données.

!!!Tip Les exemples suivants (toutes les sections) devraient être placés dans un seul fichier.

1
Elle est divisée en sous-sections pour la clarté.

!!!Note Si vous souhaitez en savoir plus sur l'utilisation du modèle ormar dans les requêtes et réponses fastapi, consultez la documentation Réponses et demandes.

Commencer rapidement

!!!Notez que vous pouvez trouver le script de démarrage rapide complet sous l'exemple du dépôt github.

Import et initialisation

Les programmes de démarrage et d'arrêt sont définis à l'aide du cycle de vie FastAPI et utilisés dans les applications.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
à partir de typing import List, Optional

Importation de databases
Import scalalchemy
from fastapi import FastAPI

Import par Ormar

de contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    Si ce n'est pas config.database.is_connected:
        await config.database.connect()

    Yield est un

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


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

Info

Définition du modèle

Définissez le modèle ormar avec les champs appropriés.

Ces modèles seront utilisés à la place du modèle pydantique.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
base_ormar_config = OrmarConfig(
    Metadata = métadonnée
    database = database
)


Catégorie de classe (modèle ormar):
    ormar_config = base_ormar_config.copy(tablename="catégories")

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


Item de classe (modèle ormar):
    ormar_config = base_ormar_config.copy()

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

!!!Conseil Vous pouvez en savoir plus sur la définition d'un modèle dans la section Modèles.

Définition du Endpoint Fastapi

Définissez les points de terminaison souhaités, en notant comment le modèle ormar est utilisé comme paramètre de réponse_model et de requête.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@app.get("/items/", response_model=List[Item])
def get_items():
    items = await Item.objects.select_related("catégorie").all()
    Retour aux items


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


@ app.post ("/categories/", response_model=Category)
async def create_category(category: Category):
    await category.save ()
    Retour à la catégorie


@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):
    Si l'article:
        Retour sur {"deleted_rows": await item.delete()}
    item_db = await Item.objects.get(pk=item_id)
    Retour à {"deleted_rows": await item_db.delete()}

!!!Remarquez comment les méthodes du modèle ormar telles que save() sont directement disponibles après l'initialisation de fastapi pour vous.

!!!Notez que vous pouvez retourner directement un modèle (ou une liste de modèles)- fastapi le jsonize pour vous

Tester l'app

Exécuter Fastapi

Si vous souhaitez exécuter ce script et utiliser fastapi swagger, installez uvicorn en premier.

Pip installé uvicorn

Et commencer à fastapi.

uvicorn <filename_without_extension>:app --reload

Vous pouvez maintenant naviguer vers votre navigateur (l'adresse fastapi par défaut est 127.0.0.1: 8000/docs) et utiliser api.

!!!info Vous pouvez en savoir plus sur l'exécution de fastapi dans la documentation de fastapi.

Tester avec pytest

Voici un exemple de test qui prouvera que tout fonctionne comme prévu.

Assurez-vous de créer d'abord la table. si vous utilisez pytest, vous pouvez utiliser des appareils.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@pytest.fixture(autouse=True, scope="module")
def create_test_database():
    engine = sqlalchemy.create_engine(DATABASE_URL)
    metadata.create_all(engine)
    Yield, le
    drop_all(engine)



#Voici un exemple de test pour vérifier le travail de l'ordre avec fastapi

depuis starlette.testclient import TestClient

def test_all_endpoints():
    # note que le client de test est uniquement synchronisé, ne l'utilise pas ici
    client = TestClient(app)
    # note que vous devez vous connecter à la base de données manuellement
    #ou utiliser le client comme gestionnaire de contexte pendant les tests
    Avec un client comme client:
        response = client.post ("/categories/", json={"name": "test cat"})
        catégorie = response.json()
        response = client.post (
            /items/, json={"name": "test", "id": 1, "category": category}
        )
        item = Item(**response.json()))
        assert item.pk n'est pas None

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

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

        réponse = client.get("/items/")
        items = [Item(**item) for item in response.json()]
        assert items[0].name == "nouveau nom"

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

!!!Conseil Si vous souhaitez voir plus de cas de test et comment tester ormar/fastapi, consultez le catalogue de tests dans le référentiel github.

!!!info Vous pouvez en savoir plus sur le test de fastapi dans la documentation de fastapi.