Einen perfekten Blog mit FastAPI erstellen: Autorisierung hinzufügen

Im vorherigen Artikel haben wir erfolgreich ein Benutzeregistrierungssystem und grundlegende Anmeldevalidierungslogik für unseren FastAPI-Blog aufgebaut. Benutzer können Konten erstellen, und die Anwendung kann ihre Benutzernamen und Passwörter überprüfen.

Die aktuelle Anmeldung ist jedoch nur eine einmalige Validierung; der Server "erinnert" sich nicht an den Anmeldestatus des Benutzers. Jedes Mal, wenn die Seite aktualisiert oder eine neue Seite besucht wird, wird der Benutzer wieder zu einem nicht authentifizierten Gast.

In diesem Artikel werden wir Middleware verwenden, um ein echtes Anmeldestatusmanagement für unseren Blog zu implementieren. Wir lernen, wie man Seiten und Funktionen schützt, die eine Anmeldung erfordern, und wie man die Benutzeroberfläche basierend auf dem Anmeldestatus des Benutzers dynamisch aktualisiert.

Sitzungen konfigurieren

Zur Verwaltung von Sitzungen in FastAPI verwenden wir Starlettes SessionMiddleware. Starlette ist das ASGI-Framework, auf dem FastAPI aufbaut, und SessionMiddleware ist das offizielle Standardwerkzeug für die Sitzungsverwaltung.

Installieren Sie zuerst die itsdangerous-Bibliothek. SessionMiddleware verwendet sie, um Sitzungsdaten kryptografisch zu signieren und so deren Sicherheit zu gewährleisten.

pip install itsdangerous

Fügen Sie es dann Ihrer requirements.txt-Datei hinzu:

# requirements.txt
fastapi
uvicorn[standard]
sqlmodel
psycopg2-binary
jinja2
python-dotenv
python-multipart
bcrypt
itsdangerous

Redis zur Speicherung von Sitzungen verwenden

Standardmäßig verschlüsselt SessionMiddleware Sitzungsdaten und speichert sie in einem clientseitigen Cookie. Dieser Ansatz ist einfach und erfordert keine Backend-Speicherung, hat aber den Nachteil einer begrenzten Cookie-Größe (typischerweise 4 KB), was ihn für die Speicherung großer Datenmengen ungeeignet macht.

Für bessere Skalierbarkeit und Sicherheit verwenden wir Redis, eine Hochleistungs-In-Memory-Datenbank, um Sitzungen serverseitig zu speichern. Dies stellt sicher, dass der Anmeldestatus erhalten bleibt, auch wenn der Benutzer den Browser schließt oder der Server neu startet.

Was, wenn Sie kein Redis haben?

Sie können eine Redis-Instanz auf Leapcell erstellen. Leapcell bietet die meisten Tools, die Sie für eine Backend-Anwendung benötigen!

Klicken Sie in der Benutzeroberfläche auf die Schaltfläche "Redis erstellen", um eine neue Redis-Instanz zu erstellen.

Die Redis-Detailseite bietet eine Online-CLI, über die Sie Redis-Befehle direkt ausführen können.

Wenn Sie derzeit keinen Redis-Dienst zur Verfügung haben, verwendet SessionMiddleware standardmäßig signierte Cookies. Für dieses Tutorial wird die Funktionalität nicht beeinträchtigt.

Installieren Sie die Redis-bezogene Abhängigkeit:

pip install redis

Öffnen Sie nun die Datei main.py, um SessionMiddleware zu importieren und zu konfigurieren.

# main.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from starlette.middleware.sessions import SessionMiddleware # Importieren Sie die Middleware
from dotenv import load_dotenv

from database import create_db_and_tables
from routers import posts, users, auth

# Umgebungsvariablen laden
load_dotenv()

@asynccontextmanager
async def lifespan(app: FastAPI):
    print("Creating tables..")
    create_db_and_tables()
    yield

app = FastAPI(lifespan=lifespan)

# Den geheimen Schlüssel aus Umgebungsvariablen lesen
# Stellen Sie sicher, dass "your-secret-key" durch eine wirklich sichere Zufallszeichenfolge ersetzt wird. Sie können eine mit `openssl rand -hex 32` generieren.
SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key")

# SessionMiddleware hinzufügen
app.add_middleware(
    SessionMiddleware,
    secret_key=SECRET_KEY,
    session_cookie="session_id", # Der Name der Sitzungs-ID, die im Cookie gespeichert wird
    max_age=60 * 60 * 24 * 7  # Sitzung läuft nach 7 Tagen ab
)

# Statische Dateien Verzeichnis einhängen
app.mount("/static", StaticFiles(directory="public"), name="static")

# Router einbinden
app.include_router(posts.router)
app.include_router(users.router)
app.include_router(auth.router)

Hinweis: Aus Sicherheitsgründen sollte der secret_key eine komplexe, zufällig generierte Zeichenfolge sein. Wie die Datenbank-URL sollte er über Umgebungsvariablen verwaltet und nicht hartcodiert werden.

