HowtoForge

So installierst du Fast API mit MongoDB unter Ubuntu 24.04

von ·

FastAPI ist ein auf Python basierendes Web-Framework zur Erstellung von API-Diensten. Es ist ein modernes, schnelles und leistungsstarkes Framework, das asynchrone Operationen unterstützt.

In diesem Tutorial lernst du, wie du FastAPI mit MongoDB unter Ubuntu 24.04 installierst. Außerdem lernst du, wie du mit FastAPI und der Datenbank MongoDB deine erste API mit CRUD-Operationen erstellst.

Voraussetzungen

Um mit diesem Leitfaden zu beginnen, musst du folgende Voraussetzungen erfüllen:

  • Ein Ubuntu 24.04-System
  • Einen Nicht-Root-Benutzer mit Administrator-Rechten

MongoDB installieren

Bevor wir ein neues FastAPI-Projekt erstellen, müssen wir den MongoDB-Server auf unserem System installieren.

Führe zunächst den unten stehenden Befehl aus, um deinen Paketindex zu aktualisieren und „gnupg“ und „curl“ auf deinem System zu installieren.

sudo apt update && sudo apt install gnupg curl -y

Führe den unten stehenden Befehl aus, um den GPG-Schlüssel für den MongoDB-Server hinzuzufügen.

curl -fsSL https://www.mongodb.org/static/pgp/server-8.0.asc | \
sudo gpg -o /usr/share/keyrings/mongodb-server-8.0.gpg \
--dearmor

Füge das MongoDB-Repository mit dem unten stehenden Befehl hinzu.

echo "deb [ arch=amd64,arm64 signed-by=/usr/share/keyrings/mongodb-server-8.0.gpg ] https://repo.mongodb.org/apt/ubuntu noble/mongodb-org/8.0 multiverse" | sudo tee /etc/apt/sources.list.d/mongodb-org-8.0.list

Nachdem das Repository hinzugefügt wurde, führe den folgenden Befehl aus, um deinen Paketindex zu aktualisieren und den MongoDB-Server zu installieren. Gib „Y“ ein, um die Installation zu bestätigen.

sudo apt update && sudo apt install mongodb-org

Repo hinzufügen

Wenn die Installation abgeschlossen ist, starte und aktiviere den MongoDB-Dienst „mongod“ und überprüfe dann den Status deines MongoDB-Dienstes, um sicherzustellen, dass er läuft.

sudo systemctl enable --now mongod
sudo systemctl status mongod

Du kannst unten sehen, dass der MongoDB-Server läuft.

starten, aktivieren und verifizieren

Außerdem kannst du dich mit dem unten stehenden Befehl „mongosh“ beim MongoDB-Server anmelden. Zum Beenden drückst du die Tastenkombination Strg+d.

mongosh

Einrichten von Python und der virtuellen Umgebung

Nachdem du MongoDB installiert hast, installierst du die Python-Pakete und richtest das Projektverzeichnis und die virtuelle Umgebung ein.

Installiere Python, Pip und die Venv-Module mit dem folgenden Befehl. Gib „Y“ ein, um die Installation zu bestätigen.

sudo apt install python3 python3-pip python3-venv

python Pakete installieren

Sobald die Installation abgeschlossen ist, melde dich mit deinem Benutzer an.

su - username

Erstelle nun ein neues Verzeichnis„~/app“ und wechsle in dieses Verzeichnis. In diesem Verzeichnis wirst du dein FastAPI-Projekt speichern.

mkdir -p ~/app; cd ~/app

Führe den folgenden Befehl aus, um eine neue virtuelle Umgebung„venv“ zu erstellen und sie zu aktivieren. Deine Shell-Eingabeaufforderung sieht dann so aus: „(venv) user@hostname„.

python3 -m venv .venv
source .venv/bin/activate

venv einrichten

Von nun an muss sich deine Arbeitsumgebung in der virtuellen Umgebung „venv“ befinden. Mit dem folgenden Befehl kannst du dich aus der virtuellen Umgebung „venv“ abmelden.

deactivate

FastAPI-Projekt erstellen

