Usa mssql-python con Flask

Flask es un framework web ligero en Python que te da control total sobre la estructura de la aplicación. Combinado con mssql-python, puedes crear aplicaciones web y APIs REST respaldadas por Microsoft SQL y Azure SQL Database con una sobrecarga mínima.

Prerequisites

• Python 3.10 o posterior.
• Los paquetes mssql-python y flask. Instala ambos con pip install flask mssql-python.
• Instale requisitos previos específicos del sistema operativo de un solo uso. Los usuarios de Windows pueden saltarse este paso. Para detalles completos sobre la plataforma, véase Instalar mssql-python.
  • Alpino
  • Debian/Ubuntu
  • RHEL
  • SUSE
  • openSUSE
  • macOS

  apk add libtool krb5-libs krb5-dev
  

Creación de una base de datos SQL

Crea o conéctate a una base de datos SQL en una de las siguientes plataformas:

Los ejemplos de este artículo utilizan la base de datos de ejemplo AdventureWorksLT , concretamente la SalesLT.Product tabla. Si no tienes instalado AdventureWorksLT, consulta las bases de datos de ejemplo de AdventureWorks.

Configuración del proyecto

Instalación de dependencias

Instala los paquetes necesarios con pip:

pip install flask mssql-python

Estructura del proyecto

Organiza tu proyecto con módulos separados para configuración, gestión de conexiones, rutas y pruebas:

my_app/
├── app.py            # Flask app and routes
├── config.py         # database settings
├── database.py       # connection lifecycle
├── test_app.py       # pytest tests
└── blueprints/       # optional: routes grouped into modules
    ├── __init__.py
    └── products.py

Gestión de conexiones de bases de datos

Flask no incluye una capa de base de datos integrada, así que gestionas las conexiones directamente. El patrón en esta sección almacena una conexión por solicitud en el objeto de g Flask y lo cierra automáticamente cuando termina la solicitud.

Crear config.py

Centraliza la configuración de la base de datos en una clase de configuración. Las variables de entorno te permiten anular los valores predeterminados sin cambiar el código.

# config.py
import os

class Config:
    """Application configuration."""
    DATABASE_SERVER = os.getenv("DB_SERVER", "<server>.database.windows.net")
    DATABASE_NAME = os.getenv("DB_NAME", "<database>")
    POOL_SIZE = int(os.getenv("DB_POOL_SIZE", "10"))

Cree database.py

El database.py módulo gestiona el ciclo de vida de la conexión. El objeto g de Flask es un espacio de nombres específico de cada solicitud, por lo que guardar allí la conexión garantiza que cada solicitud tenga su propia conexión, que se cierra y limpia al finalizar la solicitud.

La get_connection_string() función construye la cadena de conexión desde la configuración de la app. La get_db() función crea una conexión en la primera llamada y la reutiliza para el resto de la petición. La función close_db() se ejecuta automáticamente al final de cada petición, deshaciendo la transacción si se produjo una excepción y confirmándola en caso contrario. La init_app() función detecta este comportamiento de desmontaje con la app Flask.

# database.py
import mssql_python
from flask import g, current_app