Nach der Konfiguration verarbeitet SessionMiddleware automatisch jede Anfrage, analysiert Sitzungsdaten aus dem Cookie der Anfrage und heftet sie an das request.session-Objekt zur Verwendung an.

Implementierung von echten Login- und Logout-Routen

Als Nächstes aktualisieren wir routers/auth.py, um die tatsächliche Login- und Logout-Logik zu verarbeiten.

# routers/auth.py
from fastapi import APIRouter, Request, Depends, Form, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session

from database import get_session
import auth_service

router = APIRouter()
templates = Jinja2Templates(directory="templates")

@router.get("/auth/login", response_class=HTMLResponse)
def show_login_form(request: Request):
    return templates.TemplateResponse("login.html", {"request": request, "title": "Login"})

@router.post("/auth/login")
def login(
    request: Request, # Injizieren Sie das Request-Objekt, um auf die Sitzung zuzugreifen
    username: str = Form(...),
    password: str = Form(...),
    session: Session = Depends(get_session)
):
    user = auth_service.validate_user(username, password, session)
    if not user:
        raise HTTPException(status_code=401, detail="Incorrect username or password")
    
    # Validierung erfolgreich, Benutzerinformationen in der Sitzung speichern
    # SessionMiddleware kümmert sich automatisch um die anschließende Verschlüsselung und Cookie-Setzung
    request.session["user"] = {"username": user.username, "id": str(user.id)}
    
    return RedirectResponse(url="/posts", status_code=302)

@router.get("/auth/logout")
def logout(request: Request):
    # Sitzung löschen
    request.session.clear()
    return RedirectResponse(url="/", status_code=302)

In der login-Funktion speichern wir nach erfolgreicher Validierung des Benutzers ein Wörterbuch mit grundlegenden Benutzerinformationen in request.session["user"]. SessionMiddleware verschlüsselt und signiert diese Sitzungsdaten automatisch und setzt ein Cookie im Browser, das diese enthält. Der Browser enthält dieses Cookie dann automatisch bei allen nachfolgenden Anfragen, sodass der Server den Anmeldestatus des Benutzers erkennen kann.

In der logout-Funktion rufen wir request.session.clear() auf, was die Sitzungsdaten löscht und den Benutzer effektiv abmeldet.

Routen schützen und die Benutzeroberfläche aktualisieren

Nachdem wir nun einen Anmeldemechanismus haben, ist der letzte Schritt, ihn zum Schutz unseres "Beitrag erstellen"-Features zu verwenden und verschiedene Benutzeroberflächenelemente basierend auf dem Anmeldestatus anzuzeigen.

Erstellen einer Authentifizierungsabhängigkeit

In FastAPI ist der eleganteste Weg, Routen zu schützen, die Verwendung von Dependency Injection. Wir erstellen eine Abhängigkeitsfunktion, um zu überprüfen, ob ein Benutzer angemeldet ist.

Erstellen Sie im Stammverzeichnis des Projekts eine neue Datei namens auth_dependencies.py:

# auth_dependencies.py
from fastapi import Request, Depends, HTTPException, status
from fastapi.responses import RedirectResponse

def login_required(request: Request):
    """
    Eine Abhängigkeit zur Überprüfung, ob der Benutzer angemeldet ist.
    Wenn nicht angemeldet, Umleitung zur Anmeldeseite.
    """
    if not request.session.get("user"):
        # Sie können auch eine HTTPException auslösen
        # raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
        return RedirectResponse(url="/auth/login", status_code=status.HTTP_302_FOUND)
    return request.session.get("user")

def get_user_from_session(request: Request) -> dict | None:
    """
    Ruft Benutzerinformationen aus der Sitzung ab (falls vorhanden).
    Diese Abhängigkeit erzwingt keine Anmeldung, sie dient nur dem bequemen Abrufen von Benutzerinformationen in Vorlagen.
    """
    return request.session.get("user")

Die Logik der ersten Funktion, login_required, ist einfach: Wenn user nicht in request.session vorhanden ist, wird der Benutzer zur Anmeldeseite umgeleitet. Wenn sie vorhanden ist, gibt sie die Benutzerinformationen zurück, sodass die Routenfunktion sie direkt verwenden kann.

Anwenden der Abhängigkeit

Öffnen Sie routers/posts.py und wenden Sie die login_required-Abhängigkeit auf die Routen an, die geschützt werden müssen.

# routers/posts.py
import uuid
from fastapi import APIRouter, Request, Depends, Form
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session, select

from database import get_session
from models import Post
from auth_dependencies import login_required # Importieren Sie die Abhängigkeit

router = APIRouter()
templates = Jinja2Templates(directory="templates")

# ... andere Routen ...

# Wenden Sie die Abhängigkeit an, um diese Route zu schützen
@router.get("/posts/new", response_class=HTMLResponse)
def new_post_form(request: Request, user: dict = Depends(login_required)):
    return templates.TemplateResponse("new-post.html", {"request": request, "title": "New Post", "user": user})