Nachdem du nun deine virtuelle Python-Umgebung erstellt und aktiviert hast, installieren wir FastAPI und erstellen die Projektstruktur.

Führe den Befehl „pip3“ aus, um die Pakete „fastapi“ und „uvicorn“ zu installieren.

pip3 install fastapi uvicorn
  • Fastapi“ ist das FastAPI-Webframework zur Erstellung von APIs in Python.
  • uvicorn“ ist die Implementierung des ASGI (Asynchronous Server Gateway Interface) Webservers in Python.

Nach der Installation erstellst du neue Dateien und Verzeichnisse mit dem folgenden Befehl.

mkdir -p server/{models,routes}
touch main.py server/{app.py,database.py} server/models/itemModels.py server/routes/item.py

Im Folgenden siehst du die Struktur unseres FastAPI-Projekts.

Strukturen

server/app.py

Jetzt, wo dein Projekt fertig ist, können wir die Datei „server/app.py“ ändern, die die Hauptanwendung deines FastAPI-Projekts ist.

Öffne die Datei„app.py“ mit deinem Texteditor und kopiere das folgende Skript.

from fastapi import FastAPI
app = FastAPI()

@app.get("/")
async def root():
return {"message": "Hello FastAPI!"}
  • Importiere das FastAPI-Modul in dein Projekt und binde es an die Variable „app“.
  • Erstelle eine neue „root“-Funktion, die das „Hello FastAPI!“ zurückgibt.
  • Die „root“-Funktion antwortet auf die GET-Methode der Root-URL
  • async“ markiert deine Funktion als asynchrone Funktion und kann beim Aufruf „await“ in ihrem Körper verwenden

main.py

In diesem Abschnitt ändern wir die Datei „main.py“, die verwendet wird, um dein FastAPI-Projekt über „uvicorn“ (ASGI-Webserver in Python) auszuführen.

Öffne und ändere das Skript„main.py“ und füge den folgenden Code ein.

import uvicorn
if __name__ == "__main__":
uvicorn.run("server.app:app", host="0.0.0.0", port=8080, reload=True)
  • Importiere das „uvicorn“-Modul
  • Wenn das Skript „main.py“ ausgeführt wird, lädt es das Modul „app“ oder FastAPI in der Datei „server/app.py“.
  • FastAPI wird auf „0.0.0.0“ mit Port „8080“ ausgeführt
  • Aktiviere das automatische Neuladen, wenn sich der Code ändert, mit ‚reload=True‘.

Starte dein FastAPI-Projekt

Jetzt, wo dein Projekt fertig ist, lass uns dein erstes FastAPI-Projekt ausführen.

Führe das Skript „main.py“ wie folgt aus und deine FastAPI wird auf deinem System ausgeführt.

python3 main.py

Fastapi ausführen

Öffne nun deinen Webbrowser und rufe http://SERVERIP:8080/ auf. Wenn die Installation erfolgreich war, siehst du die Meldung „Hello FastAPI!“. Du kannst auch über „curl“ im Terminal darauf zugreifen.

Und schließlich kannst du auf die API-Dokumentation auf http://SERVERIP:8080/docs> zugreifen, die von der Swagger UI bereitgestellt wird.

fastapi docs

Verbinden von FastAPI mit MongoDB

In dieser Anleitung erstellst du eine einfache API mit FastAPI und MongoDB. Deine API sollte in der Lage sein, CRUD mit dem MongoDB-Datenbankserver durchzuführen. In diesem Schritt wirst du dein Projekt mit dem MongoDB-Server verbinden.

Führe zunächst den Befehl „pip3“ aus, um den MongoDB-Treiber„motor“ in deinem Projekt zu installieren. motor“ bietet einen nicht-blockierenden und Coroutine-basierten API-Zugriff auf den MongoDB-Server.

pip3 install motor

server/database.py

Nachdem das„motor„-Modul installiert ist, musst du das Skript„server/database.py“ anpassen.

Öffne die Datei „server/database.py“ mit deinem Texteditor und gib das folgende Skript ein. Damit wird die Verbindung zum MongoDB-Server über das Modul „motor“ hergestellt.

