Integración de base de datos en Flask

La mayoría de las aplicaciones web necesitan almacenar datos de forma persistente.

Este capítulo utiliza la base de datos SQLite incorporada en Python para mostrar cómo integrar una base de datos en Flask, sin necesidad de instalar dependencias adicionales.

Inicio rápido con SQLite

SQLite es una base de datos ligera que almacena los datos en un solo archivo. La biblioteca estándar de Python incluye soporte para ella.

Es la mejor opción para aprender y para el desarrollo de prototipos: no requiere instalar un servidor de base de datos ni configuración.

Ejemplo

# Un ejemplo simple de SQLite (sin usar Flask)
import sqlite3

# Conectar a la base de datos (se crea automáticamente si ejemplo.db no existe)
conn = sqlite3.connect("ejemplo.db")
# row_factory permite acceder a los resultados de la consulta por nombre de campo, como un diccionario
conn.row_factory = sqlite3.Row

# Crear tabla
conn.execute("CREATE TABLE IF NOT EXISTS user (id INTEGER PRIMARY KEY, username TEXT UNIQUE, password TEXT)")
# Insertar datos
conn.execute("INSERT INTO user (username, password) VALUES (?, ?)", ("ejemplo", "pass123"))
conn.commit()

# Consultar datos
rows = conn.execute("SELECT * FROM user").fetchall()
for row in rows:
    print(dict(row))  # Salida: {'id': 1, 'username': 'ejemplo', 'password': 'pass123'}

conn.close()

Usar el objeto g en Flask para gestionar conexiones de base de datos

gEs un objeto especial cuyo ciclo de vida es una sola petición.

Guardar la conexión de base de datos engpuede garantizar que las múltiples operaciones de base de datos dentro de la misma petición compartan la misma conexión y se limpien automáticamente al finalizar la petición.

Ejemplo

# Ruta del archivo: db.py (módulo de gestión de base de datos)
import sqlite3
from flask import g, current_app

def get_db():
    """Obtener la conexión de base de datos, reutilizada dentro de la misma petición"""
    if "db" not in g:
        # Leer la ruta de la base de datos desde la configuración
        db_path = current_app.config.get("DATABASE", "ejemplo.db")
        g.db = sqlite3.connect(db_path)
        g.db.row_factory = sqlite3.Row  # Hacer que los resultados admitan acceso por nombre de campo
    return g.db

def close_db(error=None):
    """Cerrar la conexión de base de datos (se llama automáticamente al final de la petición)"""
    db = g.pop("db", None)
    if db is not None:
        db.close()

def init_db():
    """Inicializar la estructura de las tablas de la base de datos"""
    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):
    """Registrar funciones relacionadas con la base de datos en la aplicación Flask"""
    # Cerrar automáticamente la conexión de base de datos al final de la petición
    app.teardown_appcontext(close_db)
    # Registrar comando CLI: flask init-db
    app.cli.command("init-db")(init_db)

Integración con el patrón de fábrica

Usar el patrón de fábrica para integrar la base de datos con otros módulos:

Ejemplo

# Ruta del archivo: app.py
from flask import Flask

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

    # Configurar la ruta de la base de datos (almacenada en la carpeta instance)
    import os
    app.config["DATABASE"] = os.path.join(app.instance_path, "ejemplo.db")

    # Asegurarse de que la carpeta instance exista
    os.makedirs(app.instance_path, exist_ok=True)

    # Inicializar la base de datos
    from db import init_app as init_db
    init_db(app)

    # Registrar el blueprint
    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

Escribir una API de blog completa

Combinando con la base de datos, implementar las operaciones CRUD (crear, leer, actualizar, eliminar) de artículos:

Ejemplo

# Ruta del archivo: 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():
    """Obtener la lista de todos los artículos"""
    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()
    # Convertir la lista de objetos Row en una lista de diccionarios
    return jsonify([dict(post) for post in posts])

@bp.get("/api/posts/<int:post_id>")
def get_post(post_id):
    """Obtener los detalles de un solo artículo"""
    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"El artículo {post_id} no existe")

    return jsonify(dict(post))

@bp.post("/api/posts")
def create_post():
    """Crear un nuevo artículo"""
    data = request.json
    title = data.get("title", "").strip()
    body = data.get("body", "").strip()
    author_id = data.get("author_id", 1)  # ID de autor predeterminado

    if not title:
        return jsonify({"error": "El título no puede estar vacío"}), 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": "Artículo creado con éxito"}), 201

@bp.put("/api/posts/<int:post_id>")
def update_post(post_id):
    """Actualizar artículo"""
    data = request.json
    title = data.get("title", "").strip()
    body = data.get("body", "").strip()

    if not title:
        return jsonify({"error": "El título no puede estar vacío"}), 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": "Artículo actualizado con éxito"})

@bp.delete("/api/posts/<int:post_id>")
def delete_post(post_id):
    """Eliminar artículo"""
    db = get_db()
    db.execute("DELETE FROM post WHERE id = ?", (post_id,))
    db.commit()
    return jsonify({"message": "Artículo eliminado"})

Probar la API:

# 获取文章列表
$ 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":"EJEMPLO 教程","body":"Flask 入门教程内容","author_id":1}'

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

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

Mejores prácticas: cuando la API devuelve datos JSON, usa los códigos de estado HTTP correctos: para creación usa201, para éxito usa200, para errores del cliente usa400, para recursos inexistentes usa404. Esto permite que el código del frontend determine el resultado de la solicitud más fácilmente.

Comando de inicialización de base de datos

A través de la CLI de Flask se pueden crear comandos personalizados para inicializar la base de datos:

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

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

Protección contra inyección SQL

Usar consultas parametrizadas (?el marcador de posición) es la clave para prevenir la inyección SQL:

Ejemplo

# Forma correcta: usar el marcador de posición ? (consulta parametrizada)
# El controlador de SQLite escapa automáticamente los parámetros, evitando que entradas maliciosas se ejecuten como SQL
db.execute("SELECT * FROM user WHERE username = ?", (username,))

# Forma incorrecta: usar f-string para concatenar SQL (¡extremadamente peligroso!)
# db.execute(f"SELECT * FROM user WHERE username = '{username}'")
# La entrada maliciosa '; DROP TABLE user; -- eliminaría toda la tabla!

Usar Flask-SQLAlchemy (lectura avanzada opcional)

Para proyectos grandes, escribir SQL manualmente se vuelve cada vez más tedioso.

Flask-SQLAlchemyes una extensión muy popular que proporciona capacidades ORM (mapeo objeto-relacional):

Ejemplo

# Instalación: pip install flask-sqlalchemy
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

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

# Definir el modelo de datos (la estructura de la tabla se representa con clases de Python)
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}>"

# Crear tabla
with app.app_context():
    db.create_all()

# Crear usuario
user = User(username="ejemplo", email="[email protected]")
db.session.add(user)
db.session.commit()

# Consultar usuario
users = User.query.all()
ejemplo_user = User.query.filter_by(username="ejemplo").first()

Este capítulo utiliza SQLite nativo para no introducir dependencias adicionales, adecuado para el aprendizaje y proyectos pequeños. Para proyectos de producción reales, se recomienda usar SQLAlchemy o un ORM similar, lo que reduce significativamente el código repetido y mejora la seguridad.

Otras extensiones