def get_connection_string() -> str:
    """Build connection string from Flask app config."""
    cfg = current_app.config
    return (
        f"Server={cfg['DATABASE_SERVER']};"
        f"Database={cfg['DATABASE_NAME']};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

def get_db():
    """Get a database cursor for the current request.

    The connection is stored on Flask's g object so it persists
    for the duration of the request and is reused across calls.
    """
    if "db_conn" not in g:
        g.db_conn = mssql_python.connect(get_connection_string())
        g.db_cursor = g.db_conn.cursor()
    return g.db_cursor

def close_db(exception=None):
    """Close the database connection at the end of the request."""
    cursor = g.pop("db_cursor", None)
    conn = g.pop("db_conn", None)

    if cursor is not None:
        cursor.close()
    if conn is not None:
        if exception:
            conn.rollback()
        else:
            conn.commit()
        conn.close()

def init_app(app):
    """Register database teardown with the Flask app."""
    app.teardown_appcontext(close_db)

Aplicación Flask

El siguiente ejemplo muestra una aplicación completa de Flask con rutas para listar, recuperar, crear, actualizar y eliminar productos.

Crea app.py

El módulo de aplicación crea la app Flask, carga la configuración y registra el desmontaje de la base de datos. Cada función de ruta llama a get_db() para obtener un cursor, ejecuta consultas con SQL parametrizado (usando los marcadores de posición %(name)s y un diccionario de valores) y devuelve respuestas JSON.

# app.py
from flask import Flask, jsonify, request, abort
from config import Config
from database import init_app, get_db

app = Flask(__name__)
app.config.from_object(Config)
init_app(app)

@app.route("/")
def index():
    return jsonify({"message": "Product API", "docs": "/products"})

@app.route("/products")
def list_products():
    """List products with pagination."""
    page = request.args.get("page", 1, type=int)
    page_size = request.args.get("page_size", 10, type=int)
    skip = (page - 1) * page_size

    cursor = get_db()

    cursor.execute("SELECT COUNT(*) FROM SalesLT.Product")
    total = cursor.fetchval()

    cursor.execute("""
        SELECT ProductID, Name, ProductNumber, ListPrice, Color, ProductCategoryID
        FROM SalesLT.Product
        ORDER BY ProductID
        OFFSET %(skip)s ROWS
        FETCH NEXT %(limit)s ROWS ONLY
    """, {"skip": skip, "limit": page_size})

    items = [{
        "id": row.ProductID,
        "name": row.Name,
        "product_number": row.ProductNumber,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    } for row in cursor.fetchall()]

    return jsonify({
        "items": items,
        "total": total,
        "page": page,
        "page_size": page_size,
        "pages": (total + page_size - 1) // page_size
    })

@app.route("/products/<int:product_id>")
def get_product(product_id):
    """Get a single product by ID."""
    cursor = get_db()
    cursor.execute("""
        SELECT ProductID, Name, ProductNumber, ListPrice, Color, ProductCategoryID
        FROM SalesLT.Product
        WHERE ProductID = %(id)s
    """, {"id": product_id})

    row = cursor.fetchone()
    if not row:
        abort(404)

    return jsonify({
        "id": row.ProductID,
        "name": row.Name,
        "product_number": row.ProductNumber,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    })

@app.route("/products", methods=["POST"])
def create_product():
    """Create a new product."""
    data = request.get_json()
    if not data:
        abort(400)

    cursor = get_db()

    # OUTPUT INSERTED returns the new row's columns in the same statement,
    # so you don't need a separate SELECT to get the generated ID and defaults.
    # ProductNumber is required and unique. StandardCost and SellStartDate are
    # also NOT NULL in SalesLT.Product, so supply values for them.
    cursor.execute("""
        INSERT INTO SalesLT.Product
            (Name, ProductNumber, ListPrice, Color, Size, ProductCategoryID, StandardCost, SellStartDate)
        OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber, INSERTED.ListPrice,
               INSERTED.Color, INSERTED.ProductCategoryID
        VALUES (%(name)s, %(product_number)s, %(price)s, %(color)s, %(size)s, %(category_id)s, 0, GETDATE())
    """, {
        "name": data["name"],
        "product_number": data["product_number"],
        "price": data["price"],
        "color": data.get("color"),
        "size": data.get("size"),
        "category_id": data["category_id"]
    })

    row = cursor.fetchone()
    return jsonify({
        "id": row.ProductID,
        "name": row.Name,
        "product_number": row.ProductNumber,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    }), 201

@app.route("/products/<int:product_id>", methods=["PUT"])
def update_product(product_id):
    """Update an existing product."""
    data = request.get_json()
    if not data:
        abort(400)

    cursor = get_db()

    updates = []
    params = {"id": product_id}

    for field in ("name", "product_number", "price", "color", "category_id"):
        if field in data:
            col = {"name": "Name", "product_number": "ProductNumber",
                   "price": "ListPrice", "color": "Color",
                   "category_id": "ProductCategoryID"}[field]
            updates.append(f"{col} = %({field})s")
            params[field] = data[field]

    if not updates:
        abort(400)

    cursor.execute(f"""
        UPDATE SalesLT.Product SET {', '.join(updates)}
        OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber, INSERTED.ListPrice,
               INSERTED.Color, INSERTED.ProductCategoryID
        WHERE ProductID = %(id)s
    """, params)

    row = cursor.fetchone()
    if not row:
        abort(404)

    return jsonify({
        "id": row.ProductID,
        "name": row.Name,
        "product_number": row.ProductNumber,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    })

@app.route("/products/<int:product_id>", methods=["DELETE"])
def delete_product(product_id):
    """Delete a product."""
    cursor = get_db()
    cursor.execute("DELETE FROM SalesLT.Product WHERE ProductID = %(id)s", {"id": product_id})
    if cursor.rowcount == 0:
        abort(404)
    return "", 204

@app.route("/health")
def health_check():
    """Check database connectivity."""
    try:
        cursor = get_db()
        cursor.execute("SELECT 1")
        return jsonify({"status": "healthy", "database": "connected"})
    except Exception as e:
        return jsonify({"status": "unhealthy", "error": str(e)}), 503

Ejecutar la aplicación

Inicie el servidor de desarrollo:

flask --app app run --debug --port 5000

El servidor escucha en http://localhost:5000. Abre un segundo terminal y llama a los endpoints usando curl para confirmar que la aplicación se está comunicando con tu base de datos:

# Check database connectivity
curl http://localhost:5000/health

# List the first page of products
curl "http://localhost:5000/products?page_size=5"

# Get a single product by ID
curl http://localhost:5000/products/680

Cada endpoint devuelve JSON. También puedes abrir http://localhost:5000/products en un navegador para ver la lista paginada.

Agrupación de conexiones

Sin pooling de conexiones, cada solicitud abre y cierra una conexión TCP a Microsoft SQL, lo que añade latencia. La agrupación de conexiones mantiene un conjunto de conexiones ociosas listas para reutilizarse. Para habilitar la agrupación de conexiones, llama a mssql_python.pooling() una vez a nivel de módulo. Con la agrupación activada, conn.close() durante el desmontaje de close_db devuelve la conexión al grupo en lugar de cerrarla.

Habilitación de la agrupación de conexiones

Habilitar la agrupación llamando mssql_python.pooling() a nivel de módulo antes de abrir cualquier conexión:

# database.py with connection pooling
import mssql_python
from flask import g, current_app

# Configure pool at module level
mssql_python.pooling(max_size=20, idle_timeout=300)

def get_db():
    """Get a database cursor with connection pooling."""
    if "db_conn" not in g:
        g.db_conn = mssql_python.connect(get_connection_string())
        g.db_cursor = g.db_conn.cursor()
    return g.db_cursor

Gestión de errores

Flask te permite registrar manejadores para tipos específicos de excepciones. Capturar mssql_python.DatabaseError y mssql_python.IntegrityError permite devolver respuestas de error estructuradas en formato JSON en lugar de las páginas de error HTML por defecto.

Registrar controladores de errores

Añade estos controladores al app.py existente, después de la línea app = Flask(__name__). Como los manejadores hacen referencia al app objeto, deben venir después de que la app esté creada. app.py necesita import mssql_python arriba. Los manejadores devolven respuestas JSON estructuradas en lugar de páginas de error HTML por defecto:

# app.py
import mssql_python

@app.errorhandler(mssql_python.DatabaseError)
def handle_database_error(error):
    """Handle database errors."""
    return jsonify({"error": "Database error occurred"}), 500

@app.errorhandler(mssql_python.IntegrityError)
def handle_integrity_error(error):
    """Handle integrity constraint violations."""
    error_msg = str(error)
    if "UNIQUE" in error_msg:
        return jsonify({"error": "Resource already exists"}), 409
    if "FOREIGN KEY" in error_msg:
        return jsonify({"error": "Referenced resource not found"}), 400
    return jsonify({"error": "Data integrity error"}), 400

@app.errorhandler(404)
def not_found(error):
    return jsonify({"error": "Resource not found"}), 404

@app.errorhandler(400)
def bad_request(error):
    return jsonify({"error": "Bad request"}), 400

Blueprints

A medida que tu aplicación crece, poner todas las rutas en un solo archivo se vuelve difícil de mantener. Flask Blueprints te permite agrupar rutas relacionadas en módulos separados que se registran en la app.

Organizar rutas con planos

Crea un módulo blueprint para las rutas del producto que importe get_db y defina puntos de conexión con un prefijo de URL compartido:

# blueprints/products.py
from flask import Blueprint, jsonify, request, abort
from database import get_db

products_bp = Blueprint("products", __name__, url_prefix="/api/products")

@products_bp.route("/")
def list_products():
    """List all products."""
    cursor = get_db()
    cursor.execute("""
        SELECT ProductID, Name, ListPrice, Color, ProductCategoryID
        FROM SalesLT.Product ORDER BY ProductID
    """)
    return jsonify([{
        "id": row.ProductID,
        "name": row.Name,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    } for row in cursor.fetchall()])

@products_bp.route("/<int:product_id>")
def get_product(product_id):
    """Get a product by ID."""
    cursor = get_db()
    cursor.execute(
        "SELECT ProductID, Name, ListPrice, Color FROM SalesLT.Product WHERE ProductID = %(id)s",
        {"id": product_id}
    )
    row = cursor.fetchone()
    if not row:
        abort(404)
    return jsonify({"id": row.ProductID, "name": row.Name, "price": float(row.ListPrice), "color": row.Color})

Registrar el plano

Guarda el blueprint como blueprints/products.py, y añade un archivo vacío blueprints/__init__.py para que Python trate la carpeta como un paquete. Luego, en app.py, importa el blueprint con tus otras importaciones y regístralo después de la app = Flask(__name__) línea:

# app.py
from blueprints.products import products_bp

app.register_blueprint(products_bp)

Como el blueprint define url_prefix="/api/products", sus rutas se sirven bajo ese prefijo. Por ejemplo, la ruta de lista está disponible en http://localhost:5000/api/products/, separada de las /products rutas definidas directamente en app.py.

Testing

Flask proporciona un cliente de prueba que envía solicitudes a tu aplicación sin iniciar un servidor HTTP real. Usa pytest fixtures para crear el cliente y reutilízalo en todas las pruebas.

Configuración de pruebas con pytest

Crea un dispositivo pytest que proporcione un cliente de prueba y escribe pruebas para verificar el comportamiento de las rutas:

# test_app.py
import uuid

import pytest
from app import app

@pytest.fixture
def client():
    app.config["TESTING"] = True
    with app.test_client() as client:
        yield client

def test_health_check(client):
    response = client.get("/health")
    assert response.status_code == 200
    data = response.get_json()
    assert data["status"] == "healthy"

def test_list_products(client):
    response = client.get("/products")
    assert response.status_code == 200
    data = response.get_json()
    assert "items" in data
    assert "total" in data

def test_create_product(client):
    suffix = uuid.uuid4().hex[:8]
    name = f"Test Product {suffix}"
    response = client.post("/products", json={
        "name": name,
        "product_number": f"TEST-{suffix}",
        "price": 19.99,
        "category_id": 18
    })
    assert response.status_code == 201
    data = response.get_json()
    assert data["name"] == name

def test_get_product_not_found(client):
    response = client.get("/products/99999")
    assert response.status_code == 404

Estas pruebas se ejecutan contra tu base de datos en vivo en lugar de mocks, así que test_create_product inserta una fila real en SalesLT.Product. En AdventureWorksLT, tanto Name como ProductNumber tienen restricciones únicas, por lo que la prueba genera un valor único para cada una en cada partida. Si codificas esos valores de forma fija, la prueba falla con un conflicto en la segunda ejecución a menos que elimines primero la fila.

Ejecución de las pruebas

Guarda las pruebas en la test_app.py carpeta de tu proyecto. Con tu entorno virtual activado, instálalo pytest y ejecuta desde esa carpeta. Instalar y ejecutar pytest dentro del mismo entorno virtual que flask y mssql-python garantiza que las pruebas importen los paquetes que utiliza tu app. pytest Descubre test_app.py y reporta automáticamente los resultados:

pip install pytest
pytest

pytest descubre test_app.py automáticamente y reporta los resultados:

==================== test session starts ====================
collected 4 items

test_app.py ....                                       [100%]

===================== 4 passed in 3.21s =====================

Contenido relacionado

• Administración de conexiones
• Agrupación de conexiones
• Control de errores
• Integración con FastAPI
• Autenticación de Microsoft Entra