from motor.motor_asyncio import AsyncIOMotorClient

MONGODB_HOST = „mongodb://localhost:27017“

connection = AsyncIOMotorClient(MONGODB_HOST)

database = connection.items
item_collection = database.get_collection(„item_collection“)

  • Importiere den „AsyncIOMotorClient“ aus dem Modul „motor.motor_asyncio“.
  • Erstelle eine neue Konstante „MONGODB_HOST“ und verweise auf den MongoDB-Server „mongodb://localhost:27017“.
  • Verbinde dich mit dem MongoDB-Server über die Variable ‚connection‘.
  • Verbinde dich mit der Datenbank „items“ über die Variable „database“.
  • Auf Sammlungen in der Datenbank über die Variable ‚item_collection‘ zugreifen

Datenbankmodell mit pydantic erstellen

In diesem Abschnitt entwirfst du deine Daten mit dem Modul„pydantic„, das die Modellierung für unsere MongoDB-Datenbank ermöglicht.

Installiere das Modul „pydantic“ mit dem unten stehenden „pip3“-Befehl. Das Modul„pydantic“ ist eine Datenvalidierungsbibliothek, mit der du ein Datenbankschema über ein Modell erstellen kannst.

pip3 install pydantic

Öffne nun die Datei„server/models/itemModels.py“ mit deinem Texteditor und kopiere das folgende Skript.

from pydantic import BaseModel, Field
from typing import Optional
class Item(BaseModel):
name: str
category: str
stocks: int
price: int = Field(gt=0)

class Config:
json_schema_extra = {
"example": {
"name": "Company Smart Watch",
"category": "smartwatch",
"stocks": 10,
"price": 1000,
}
}

class ItemUpdate(BaseModel):
name: Optional[str] = None
category: Optional[str] = None
stocks: Optional[int] = None
price: Optional[int] = None

class Config:
json_schema_extra = {
"example": {
"name": "New Smart watch",
"category": "new-smartwatch",
"stocks": 5,
"price": 500,
}
}
  • Importiere die Module ‚BaseModel‘ und ‚Field‘ aus ‚pydantic‘.
  • Importiere das Modul „Optional“ von „typing“.
  • Erstelle das folgende „Item“-Datenbankschema für die FastAPI:
    • ‚Name‘ und ‚Kategorie‘ mit Typ String
    • ‚Bestand‘ und ‚Preis‘ mit der Kategorie Integer
    • Preis‘ muss größer als 0 sein
  • Erweitern Sie das Datenmodell über die Klasse „Config“, indem Sie ein Beispiel für Daten bereitstellen, die der Benutzer in die Anfrage aufnehmen kann
  • Erstelle das folgende „ItemUpdate“-Schema mit jedem Feld als optional.

Hinzufügen von CRUD-Operationen

Jetzt hast du eine Verbindung zum MongoDB-Server hergestellt und mit „pydantic“ ein Datenbankschema erstellt. Jetzt geht es darum, die CRUD-Operationen mit FastAPI und MongoDB zu erstellen.

Öffne das Skript„server/database.py“ erneut und erstelle eine neue Funktion„item_helper“ mit dem Typ„dict„. Rufe diese Funktion auf, um detaillierte Daten zu einem Artikel zu erhalten.

def item_helper(item) -> dict:
 return {
 "id": str(item["_id"]),
 "name": item["name"],
 "category": item["category"],
 "stocks": item["stocks"],
 "price": item["price"],
 }

Füge die folgende Zeile hinzu, um das Modul„ObjectId“ von „bson.objectid“ zu importieren. Die‚ObjectId‚ ist ein Teil des Datentyps in BSON, den MongoDB für die Datenspeicherung und Darstellung von JSON-Daten verwendet.

from bson.objectid import ObjectId

Ein Element erstellen

Als Erstes erstellst du eine asynchrone Funktion, mit der du der MongoDB-Datenbank neue Daten hinzufügen kannst.

Erstelle eine neue Funktion „add_item“ wie die folgende:

