Flask-Datenbankintegration

Die meisten Webanwendungen benötigen eine persistente Datenspeicherung.

Dieses Kapitel verwendet die in Python integrierte SQLite-Datenbank, um zu zeigen, wie eine Datenbank in Flask integriert wird, ohne zusätzliche Abhängigkeiten zu installieren.

SQLite-Schnellstart

SQLite ist eine leichtgewichtige Datenbank, die Daten in einer einzigen Datei speichert und von der Python-Standardbibliothek unterstützt wird.

Es ist die beste Wahl zum Lernen und für Prototyping – es ist nicht nötig, einen Datenbankserver zu installieren oder irgendetwas zu konfigurieren.

Beispiel

# Ein einfaches SQLite-Beispiel (ohne Verwendung von Flask)
import sqlite3

# Verbindung zur Datenbank herstellen (example.db wird automatisch erstellt, wenn sie nicht existiert)
conn = sqlite3.connect("example.db")
# row_factory macht Abfrageergebnisse über Feldnamen wie ein Wörterbuch zugänglich
conn.row_factory = sqlite3.Row

# Tabelle erstellen
conn.execute("CREATE TABLE IF NOT EXISTS user (id INTEGER PRIMARY KEY, username TEXT UNIQUE, password TEXT)")
# Daten einfügen
conn.execute("INSERT INTO user (username, password) VALUES (?, ?)", ("example", "pass123"))
conn.commit()

# Daten abfragen
rows = conn.execute("SELECT * FROM user").fetchall()
for row in rows:
    print(dict(row))  # Ausgabe: {'id': 1, 'username': 'example', 'password': 'pass123'}

conn.close()

Verwendung des g-Objekts in Flask zur Verwaltung von Datenbankverbindungen

gEs ist ein spezielles Objekt, dessen Lebenszyklus eine einzelne Anfrage ist.

Die Datenbankverbindung speicherngEs stellt sicher, dass mehrere Datenbankoperationen innerhalb derselben Anfrage dieselbe Verbindung gemeinsam nutzen, die nach dem Ende der Anfrage automatisch bereinigt wird.

Beispiel

# Dateipfad: db.py (Datenbankverwaltungsmodul)
import sqlite3
from flask import g, current_app

def get_db():
    """Eine Datenbankverbindung abrufen, die innerhalb derselben Anfrage wiederverwendet wird"""
    if "db" not in g:
        # Datenbankpfad aus der Konfiguration lesen
        db_path = current_app.config.get("DATABASE", "example.db")
        g.db = sqlite3.connect(db_path)
        g.db.row_factory = sqlite3.Row  # Erlaubt den Zugriff auf Ergebnisse über Feldnamen
    return g.db

def close_db(error=None):
    """Datenbankverbindung schließen (wird automatisch am Ende der Anfrage aufgerufen)"""
    db = g.pop("db", None)
    if db is not None:
        db.close()