# Wenden Sie die Abhängigkeit an, um diese Route zu schützen
@router.post("/posts", response_class=HTMLResponse)
def create_post(
    title: str = Form(...), 
    content: str = Form(...),
    session: Session = Depends(get_session),
    user: dict = Depends(login_required) # Stellen Sie sicher, dass nur angemeldete Benutzer Beiträge erstellen können
):
    new_post = Post(title=title, content=content)
    session.add(new_post)
    session.commit()
    return RedirectResponse(url="/posts", status_code=302)

# ... andere Routen ...

Wenn nun ein nicht authentifizierter Benutzer versucht, auf /posts/new zuzugreifen, wird er automatisch zur Anmeldeseite umgeleitet.

Aktualisieren der Frontend-Benutzeroberfläche

Schließlich aktualisieren wir die Benutzeroberfläche, um verschiedene Schaltflächen basierend auf dem Anmeldestatus des Benutzers anzuzeigen. Wir verwenden die Abhängigkeit get_user_from_session, um Benutzerinformationen abzurufen und sie an die Vorlagen zu übergeben.

Ändern Sie templates/_header.html:

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8" />
  <meta name="viewport" content="width=device-width, initial-scale=1.0" />
  <title>{{ title }}</title>
  <link rel="stylesheet" href="/static/css/style.css" />
</head>
<body>
  <header>
    <h1><a href="/">My Blog</a></h1>
    <nav>
      {% if user %}
        <span class="welcome-msg">Welcome, {{ user.username }}</span>
        <a href="/posts/new" class="new-post-btn">New Post</a>
        <a href="/auth/logout" class="nav-link">Logout</a>
      {% else %}
        <a href="/users/register" class="nav-link">Register</a>
        <a href="/auth/login" class="nav-link">Login</a>
      {% endif %}
    </nav>
  </header>
  <main>

Damit die obige Vorlage korrekt funktioniert, müssen wir alle Routen, die Seiten rendern, mit den Benutzerinformationen für die Ansicht aktualisieren.

Ändern Sie in routers/posts.py alle Methoden, die Ansichten rendern:

# routers/posts.py
# ... Imports ...
from auth_dependencies import get_user_from_session, login_required # Importieren Sie die neuen Abhängigkeiten

# ...

@router.get("/", response_class=HTMLResponse)
def root():
    return RedirectResponse(url="/posts", status_code=302)

@router.get("/posts", response_class=HTMLResponse)
def get_all_posts(
    request: Request, 
    session: Session = Depends(get_session),
    user: dict | None = Depends(get_user_from_session) # Sitzungsbenutzerinformationen abrufen
):
    statement = select(Post).order_by(Post.createdAt.desc())
    posts = session.exec(statement).all()
    # Benutzer an die Vorlage übergeben
    return templates.TemplateResponse("index.html", {"request": request, "posts": posts, "title": "Home", "user": user})

# ... new_post_form route wurde oben aktualisiert ...

@router.get("/posts/{post_id}", response_class=HTMLResponse)
def get_post_by_id(
    request: Request, 
    post_id: uuid.UUID, 
    session: Session = Depends(get_session),
    user: dict | None = Depends(get_user_from_session) # Sitzungsbenutzerinformationen abrufen
):
    post = session.get(Post, post_id)
    # Benutzer an die Vorlage übergeben
    return templates.TemplateResponse("post.html", {"request": request, "post": post, "title": post.title, "user": user})

Ebenso müssen wir auch die Vorlagen-Rendering-Routen in routers/users.py und routers/auth.py aktualisieren, indem wir user: dict | None = Depends(get_user_from_session) hinzufügen und den user an die Vorlage übergeben.

Ausführen und Testen

Starten Sie nun Ihre Anwendung neu:

uvicorn main:app --reload

Besuchen Sie http://localhost:8000. Sie sollten oben rechts die Schaltflächen "Anmelden" und "Registrieren" sehen.

Versuchen Sie, http://localhost:8000/posts/new aufzurufen. Sie werden automatisch zur Anmeldeseite weitergeleitet.

Registrieren Sie sich nun und melden Sie sich an. Nach erfolgreicher Anmeldung werden Sie zur Homepage weitergeleitet, und Sie sehen oben rechts die Schaltflächen "Willkommen, [Ihr Benutzername]", "Neuer Beitrag" und "Abmelden".

An dieser Stelle können Sie auf "Neuer Beitrag" klicken, um einen neuen Artikel zu erstellen. Wenn Sie sich abmelden und versuchen, erneut auf /posts/new zuzugreifen, werden Sie wieder weitergeleitet.

Damit haben wir ein vollständiges Benutzerauthentifizierungssystem zu unserem Blog hinzugefügt. Sie müssen sich keine Sorgen mehr machen, dass Ihre Freunde mit Ihrem Blog herumspielen!