# Add a new item
async def add_item(item_details: dict) -> dict :
 item = await item_collection.insert_one(item_details)
 new_item = await item_collection.find_one({"_id": item.inserted_id})
 return item_helper(new_item)
  • Die Funktion „add_item“ akzeptiert Wörterbuchdaten zu deinen Artikeldetails
  • Über die „item_collection“ greifst du auf Dokumente in der Datenbank zu und führst die Abfrage „insert_one“ aus, um einen neuen Artikel hinzuzufügen.
  • Der neue Artikel wird bei erfolgreicher Ausführung ausgedruckt

Alle Artikel abrufen

Als Zweites erstellst du eine neue Funktion „get_items“, mit der du alle verfügbaren Einträge in deiner Datenbank abrufen kannst.

Erstelle eine neue Funktion„get_items„, um alle Einträge aus der MongoDB abzurufen.

# retrieve all items
async def get_items():
 items = []
 async for item in item_collection.find():
 items.append(item_helper(item))
 return items
  • Du erstellst eine leere Liste von „items“.
  • Mit der Abfrage „find()“ durchsuchst du alle Daten und durchläufst sie in einer Schleife
  • Jedes Element wird der Liste „items“ mit der Methode „append“ hinzugefügt.
  • Nachdem die Schleife abgeschlossen ist, werden die Einträge angezeigt

Bestimmten Eintrag anhand der ID abrufen

Als Nächstes erstellst du eine neue Funktion, mit der du Daten für den Selektor „id“ abrufen kannst. So erhältst du detaillierte Daten zu bestimmten Artikeln.

Erstelle eine neue Funktion„get_item„, um Daten zu bestimmten Artikeln abzurufen. Die Funktion „get_item“ akzeptiert den String „id“ als Selektor und gibt ein Wörterbuch zurück.

# retrieve specific item
async def get_item(id: str) -> dict:
 item = await item_collection.find_one({"_id": ObjectId(id)})
 if item:
 return item_helper(item)
 return "Item Not Found."
  • Die Abfrage ‚find_one()‘ wird verwendet, um Daten auf der Grundlage der ‚id‘ abzurufen.
  • Wenn ein Element gefunden wird, werden die Details im Wörterbuchformat „item_helper“ angezeigt.
  • Wenn der Artikel nicht verfügbar ist, lautet die Antwort „Artikel nicht gefunden“.

Einen Artikel aktualisieren

Jetzt erstellst du eine neue Funktion, um einen Artikel zu aktualisieren. Du kannst deinen Artikel auch teilweise über die API aktualisieren.

Definiere die neue Funktion„change_item“ wie in den folgenden Codes:

# update item
async def change_item(id: str, data: dict):
 if len(data) < 1:
 return "Please input your data"
 find_item = await item_collection.find_one({"_id": ObjectId(id)})
 if find_item:
 item_update = await item_collection.update_one({"_id": ObjectId(id)}, {"$set": data})
 if item_update:
 return True
 return False
  • Die Funktion „change_item“ benötigt zwei Argumente, die „id“ des Zielartikels und „data“ als neue Daten
  • Wenn data leer oder ‚< 1‘ ist, wird der Vorgang abgebrochen
  • Diese Funktion findet einen Eintrag anhand des Selektors „id“.
  • Wenn die „id“ gefunden wird, wird die „item_update“ ausgeführt
  • Ist das ‚item_update‘ erfolgreich, wird ‚True‘ zurückgegeben, andernfalls ‚False‘.

Ein Element löschen

Zuletzt erstellst du die Funktion „delete_item“, um Elemente über einen bestimmten Selektor zu löschen.

Um einen Eintrag zu löschen, erstellen wir die Funktion„delete_item“ und verwenden den Selektor „id“ wie folgt:

# delete an item
async def delete_item(id: str):
 item = await item_collection.find_one({"_id": ObjectId(id)})
 if item:
 await item_collection.delete_one({"_id": ObjectId(id)})
 return(f'Item {id} deleted.')
 return "Item Not Found."
  • Die Abfrage „find_one()“ sucht anhand der angegebenen „id“ nach einem Eintrag.
  • Wenn ein Element gefunden wird, wird es mit der Abfrage „delete_one()“ gelöscht und die „Item id deleted“ zurückgegeben.
  • Wenn das Objekt nicht verfügbar ist, wird die Meldung „Objekt nicht gefunden“ angezeigt.