def init_db():
    """Datenbanktabellenstruktur initialisieren"""
    db = get_db()
    db.execute("""
        CREATE TABLE IF NOT EXISTS user (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            password TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    db.execute("""
        CREATE TABLE IF NOT EXISTS post (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            body TEXT NOT NULL,
            author_id INTEGER NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (author_id) REFERENCES user (id)
        )
    """)
    db.commit()

def init_app(app):
    """Datenbankbezogene Funktionen in der Flask-Anwendung registrieren"""
    # Datenbankverbindung am Ende der Anfrage automatisch schließen
    app.teardown_appcontext(close_db)
    # CLI-Befehl registrieren: flask init-db
    app.cli.command("init-db")(init_db)

Integration des Factory-Musters

Verwende das Factory-Muster, um die Datenbank mit anderen Modulen zu integrieren:

Beispiel

# Dateipfad: app.py
from flask import Flask

def create_app():
    app = Flask(__name__)
    app.secret_key = "dev-secret-key"

    # Konfiguriere den Datenbankpfad (im Instanzordner gespeichert)
    import os
    app.config["DATABASE"] = os.path.join(app.instance_path, "example.db")

    # Stelle sicher, dass der Instanzordner existiert
    os.makedirs(app.instance_path, exist_ok=True)

    # Datenbank initialisieren
    from db import init_app as init_db
    init_db(app)

    # Blueprint registrieren
    from auth import bp as auth_bp
    from blog import bp as blog_bp
    app.register_blueprint(auth_bp, url_prefix="/auth")
    app.register_blueprint(blog_bp)

    return app

Schreibe eine vollständige Blog-API

Kombiniere die Datenbank, um CRUD-Operationen (Create, Read, Update, Delete) für Artikel zu implementieren:

Beispiel

# Dateipfad: blog.py
from flask import Blueprint, request, jsonify, abort, g
from db import get_db

bp = Blueprint("blog", __name__)

@bp.get("/api/posts")
def list_posts():
    """Die Liste aller Artikel abrufen"""
    db = get_db()
    posts = db.execute(
        "SELECT p.id, p.title, p.body, p.created_at, u.username"
        " FROM post p JOIN user u ON p.author_id = u.id"
        " ORDER BY p.created_at DESC"
    ).fetchall()
    # Konvertiere eine Liste von Row-Objekten in eine Liste von Dicts
    return jsonify([dict(post) for post in posts])

@bp.get("/api/posts/<int:post_id>")
def get_post(post_id):
    """Ruft die Details eines einzelnen Artikels ab"""
    db = get_db()
    post = db.execute(
        "SELECT id, title, body, created_at FROM post WHERE id = ?",
        (post_id,)
    ).fetchone()

    if post is None:
        abort(404, description=f"Artikel {post_id} existiert nicht")

    return jsonify(dict(post))

@bp.post("/api/posts")
def create_post():
    """Einen neuen Artikel erstellen"""
    data = request.json
    title = data.get("title", "").strip()
    body = data.get("body", "").strip()
    author_id = data.get("author_id", 1)  # Standard-Autor-ID

    if not title:
        return jsonify({"error": "Titel darf nicht leer sein"}), 400

    db = get_db()
    cursor = db.execute(
        "INSERT INTO post (title, body, author_id) VALUES (?, ?, ?)",
        (title, body, author_id)
    )
    db.commit()

    return jsonify({"id": cursor.lastrowid, "title": title, "message": "Artikel erfolgreich erstellt"}), 201

@bp.put("/api/posts/<int:post_id>")
def update_post(post_id):
    """Einen Artikel aktualisieren"""
    data = request.json
    title = data.get("title", "").strip()
    body = data.get("body", "").strip()

    if not title:
        return jsonify({"error": "Titel darf nicht leer sein"}), 400

    db = get_db()
    db.execute(
        "UPDATE post SET title = ?, body = ? WHERE id = ?",
        (title, body, post_id)
    )
    db.commit()

    return jsonify({"id": post_id, "message": "Artikel erfolgreich aktualisiert"})

@bp.delete("/api/posts/<int:post_id>")
def delete_post(post_id):
    """Einen Artikel löschen"""
    db = get_db()
    db.execute("DELETE FROM post WHERE id = ?", (post_id,))
    db.commit()
    return jsonify({"message": "Artikel gelöscht"})

API testen:

# 获取文章列表
$ curl http://127.0.0.1:5000/api/posts

# 创建文章
$ curl -X POST http://127.0.0.1:5000/api/posts \
  -H "Content-Type: application/json" \
  -d '{"title":"EXAMPLE 教程","body":"Flask 入门教程内容","author_id":1}'

# 更新文章
$ curl -X PUT http://127.0.0.1:5000/api/posts/1 \
  -H "Content-Type: application/json" \
  -d '{"title":"EXAMPLE 教程（修订版）","body":"更新后的内容"}'

# 删除文章
$ curl -X DELETE http://127.0.0.1:5000/api/posts/1

Bewährte Methoden: Wenn die API JSON-Daten zurückgibt, verwenden Sie den korrekten HTTP-Statuscode—für die Erstellung verwenden Sie201, verwenden Sie200Verwenden Sie für Client-Fehler400Verwenden Sie für nicht existierende Ressourcen404Dadurch kann Frontend-Code das Ergebnis der Anfrage leichter ermitteln.

Befehl zur Initialisierung der Datenbank

Über die Flask-CLI können Sie benutzerdefinierte Befehle erstellen, um die Datenbank zu initialisieren:

# 初始化数据库（创建表结构）
(.venv) $ flask init-db

# 验证表已创建
(.venv) $ sqlite3 instance/example.db ".tables"
post  user

Schutz vor SQL-Injection

Die Verwendung parametrisierter Abfragen (?Platzhalter) ist entscheidend, um SQL-Injection zu verhindern:

Beispiel

# Richtiger Ansatz: ?-Platzhalter verwenden (parametrisierte Abfragen)
# Der SQLite-Treiber maskiert Parameter automatisch, um zu verhindern, dass bösartige Eingaben als SQL ausgeführt werden
db.execute("SELECT * FROM user WHERE username = ?", (username,))

# Falscher Ansatz: f-String zur Verkettung von SQL verwenden (äußerst gefährlich!)
# db.execute(f"SELECT * FROM user WHERE username = '{username}'")
# Bösartige Eingabe '; DROP TABLE user; -- würde die gesamte Tabelle löschen!

Verwendung von Flask-SQLAlchemy (Erweiterte optionale Lektüre)

Bei großen Projekten wird das manuelle Schreiben von SQL mit der Zeit mühsam.

Flask-SQLAlchemyEs ist eine weit verbreitete Erweiterung, die ORM-Funktionalität (Object-Relational Mapping) bereitstellt:

Beispiel

# Installation: pip install flask-sqlalchemy
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///example.db"
db = SQLAlchemy(app)

# Datenmodell definieren (Tabellenstruktur durch eine Python-Klasse dargestellt)
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return f"<User {self.username}>"

# Tabelle erstellen
with app.app_context():
    db.create_all()

# Benutzer erstellen
user = User(username="example", email="[email protected]")
db.session.add(user)
db.session.commit()

# Einen Benutzer abfragen
users = User.query.all()
example_user = User.query.filter_by(username="example").first()

Dieses Kapitel verwendet natives SQLite, um zusätzliche Abhängigkeiten zu vermeiden, was es für Lernzwecke und kleine Projekte geeignet macht. Für echte Produktionsprojekte empfehlen wir die Verwendung von SQLAlchemy oder einem ähnlichen ORM, der wiederholten Code erheblich reduzieren und die Sicherheit verbessern kann.

Weitere Erweiterungen