FastAPI-Kernkonzepte
FastAPI ist ein modernes, schnelles (hochleistungsfähiges) Python-Webframework zum Erstellen von APIs. Es basiert auf standardmäßigen Python-Type-Hints und verwendet Starlette und Pydantic.
Hauptfunktionen
- Hohe Leistung: Vergleichbar mit NodeJS und Go, eines der schnellsten Python-Frameworks
- Schnelle Entwicklung: Entwicklungsgeschwindigkeit um etwa 200%-300% erhöht
- Fehler reduzieren: Menschliche Fehler um etwa 40% reduzieren
- Intuitiv: Leistungsstarke Editorunterstützung, Autovervollständigung überall
- Einfach: Einfach zu erlernen und zu verwenden, reduziert die Zeit für das Lesen von Dokumentation
ASGI und asynchrone Programmierung
Was ist ASGI?
ASGI (Asynchronous Server Gateway Interface) ist die Standardschnittstelle zwischen asynchronen Python-Webservern und Anwendungen.
WSGI vs ASGI Vergleich:
Beispiel
# WSGI-Anwendung (synchron) - traditioneller Flask-Stil
def wsgi_app(environ, start_response):
status = '200 OK'
response_headers = [('Content-type', 'text/plain')]
start_response(status, response_headers)
return [b'Hello World']
# ASGI-Anwendung (async) – FastAPI-Stil
async def asgi_app(scope, receive, send):
await send({
'type': 'http.response.start',
'status': 200,
'headers': [(b'content-type', btext/plain')],
})
await send({
'type': 'http.response.body',
'body': b'Hello World',
})
def wsgi_app(environ, start_response):
status = '200 OK'
response_headers = [('Content-type', 'text/plain')]
start_response(status, response_headers)
return [b'Hello World']
# ASGI-Anwendung (async) – FastAPI-Stil
async def asgi_app(scope, receive, send):
await send({
'type': 'http.response.start',
'status': 200,
'headers': [(b'content-type', btext/plain')],
})
await send({
'type': 'http.response.body',
'body': b'Hello World',
})
Kernkonzepte der asynchronen Ausführung
Synchrone vs. asynchrone Ausführung:
Beispiel
import asyncio
import time
import httpx
# Synchroner Modus - blockierende Ausführung
def sync_fetch_data():
start_time = time.time()
# Drei Netzwerkanfragen simulieren
time.sleep(1) # Erste Anfrage
time.sleep(1) # Zweite Anfrage
time.sleep(1) # Dritte Anfrage
print(f"Verstrichene Zeit für synchrone Ausführung: {time.time() - start_time:.2f} Sekunden") # etwa 3 Sekunden
# Asynchroner Modus - nebenläufige Ausführung
async def async_fetch_data():
start_time = time.time()
# Drei Anfragen nebenläufig ausführen
await asyncio.gather(
asyncio.sleep(1), # Erste Anfrage
asyncio.sleep(1), # Zweite Anfrage
asyncio.sleep(1), # Dritte Anfrage
)
print(fAsynchrone Ausführungszeit: {time.time() - start_time:.2f} Sekunden) # etwa 1 Sekunde
# Beispiel ausführen
sync_fetch_data() # Ausgabe: Synchrone Ausführungszeit: 3.00 Sekunden
asyncio.run(async_fetch_data()) # Ausgabe: Asynchrone Ausführungszeit: 1.00 Sekunden
import time
import httpx
# Synchroner Modus - blockierende Ausführung
def sync_fetch_data():
start_time = time.time()
# Drei Netzwerkanfragen simulieren
time.sleep(1) # Erste Anfrage
time.sleep(1) # Zweite Anfrage
time.sleep(1) # Dritte Anfrage
print(f"Verstrichene Zeit für synchrone Ausführung: {time.time() - start_time:.2f} Sekunden") # etwa 3 Sekunden
# Asynchroner Modus - nebenläufige Ausführung
async def async_fetch_data():
start_time = time.time()
# Drei Anfragen nebenläufig ausführen
await asyncio.gather(
asyncio.sleep(1), # Erste Anfrage
asyncio.sleep(1), # Zweite Anfrage
asyncio.sleep(1), # Dritte Anfrage
)
print(fAsynchrone Ausführungszeit: {time.time() - start_time:.2f} Sekunden) # etwa 1 Sekunde
# Beispiel ausführen
sync_fetch_data() # Ausgabe: Synchrone Ausführungszeit: 3.00 Sekunden
asyncio.run(async_fetch_data()) # Ausgabe: Asynchrone Ausführungszeit: 1.00 Sekunden
Asynchrone Pfadoperationen in FastAPI:
Beispiel
from fastapi import FastAPI
import asyncio
app = FastAPI()
# Synchrone Pfadoperation
@app.get("/sync")
def sync_endpoint():
# Synchrone Operationen blockieren die gesamte Anwendung
time.sleep(2)
return {"message": "Synchrone Antwort"}
# Asynchrone Pfadoperation (empfohlen)
@app.get("/async")
async def async_endpoint():
# Asynchrone Vorgänge blockieren andere Anfragen nicht
await asyncio.sleep(2)
return {"message": "Asynchrone Antwort"}
# Beispiel für eine asynchrone Datenbankoperation
@app.get("/users/{user_id}")
async def get_user(user_id: int):
# Asynchrone Datenbankabfrage
user = await database.fetch_one(
"SELECT * FROM users WHERE id = :user_id",
{"user_id": user_id}
)
return user
import asyncio
app = FastAPI()
# Synchrone Pfadoperation
@app.get("/sync")
def sync_endpoint():
# Synchrone Operationen blockieren die gesamte Anwendung
time.sleep(2)
return {"message": "Synchrone Antwort"}
# Asynchrone Pfadoperation (empfohlen)
@app.get("/async")
async def async_endpoint():
# Asynchrone Vorgänge blockieren andere Anfragen nicht
await asyncio.sleep(2)
return {"message": "Asynchrone Antwort"}
# Beispiel für eine asynchrone Datenbankoperation
@app.get("/users/{user_id}")
async def get_user(user_id: int):
# Asynchrone Datenbankabfrage
user = await database.fetch_one(
"SELECT * FROM users WHERE id = :user_id",
{"user_id": user_id}
)
return user
Wann sollte man async verwenden?
Geeignete Szenarien für asynchrone Operationen:
- Datenbankoperation
- Netzwerkanfragen (API-Aufrufe)
- Datei-I/O-Operationen
- Operationen mit langen Wartezeiten
Szenarien, die nicht für asynchrone Operationen geeignet sind:
- CPU-intensive Berechnung
- Einfache Datenverarbeitung
- Operationen ohne I/O-Wartezeit
Beispiel
# Gute Async-Verwendung
@app.get("/weather")
async def get_weather():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.weather.com/data")
return response.json()
# Unnötiges async (kein I/O-Warten)
@app.get("/calculate")
def calculate_sync(): # Einfach synchron halten
result = sum(range(1000000))
return {"result": result}
@app.get("/weather")
async def get_weather():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.weather.com/data")
return response.json()
# Unnötiges async (kein I/O-Warten)
@app.get("/calculate")
def calculate_sync(): # Einfach synchron halten
result = sum(range(1000000))
return {"result": result}
REST-API-Designprinzipien
REST-Grundlagen
REST (Representational State Transfer) ist ein Web-API-Designstil, der Folgendes betont:
- Ressourcenorientiert: URLs repräsentieren Ressourcen
- Zustandslos: Jede Anfrage ist unabhängig
- Einheitliche Schnittstelle: Standard-HTTP-Methoden verwenden
- Mehrschichtiges System: Unterstützt Caching, Lastverteilung usw.
RESTful-URL-Design
Beispiele für gutes URL-Design:
Beispiel
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
# Ressourcensammlungen und einzelne Ressourcen
@app.get("/users") # GET /users - Benutzerliste abrufen
async def get_users():
return users_db
@app.get("/users/{user_id}") # GET /users/123 - Einen bestimmten Benutzer abrufen
async def get_user(user_id: int):
return find_user(user_id)
@app.post("/users") # POST /users - Neuen Benutzer erstellen
async def create_user(user: UserCreate):
return create_new_user(user)
@app.put("/users/{user_id}") # PUT /users/123 - Benutzer vollständig aktualisieren
async def update_user(user_id: int, user: UserUpdate):
return update_existing_user(user_id, user)
@app.patch("/users/{user_id}") # PATCH /users/123 - Benutzer teilweise aktualisieren
async def patch_user(user_id: int, user: UserPatch):
return patch_existing_user(user_id, user)
@app.delete("/users/{user_id}") # DELETE /users/123 - Benutzer löschen
async def delete_user(user_id: int):
return delete_existing_user(user_id)
# Verschachtelte Ressourcen
@app.get("/users/{user_id}/posts") # Alle Artikel des Benutzers abrufen
async def get_user_posts(user_id: int):
return get_posts_by_user(user_id)
@app.post("/users/{user_id}/posts") # Einen neuen Artikel für den Benutzer erstellen
async def create_user_post(user_id: int, post: PostCreate):
return create_post_for_user(user_id, post)
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
# Ressourcensammlungen und einzelne Ressourcen
@app.get("/users") # GET /users - Benutzerliste abrufen
async def get_users():
return users_db
@app.get("/users/{user_id}") # GET /users/123 - Einen bestimmten Benutzer abrufen
async def get_user(user_id: int):
return find_user(user_id)
@app.post("/users") # POST /users - Neuen Benutzer erstellen
async def create_user(user: UserCreate):
return create_new_user(user)
@app.put("/users/{user_id}") # PUT /users/123 - Benutzer vollständig aktualisieren
async def update_user(user_id: int, user: UserUpdate):
return update_existing_user(user_id, user)
@app.patch("/users/{user_id}") # PATCH /users/123 - Benutzer teilweise aktualisieren
async def patch_user(user_id: int, user: UserPatch):
return patch_existing_user(user_id, user)
@app.delete("/users/{user_id}") # DELETE /users/123 - Benutzer löschen
async def delete_user(user_id: int):
return delete_existing_user(user_id)
# Verschachtelte Ressourcen
@app.get("/users/{user_id}/posts") # Alle Artikel des Benutzers abrufen
async def get_user_posts(user_id: int):
return get_posts_by_user(user_id)
@app.post("/users/{user_id}/posts") # Einen neuen Artikel für den Benutzer erstellen
async def create_user_post(user_id: int, post: PostCreate):
return create_post_for_user(user_id, post)
URL-Design-Prinzipien:
好的设计 GET /api/v1/users # 获取用户列表 GET /api/v1/users/123 # 获取用户 123 POST /api/v1/users # 创建用户 GET /api/v1/users/123/orders # 获取用户 123 的订单 # 不好的设计 GET /api/v1/getAllUsers # 动词不应该在 URL 中 GET /api/v1/user/123 # 集合名应该用复数 POST /api/v1/createUser # URL 中包含动作 GET /api/v1/users-orders-123 # 关系不清晰
Hierarchische Beziehung von Ressourcen
Beispiel
# Beziehungsdesign zwischen Benutzern und Artikeln
class User(BaseModel):
id: int
name: str
email: str
class Post(BaseModel):
id: int
title: str
content: str
author_id: int
# Hauptressourcenoperation
@app.get("/users")
async def list_users() -> List[User]:
return users
@app.get("/posts")
async def list_posts() -> List[Post]:
return posts
# Zugehörige Ressourcenoperationen
@app.get("/users/{user_id}/posts")
async def get_user_posts(user_id: int) -> List[Post]:
"""Alle Artikel für einen bestimmten Benutzer abrufen"""
return [post for post in posts if post.author_id == user_id]
@app.get("/posts/{post_id}/author")
async def get_post_author(post_id: int) -> User:
"""Die Informationen zum Autor des Artikels abrufen"""
post = find_post(post_id)
if not post:
raise HTTPException(status_code=404, detail="Artikel nicht gefunden")
return find_user(post.author_id)
class User(BaseModel):
id: int
name: str
email: str
class Post(BaseModel):
id: int
title: str
content: str
author_id: int
# Hauptressourcenoperation
@app.get("/users")
async def list_users() -> List[User]:
return users
@app.get("/posts")
async def list_posts() -> List[Post]:
return posts
# Zugehörige Ressourcenoperationen
@app.get("/users/{user_id}/posts")
async def get_user_posts(user_id: int) -> List[Post]:
"""Alle Artikel für einen bestimmten Benutzer abrufen"""
return [post for post in posts if post.author_id == user_id]
@app.get("/posts/{post_id}/author")
async def get_post_author(post_id: int) -> User:
"""Die Informationen zum Autor des Artikels abrufen"""
post = find_post(post_id)
if not post:
raise HTTPException(status_code=404, detail="Artikel nicht gefunden")
return find_user(post.author_id)
HTTP-Methoden und Statuscodes
Semantik der HTTP-Methoden
Beispiel
from fastapi import FastAPI, status, HTTPException
from fastapi.responses import JSONResponse
app = FastAPI()
# GET - sicher und idempotent
@app.get("/users/{user_id}")
async def get_user(user_id: int):
"""Ressource abrufen, ohne den Serverzustand zu ändern"""
user = find_user(user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Benutzer nicht gefunden"
)
return user
# POST - unsicher und nicht idempotent
@app.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
"""Neue Ressource erstellen"""
if user_exists(user.email):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Benutzer existiert bereits"
)
new_user = create_new_user(user)
return new_user
# PUT - unsicher, aber idempotent
@app.put("/users/{user_id}")
async def replace_user(user_id: int, user: UserUpdate):
"""Ressource vollständig ersetzen"""
if not user_exists(user_id):
# PUT kann eine Ressource erstellen
return create_user_with_id(user_id, user)
return replace_existing_user(user_id, user)
# PATCH – unsicher und normalerweise nicht idempotent
@app.patch("/users/{user_id}")
async def update_user(user_id: int, user: UserPatch):
"""Ressource teilweise aktualisieren"""
existing_user = find_user(user_id)
if not existing_user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Benutzer nicht gefunden"
)
return patch_user_fields(existing_user, user)
# DELETE – unsicher, aber idempotent
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int):
"""Ressource löschen"""
if not user_exists(user_id):
# Idempotenz: Das Löschen einer nicht existierenden Ressource gibt weiterhin Erfolg zurück
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT)
delete_existing_user(user_id)
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT)
from fastapi.responses import JSONResponse
app = FastAPI()
# GET - sicher und idempotent
@app.get("/users/{user_id}")
async def get_user(user_id: int):
"""Ressource abrufen, ohne den Serverzustand zu ändern"""
user = find_user(user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Benutzer nicht gefunden"
)
return user
# POST - unsicher und nicht idempotent
@app.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
"""Neue Ressource erstellen"""
if user_exists(user.email):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Benutzer existiert bereits"
)
new_user = create_new_user(user)
return new_user
# PUT - unsicher, aber idempotent
@app.put("/users/{user_id}")
async def replace_user(user_id: int, user: UserUpdate):
"""Ressource vollständig ersetzen"""
if not user_exists(user_id):
# PUT kann eine Ressource erstellen
return create_user_with_id(user_id, user)
return replace_existing_user(user_id, user)
# PATCH – unsicher und normalerweise nicht idempotent
@app.patch("/users/{user_id}")
async def update_user(user_id: int, user: UserPatch):
"""Ressource teilweise aktualisieren"""
existing_user = find_user(user_id)
if not existing_user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Benutzer nicht gefunden"
)
return patch_user_fields(existing_user, user)
# DELETE – unsicher, aber idempotent
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int):
"""Ressource löschen"""
if not user_exists(user_id):
# Idempotenz: Das Löschen einer nicht existierenden Ressource gibt weiterhin Erfolg zurück
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT)
delete_existing_user(user_id)
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT)
Verwendung von HTTP-Statuscodes
Beispiel
from fastapi import status
# 2xx Erfolgsantwort
@app.post("/users", status_code=status.HTTP_201_CREATED) # 201 Erstellt
@app.get("/users") # 200 OK (Standard)
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT) # 204 Kein Inhalt
# 4xx Clientfehler
@app.get("/users/{user_id}")
async def get_user(user_id: int):
if user_id < 1:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, # 400 Ungültige Anfrage
detail="Benutzer-ID muss positiv sein"
)
user = find_user(user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, # 404 Nicht gefunden
detail="Benutzer nicht gefunden"
)
return user
# Fehler in der Geschäftslogik
@app.post("/users/{user_id}/follow")
async def follow_user(user_id: int, current_user_id: int):
if user_id == current_user_id:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, # 422 Nicht verarbeitbare Entität
detail="Sie können sich nicht selbst folgen"
)
if is_already_following(current_user_id, user_id):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, # 409 Konflikt
detail="Sie folgen diesem Benutzer bereits"
)
# 5xx Serverfehler (wird normalerweise vom Exception-Handler behandelt)
@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, # 500 Interner Serverfehler
content={"detail": "Interner Serverfehler"}
)
# 2xx Erfolgsantwort
@app.post("/users", status_code=status.HTTP_201_CREATED) # 201 Erstellt
@app.get("/users") # 200 OK (Standard)
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT) # 204 Kein Inhalt
# 4xx Clientfehler
@app.get("/users/{user_id}")
async def get_user(user_id: int):
if user_id < 1:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, # 400 Ungültige Anfrage
detail="Benutzer-ID muss positiv sein"
)
user = find_user(user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, # 404 Nicht gefunden
detail="Benutzer nicht gefunden"
)
return user
# Fehler in der Geschäftslogik
@app.post("/users/{user_id}/follow")
async def follow_user(user_id: int, current_user_id: int):
if user_id == current_user_id:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, # 422 Nicht verarbeitbare Entität
detail="Sie können sich nicht selbst folgen"
)
if is_already_following(current_user_id, user_id):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, # 409 Konflikt
detail="Sie folgen diesem Benutzer bereits"
)
# 5xx Serverfehler (wird normalerweise vom Exception-Handler behandelt)
@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, # 500 Interner Serverfehler
content={"detail": "Interner Serverfehler"}
)
Schnellreferenz für gängige Statuscodes:
# 成功状态码 200 OK # 请求成功 201 Created # 资源创建成功 204 No Content # 成功但无内容返回 206 Partial Content # 部分内容(分页、断点续传) # 重定向 301 Moved Permanently # 永久重定向 302 Found # 临时重定向 304 Not Modified # 资源未修改(缓存) # 客户端错误 400 Bad Request # 请求格式错误 401 Unauthorized # 未认证 403 Forbidden # 已认证但无权限 404 Not Found # 资源不存在 405 Method Not Allowed # HTTP 方法不允许 409 Conflict # 资源冲突 422 Unprocessable Entity # 数据验证失败 429 Too Many Requests # 请求过于频繁 # 服务器错误 500 Internal Server Error # 服务器内部错误 502 Bad Gateway # 网关错误 503 Service Unavailable # 服务不可用
JSON-Datenformat
JSON-Grundlagen und bewährte Verfahren
Beispiel
from datetime import datetime
from decimal import Decimal
from typing import Optional, List
from pydantic import BaseModel, Field
import json
# Spezifikation des JSON-Datenformats
class UserResponse(BaseModel):
id: int
username: str
email: str
full_name: Optional[str] = None
is_active: bool = True
created_at: datetime
updated_at: Optional[datetime] = None
# Konfiguration der JSON-Serialisierung
class Config:
# Feldaliase zulassen
allow_population_by_field_name = True
# JSON-Encoder
json_encoders = {
datetime: lambda v: v.isoformat(),
Decimal: lambda v: float(v)
}
# Antwortformat standardisieren
class APIResponse(BaseModel):
"""Standard-API-Antwortformat"""
success: bool = True
message: str = "Operation erfolgreich"
data: Optional[dict] = None
errors: Optional[List[str]] = None
timestamp: datetime = Field(default_factory=datetime.now)
# Format der paginierten Antwort
class PaginatedResponse(BaseModel):
items: List[dict]
total: int
page: int
size: int
pages: int
@app.get("/users", response_model=PaginatedResponse)
async def get_users(page: int = 1, size: int = 10):
"""Gibt eine paginierte Liste von Benutzern zurück"""
users = get_users_paginated(page, size)
total = count_users()
return PaginatedResponse(
items=users,
total=total,
page=page,
size=size,
pages=(total + size - 1) // size
)
from decimal import Decimal
from typing import Optional, List
from pydantic import BaseModel, Field
import json
# Spezifikation des JSON-Datenformats
class UserResponse(BaseModel):
id: int
username: str
email: str
full_name: Optional[str] = None
is_active: bool = True
created_at: datetime
updated_at: Optional[datetime] = None
# Konfiguration der JSON-Serialisierung
class Config:
# Feldaliase zulassen
allow_population_by_field_name = True
# JSON-Encoder
json_encoders = {
datetime: lambda v: v.isoformat(),
Decimal: lambda v: float(v)
}
# Antwortformat standardisieren
class APIResponse(BaseModel):
"""Standard-API-Antwortformat"""
success: bool = True
message: str = "Operation erfolgreich"
data: Optional[dict] = None
errors: Optional[List[str]] = None
timestamp: datetime = Field(default_factory=datetime.now)
# Format der paginierten Antwort
class PaginatedResponse(BaseModel):
items: List[dict]
total: int
page: int
size: int
pages: int
@app.get("/users", response_model=PaginatedResponse)
async def get_users(page: int = 1, size: int = 10):
"""Gibt eine paginierte Liste von Benutzern zurück"""
users = get_users_paginated(page, size)
total = count_users()
return PaginatedResponse(
items=users,
total=total,
page=page,
size=size,
pages=(total + size - 1) // size
)
JSON-Schema- und Datenvalidierung
Beispiel
from pydantic import BaseModel, validator, Field
from typing import Optional, List
from enum import Enum
class UserRole(str, Enum):
ADMIN = "admin"
USER = "user"
MODERATOR = "moderator"
class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=50, description="Benutzername")
email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$', description="E-Mail-Adresse")
password: str = Field(..., min_length=8, description="Passwort")
full_name: Optional[str] = Field(None, max_length=100, description="Vollständiger Name")
role: UserRole = Field(UserRole.USER, description="Benutzerrolle")
age: Optional[int] = Field(None, ge=0, le=150, description="Alter")
@validator('username')
def username_must_be_alphanumeric(cls, v):
if not v.isalnum():
raise ValueError('Benutzername darf nur Buchstaben und Zahlen enthalten')
return v
@validator('password')
def password_strength(cls, v):
if not any(c.isupper() for c in v):
raise ValueError('Passwort muss mindestens einen Großbuchstaben enthalten')
if not any(c.islower() for c in v):
raise ValueError('Passwort muss mindestens einen Kleinbuchstaben enthalten')
if not any(c.isdigit() for c in v):
raise ValueError('Passwort muss mindestens eine Ziffer enthalten')
return v
class Config:
# JSON-Schema-Beispiel generieren
schema_extra = {
"example": {
"username": "johndoe",
"email": "[email protected]",
"password": "SecurePass123",
"full_name": "John Doe",
"role": "user",
"age": 30
}
}
# Benutzerdefinierte Validatoren verwenden
@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
"""Erstelle einen neuen Benutzer, einschließlich Datenvalidierung"""
# Pydantic validiert Eingabedaten automatisch
return await create_new_user(user)
from typing import Optional, List
from enum import Enum
class UserRole(str, Enum):
ADMIN = "admin"
USER = "user"
MODERATOR = "moderator"
class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=50, description="Benutzername")
email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$', description="E-Mail-Adresse")
password: str = Field(..., min_length=8, description="Passwort")
full_name: Optional[str] = Field(None, max_length=100, description="Vollständiger Name")
role: UserRole = Field(UserRole.USER, description="Benutzerrolle")
age: Optional[int] = Field(None, ge=0, le=150, description="Alter")
@validator('username')
def username_must_be_alphanumeric(cls, v):
if not v.isalnum():
raise ValueError('Benutzername darf nur Buchstaben und Zahlen enthalten')
return v
@validator('password')
def password_strength(cls, v):
if not any(c.isupper() for c in v):
raise ValueError('Passwort muss mindestens einen Großbuchstaben enthalten')
if not any(c.islower() for c in v):
raise ValueError('Passwort muss mindestens einen Kleinbuchstaben enthalten')
if not any(c.isdigit() for c in v):
raise ValueError('Passwort muss mindestens eine Ziffer enthalten')
return v
class Config:
# JSON-Schema-Beispiel generieren
schema_extra = {
"example": {
"username": "johndoe",
"email": "[email protected]",
"password": "SecurePass123",
"full_name": "John Doe",
"role": "user",
"age": 30
}
}
# Benutzerdefinierte Validatoren verwenden
@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
"""Erstelle einen neuen Benutzer, einschließlich Datenvalidierung"""
# Pydantic validiert Eingabedaten automatisch
return await create_new_user(user)
Format der Fehlerantwort
Beispiel
class ErrorDetail(BaseModel):
field: str
message: str
code: str
class ErrorResponse(BaseModel):
error: str
message: str
details: Optional[List[ErrorDetail]] = None
timestamp: datetime = Field(default_factory=datetime.now)
# Benutzerdefinierte Ausnahmebehandlung
from fastapi import Request
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
errors = []
for error in exc.errors():
errors.append(ErrorDetail(
field='.'.join(str(x) for x in error['loc']),
message=error['msg'],
code=error['type']
))
return JSONResponse(
status_code=422,
content=ErrorResponse(
error="Validierungsfehler",
message="Validierung der Anfragedaten fehlgeschlagen",
details=errors
).dict()
)
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
return JSONResponse(
status_code=exc.status_code,
content=ErrorResponse(
error=f"HTTP {exc.status_code}",
message=exc.detail
).dict()
)
field: str
message: str
code: str
class ErrorResponse(BaseModel):
error: str
message: str
details: Optional[List[ErrorDetail]] = None
timestamp: datetime = Field(default_factory=datetime.now)
# Benutzerdefinierte Ausnahmebehandlung
from fastapi import Request
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
errors = []
for error in exc.errors():
errors.append(ErrorDetail(
field='.'.join(str(x) for x in error['loc']),
message=error['msg'],
code=error['type']
))
return JSONResponse(
status_code=422,
content=ErrorResponse(
error="Validierungsfehler",
message="Validierung der Anfragedaten fehlgeschlagen",
details=errors
).dict()
)
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
return JSONResponse(
status_code=exc.status_code,
content=ErrorResponse(
error=f"HTTP {exc.status_code}",
message=exc.detail
).dict()
)
FastAPI-spezifische Konzepte
Automatische Dokumentationsgenerierung
Beispiel
from fastapi import FastAPI, Query, Path, Body
from fastapi.openapi.utils import get_openapi
app = FastAPI(
title="Meine API",
description⟦⟧Dies ist eine Beispiel-API, die die Funktionen von FastAPI demonstriert=This is an example API, demonstrating the features of FastAPI,
version="1.0.0",
terms_of_service="http://example.com/terms/",
contact={
"name": "Entwickler",
"url": "http://example.com/contact/",
"email": "[email protected]",
},
license_info={
"name": "MIT",
"url": "https://opensource.org/licenses/MIT",
},
)
@app.get(
"/users/{user_id}",
summary=Benutzerinformationen abrufen,
description⟦⟧"Detaillierte Benutzerinformationen anhand der Benutzer-ID abrufen"="Get detailed user information by user ID",
response_description=Benutzerinformationsobjekt,
tags=["Benutzerverwaltung"]
)
async def get_user(
user_id: int = Path(..., title="Benutzer-ID", description="Abzurufende Benutzer-ID", ge=1),
include_posts: bool = Query(False, title="Artikel einschließen", description=Ob die Artikelliste des Benutzers einbezogen werden soll)
):
"""
Benutzerinformationen abrufen:
- **user_id**: Die eindeutige Kennung des Benutzers.
**include_posts**: Ob die Beitragsliste des Benutzers in die Antwort einbezogen werden soll
Gibt die grundlegenden Informationen des Benutzers zurück; wenn include_posts True ist, enthält sie außerdem die Liste der Artikel.
"""
user = find_user(user_id)
if include_posts:
user.posts = get_user_posts(user_id)
return user
# Benutzerdefiniertes OpenAPI-Schema
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="Benutzerdefinierte API-Dokumentation",
version="2.5.0",
description="Dies ist ein benutzerdefiniertes OpenAPI-Schema",
routes=app.routes,
)
# Benutzerdefinierte Informationen hinzufügen
openapi_schema["info"]["x-logo"] = {
"url": "https://example.com/logo.png"
}
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
from fastapi.openapi.utils import get_openapi
app = FastAPI(
title="Meine API",
description⟦⟧Dies ist eine Beispiel-API, die die Funktionen von FastAPI demonstriert=This is an example API, demonstrating the features of FastAPI,
version="1.0.0",
terms_of_service="http://example.com/terms/",
contact={
"name": "Entwickler",
"url": "http://example.com/contact/",
"email": "[email protected]",
},
license_info={
"name": "MIT",
"url": "https://opensource.org/licenses/MIT",
},
)
@app.get(
"/users/{user_id}",
summary=Benutzerinformationen abrufen,
description⟦⟧"Detaillierte Benutzerinformationen anhand der Benutzer-ID abrufen"="Get detailed user information by user ID",
response_description=Benutzerinformationsobjekt,
tags=["Benutzerverwaltung"]
)
async def get_user(
user_id: int = Path(..., title="Benutzer-ID", description="Abzurufende Benutzer-ID", ge=1),
include_posts: bool = Query(False, title="Artikel einschließen", description=Ob die Artikelliste des Benutzers einbezogen werden soll)
):
"""
Benutzerinformationen abrufen:
- **user_id**: Die eindeutige Kennung des Benutzers.
**include_posts**: Ob die Beitragsliste des Benutzers in die Antwort einbezogen werden soll
Gibt die grundlegenden Informationen des Benutzers zurück; wenn include_posts True ist, enthält sie außerdem die Liste der Artikel.
"""
user = find_user(user_id)
if include_posts:
user.posts = get_user_posts(user_id)
return user
# Benutzerdefiniertes OpenAPI-Schema
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="Benutzerdefinierte API-Dokumentation",
version="2.5.0",
description="Dies ist ein benutzerdefiniertes OpenAPI-Schema",
routes=app.routes,
)
# Benutzerdefinierte Informationen hinzufügen
openapi_schema["info"]["x-logo"] = {
"url": "https://example.com/logo.png"
}
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
Vorschau des Dependency-Injection-Systems
Beispiel
from fastapi import Depends
# Einfache Abhängigkeit
def get_current_user_id() -> int:
# Benutzer-ID aus Token parsen
return 123
def get_database_session():
# Datenbanksitzung erstellen
db = DatabaseSession()
try:
yield db
finally:
db.close()
# Abhängigkeiten verwenden
@app.get("/profile")
async def get_profile(
user_id: int = Depends(get_current_user_id),
db = Depends(get_database_session)
):
return get_user_profile(db, user_id)
# Abhängigkeitskette
def get_current_user(
user_id: int = Depends(get_current_user_id),
db = Depends(get_database_session)
):
return db.query(User).filter(User.id == user_id).first()
@app.get("/dashboard")
async def get_dashboard(
current_user: User = Depends(get_current_user) # Hängt von anderen Abhängigkeiten ab
):
return generate_dashboard(current_user)
# Einfache Abhängigkeit
def get_current_user_id() -> int:
# Benutzer-ID aus Token parsen
return 123
def get_database_session():
# Datenbanksitzung erstellen
db = DatabaseSession()
try:
yield db
finally:
db.close()
# Abhängigkeiten verwenden
@app.get("/profile")
async def get_profile(
user_id: int = Depends(get_current_user_id),
db = Depends(get_database_session)
):
return get_user_profile(db, user_id)
# Abhängigkeitskette
def get_current_user(
user_id: int = Depends(get_current_user_id),
db = Depends(get_database_session)
):
return db.query(User).filter(User.id == user_id).first()
@app.get("/dashboard")
async def get_dashboard(
current_user: User = Depends(get_current_user) # Hängt von anderen Abhängigkeiten ab
):
return generate_dashboard(current_user)
Die Bedeutung von Typhinweisen
Beispiel
from typing import List, Optional, Union, Dict, Any
from pydantic import BaseModel
# FastAPI verlässt sich vollständig auf Typhinweise:
# 1. Datenvalidierung
# 2. Dokumentationsgenerierung
# 3. Editor-Unterstützung
@app.get("/items/{item_id}")
async def get_item(
item_id: int, # Pfadparameter, automatisch in int konvertiert
q: Optional[str] = None, # Optionale Query-Parameter
limit: int = 10 # Query-Parameter mit Standardwerten
) -> Dict[str, Any]: # Rückgabetyp-Hinweis
"""Typ-Hinweise teilen FastAPI mit, wie Parameter und Antworten behandelt werden"""
return {"item_id": item_id, "q": q, "limit": limit}
# Komplexer Typ-Hinweis
@app.post("/items/")
async def create_items(
items: List[ItemCreate] # Empfängt eine Liste von ItemCreate-Objekten
) -> List[ItemResponse]: # Gibt eine Liste von ItemResponse-Objekten zurück
return [create_item(item) for item in items]
# Union-Typ (mehrere mögliche Typen)
@app.get("/search")
async def search(
q: str,
result_type: str = "json"
) -> Union[List[Dict], str]: # Kann ein Dictionary, eine Liste oder einen String zurückgeben
if result_type == "json":
return [{"title": "Result 1"}, {"title": "Ergebnis 2"}]
else:
return "Ergebnis 1, Ergebnis 2"
from pydantic import BaseModel
# FastAPI verlässt sich vollständig auf Typhinweise:
# 1. Datenvalidierung
# 2. Dokumentationsgenerierung
# 3. Editor-Unterstützung
@app.get("/items/{item_id}")
async def get_item(
item_id: int, # Pfadparameter, automatisch in int konvertiert
q: Optional[str] = None, # Optionale Query-Parameter
limit: int = 10 # Query-Parameter mit Standardwerten
) -> Dict[str, Any]: # Rückgabetyp-Hinweis
"""Typ-Hinweise teilen FastAPI mit, wie Parameter und Antworten behandelt werden"""
return {"item_id": item_id, "q": q, "limit": limit}
# Komplexer Typ-Hinweis
@app.post("/items/")
async def create_items(
items: List[ItemCreate] # Empfängt eine Liste von ItemCreate-Objekten
) -> List[ItemResponse]: # Gibt eine Liste von ItemResponse-Objekten zurück
return [create_item(item) for item in items]
# Union-Typ (mehrere mögliche Typen)
@app.get("/search")
async def search(
q: str,
result_type: str = "json"
) -> Union[List[Dict], str]: # Kann ein Dictionary, eine Liste oder einen String zurückgeben
if result_type == "json":
return [{"title": "Result 1"}, {"title": "Ergebnis 2"}]
else:
return "Ergebnis 1, Ergebnis 2"
Praktische Beispiele für Kernkonzepte
Beispiel
from fastapi import FastAPI, HTTPException, status, Depends, Query
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
import asyncio
# Umfassendes Beispiel: Blog API
app = FastAPI(
title=Blog API,
description="Ein Blog-System, das FastAPI-Kernkonzepte veranschaulicht",
version="1.0.0"
)
# Datenmodell
class PostBase(BaseModel):
title: str = Field(..., min_length=1, max_length=200)
content: str = Field(..., min_length=1)
published: bool = Field(True)
class PostCreate(PostBase):
pass
class PostResponse(PostBase):
id: int
author_id: int
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
orm_mode = True
# Mock-Datenbank
posts_db = []
next_id = 1
# Abhängigkeit: aktuellen Benutzer abrufen (vereinfachte Version)
async def get_current_user() -> int:
# In echten Anwendungen wird es aus dem JWT-Token geparst
return 1
# Asynchrone Pfadoperation
@app.get("/posts", response_model=List[PostResponse], tags=["Artikel"])
async def list_posts(
skip: int = Query(0, ge=0, description=Anzahl übersprungener Artikel),
limit: int = Query(10, ge=1, le=100, description=Anzahl der zurückgegebenen Artikel),
published_only: bool = Query(True, description="Nur veröffentlichte Artikel zurückgeben")
):
"""
Artikelliste abrufen
Unterstützt Paginierung und Filterung
"""
# Asynchrone Datenbankabfrage simulieren
await asyncio.sleep(0.1)
filtered_posts = posts_db
if published_only:
filtered_posts = [p for p in posts_db if p["published"]]
return filtered_posts[skip:skip + limit]
@app.post("/posts", response_model=PostResponse, status_code=status.HTTP_201_CREATED, tags=["Article"])
async def create_post(
post: PostCreate,
current_user_id: int = Depends(get_current_user)
):
"""
Artikel erstellen
Erfordert Benutzerauthentifizierung
"""
global next_id
# Asynchrone Datenbankoperationen simulieren
await asyncio.sleep(0.1)
new_post = {
"id": next_id,
"title": post.title,
"content": post.content,
"published": post.published,
"author_id": current_user_id,
"created_at": datetime.now(),
"updated_at": None
}
posts_db.append(new_post)
next_id += 1
return new_post
@app.get("/posts/{post_id}", response_model=PostResponse, tags=["Artikel"])
async def get_post(post_id: int):
"""
Bestimmten Artikel abrufen
Artikeldetails basierend auf Artikel-ID zurückgeben
"""
# Asynchrone Datenbankabfrage simulieren
await asyncio.sleep(0.1)
post = next((p for p in posts_db if p["id"] == post_id), None)
if not post:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Beitrag mit ID {post_id} nicht gefunden"
)
return post
# Health-Check-Endpunkt
@app.get("/health", tags=[System])
async def health_check():
"""Systemzustandsprüfung"""
return {
"status": "gesund",
"timestamp": datetime.now(),
"posts_count": len(posts_db)
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
import asyncio
# Umfassendes Beispiel: Blog API
app = FastAPI(
title=Blog API,
description="Ein Blog-System, das FastAPI-Kernkonzepte veranschaulicht",
version="1.0.0"
)
# Datenmodell
class PostBase(BaseModel):
title: str = Field(..., min_length=1, max_length=200)
content: str = Field(..., min_length=1)
published: bool = Field(True)
class PostCreate(PostBase):
pass
class PostResponse(PostBase):
id: int
author_id: int
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
orm_mode = True
# Mock-Datenbank
posts_db = []
next_id = 1
# Abhängigkeit: aktuellen Benutzer abrufen (vereinfachte Version)
async def get_current_user() -> int:
# In echten Anwendungen wird es aus dem JWT-Token geparst
return 1
# Asynchrone Pfadoperation
@app.get("/posts", response_model=List[PostResponse], tags=["Artikel"])
async def list_posts(
skip: int = Query(0, ge=0, description=Anzahl übersprungener Artikel),
limit: int = Query(10, ge=1, le=100, description=Anzahl der zurückgegebenen Artikel),
published_only: bool = Query(True, description="Nur veröffentlichte Artikel zurückgeben")
):
"""
Artikelliste abrufen
Unterstützt Paginierung und Filterung
"""
# Asynchrone Datenbankabfrage simulieren
await asyncio.sleep(0.1)
filtered_posts = posts_db
if published_only:
filtered_posts = [p for p in posts_db if p["published"]]
return filtered_posts[skip:skip + limit]
@app.post("/posts", response_model=PostResponse, status_code=status.HTTP_201_CREATED, tags=["Article"])
async def create_post(
post: PostCreate,
current_user_id: int = Depends(get_current_user)
):
"""
Artikel erstellen
Erfordert Benutzerauthentifizierung
"""
global next_id
# Asynchrone Datenbankoperationen simulieren
await asyncio.sleep(0.1)
new_post = {
"id": next_id,
"title": post.title,
"content": post.content,
"published": post.published,
"author_id": current_user_id,
"created_at": datetime.now(),
"updated_at": None
}
posts_db.append(new_post)
next_id += 1
return new_post
@app.get("/posts/{post_id}", response_model=PostResponse, tags=["Artikel"])
async def get_post(post_id: int):
"""
Bestimmten Artikel abrufen
Artikeldetails basierend auf Artikel-ID zurückgeben
"""
# Asynchrone Datenbankabfrage simulieren
await asyncio.sleep(0.1)
post = next((p for p in posts_db if p["id"] == post_id), None)
if not post:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Beitrag mit ID {post_id} nicht gefunden"
)
return post
# Health-Check-Endpunkt
@app.get("/health", tags=[System])
async def health_check():
"""Systemzustandsprüfung"""
return {
"status": "gesund",
"timestamp": datetime.now(),
"posts_count": len(posts_db)
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)