Hinzufügen von Routes für CRUD-Operationen

An dieser Stelle haben wir asynchrone Funktionen für CRUD-Operationen mit FastAPI erstellt. Jetzt müssen wir die Route oder den Endpunkt für jede Operation erstellen.

Bearbeite die Datei„server/routes/item.py“ mit deinem bevorzugten Editor.

server/routes/item.py

Füge zunächst die Module„APIRouter“ und„Body“ aus der „fastapi“ hinzu. Dann fügst du das Modul„jsonable_encode“ aus der Datei „fastapi.encoders“ hinzu.

from fastapi import APIRouter, Body
from fastapi.encoders import jsonable_encoder

Importiere jede Funktion aus der Datei „database.py“.

from server.database import (
 add_item,
 get_items,
 get_item,
 change_item,
 delete_item,
)

Importiere die Modelle „Item“ und „ItemUpdate“ aus der Datei „itemModels.py“.

from server.models.itemModels import (
 Item,
 ItemUpdate,
)

Rufe die Klasse „APIRouter“ über die Variable „router“ auf.

router = APIRouter()

Route für das Hinzufügen eines neuen Artikels

Fügen wir nun eine Route für das Hinzufügen neuer Artikel hinzu. In diesem Beispiel kannst du einen neuen Artikel per POST über die URL„/item“ hinzufügen.

Füge die folgenden Zeilen hinzu, um Routen für das Hinzufügen neuer Artikel einzurichten. Jeder POST an die Stamm-URL des Artikels wird als Einfügen neuer Daten behandelt.

# add new item operation
@router.post("/")
async def add_item_data(item: Item = Body(...)):
 item = jsonable_encoder(item)
 new_item = await add_item(item)
 return new_item
  • Die Funktion „add_item_data“ akzeptiert das „Item“-Schema, das Teil des „Body“ deiner Anfrage sein wird
  • Dein Eintrag wird über den „jsonable_encoder“ in ein Wörterbuchformat umgewandelt.
  • Füge deine Wörterbuchdaten über die Funktion „add_item“ (siehe database.py) in die Variable „new_item“ ein
  • Gib deine eingefügten Daten „new_item“ als Antwort zurück

Route zum Abrufen aller Einträge

Füge das folgende Skript hinzu, um eine Route zum Abrufen von Daten einzurichten. Jede GET-Anfrage an die Stamm-URL des Eintrags wird alle Daten abrufen.

# get all available items
@router.get("/")
async def get_item_data():
 items = await get_items()
 if items:
 return items
 return "No available item."
  • Erstelle die Funktion „get_item_data“, die die Funktion „get_item“ aus der Datei „database.py“ ausführt.
  • Wenn Artikel vorhanden sind, erhältst du eine Liste aller deiner Artikel.
  • Wenn kein Artikel vorhanden ist, erhältst du die Antwort „Kein Artikel verfügbar“.

Route zum Abrufen eines bestimmten Eintrags

Um Details zu einem bestimmten Artikel zu erhalten, verwenden wir die „id“ als Selektor. Jede GET-Anfrage an „/id“ gibt einen detaillierten Artikel mit der angeforderten „id“ zurück.

# Show details of the item via the id
@router.get("/{id}")
async def get_item_details(id):
 item_details = await get_item(id)
 if item_details:
 return item_details
 return "Item not found."
  • Die Funktion „get_item_details“ wird erstellt und übergibt die „id“ aus der URL
  • Die Funktion „get_item“ (siehe database.py) wird aufgerufen und übergibt ebenfalls die „id“ als Argument
  • Wenn der Eintrag gefunden wird, werden die Details des Eintrags angezeigt
  • Wenn es keinen Artikel mit der entsprechenden „id“ gibt, wird „Artikel nicht gefunden“ angezeigt.

Route zum Aktualisieren eines Artikels

Kopiere den folgenden Code, um die Route für den aktualisierten Artikel festzulegen:

# Update Item
@router.put("/{id}")
async def update_item(id: str, data: ItemUpdate = Body(...)):
 data = {k: v for k, v in data.dict().items() if v is not None}
 updated_item = await change_item(id, data)
 if updated_item:
 return{f'Success: item {id} updated.'}
 return "Error"
  • Die Funktion „update_item“ nimmt zwei Argumente entgegen: „id“ als Selektor und „data“, das auf dem Modell „ItemUpdate“ basiert.
  • Die Daten werden in der Variable „data“ überprüft.
  • Die Funktion „updated_item“ führt die Funktion „change_item“ aus der Datei „database.py“ aus.
  • Wenn die Aktualisierung erfolgreich war, siehst du die Ausgabe „Success“.

Route zum Löschen eines Eintrags

Füge die folgenden Zeilen ein, um die Funktion „remove_item“ zum Löschen von Einträgen zu erstellen. Jede DELETE-Operation für eine bestimmte „/id“ entfernt den Eintrag, der mit der „id“ übereinstimmt.

# delete item via id
@router.delete("/{id}")
async def remove_item(id):
 item_to_delete = await delete_item(id)
 if item_to_delete:
 return item_to_delete
 return{f'Item {id} Not Available.'}
  • Die Funktion „remove_item“ führt „delete_item“ aus und übergibt den Selektor „id“.
  • Der Löschvorgang wird gespeichert und über die Variable „item_to_delete“ ausgeführt.
  • Wenn ein Artikel nicht verfügbar ist, erhältst du die Rückmeldung ‚Item id Not Available‘.

server/app.py

Nachdem du die Datei „server/routes/item.py“ fertiggestellt hast, fügen wir sie nun in die Datei „server/app.py“ ein.

Öffne die Datei„app.py“ mit deinem Texteditor.

Importiere den „Router“ aus der Datei„server/routes/item.py“ als„ItemRouter„.

from server.routes.item import router as ItemRouter

Füge den „ItemRouter“ mit dem Standardpräfix„/item“ ein. Die CRUD-Operationen werden über die URL„/item“ abgewickelt.

app.include_router(ItemRouter, tags=["Item"], prefix="/item")

Jetzt ist dein CRUD-Endpunkt unter folgendem Link verfügbar:

  • Neuen Artikel hinzufügen: POST an‚/item‚
  • Alle Artikel abrufen: GET zu‚/item‚
  • Einen bestimmten Artikel abrufen: GET zu ‚/item/id‘. Die „id“ wird von MongoDB generiert.
  • Artikel aktualisieren: PUT zu ‚/item/id‘
  • Element löschen: DELETE zu ‚/item/id‘

CRUD-Operationen testen

Stelle zunächst sicher, dass dein FastAPI-Projekt läuft, oder führe das Skript „main.py“ wie folgt aus:

python3 main.py

Navigiere durch die http://SERVERIP:8080/docs und du wirst jede Route sehen, die du erstellt hast.

CRUD-Vorgänge

Im Folgenden siehst du ein Beispiel für das Hinzufügen eines neuen Artikels.

Neue Position hinzufügen

Rufe alle verfügbaren Artikel über die API ab.

Artikel abrufen alle

Bestimmte Artikel über den Selektor „id“ abrufen.

einen bestimmten Artikel abrufen

Aktualisiere Teildaten von bestimmten Artikeln.

Artikel aktualisieren

Unten siehst du die aktualisierten Daten.

aktualisierte Daten

Unten siehst du den Löschvorgang über den „id“-Selektor.

Daten löschen

Fazit

Herzlichen Glückwunsch! Du hast die Installation von FastAPI mit MongoDB auf Ubuntu 24.04 abgeschlossen. Du hast außerdem gelernt, wie du FastAPI mit dem Modul „motor“ mit MongoDB verbindest, Datenmodelle mit „pydantic“ erstellst, CRUD-Operationen mit FastAPI erstellst und Endpunkte für deine API erstellst.

Das könnte dich auch interessieren …