Usa mssql-python con FastAPI

FastAPI es un framework web moderno en Python para construir APIs. Combinado con mssql-python, puedes construir APIs REST de alto rendimiento respaldadas por Microsoft SQL y Azure SQL Database.

Prerequisites

• Python 3.10 o posterior.
• Los paquetes mssql-python, fastapi, uvicorn, pydantic y PyJWT. Instala todo con pip install fastapi uvicorn mssql-python pydantic pyjwt.
• 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

Creación de un entorno virtual

Crea y activa un entorno virtual para que los paquetes de este proyecto permanezcan aislados de otras instalaciones de Python. Este paso también evita el problema común de instalar paquetes en un intérprete mientras ejecutas tu aplicación o pruebas con otro.

py -m venv .venv
.\.venv\Scripts\Activate.ps1

python3 -m venv .venv
source .venv/bin/activate

Después de activar el entorno, python, pip y pytest, todos apuntan al mismo intérprete. Ejecuta los comandos restantes de este artículo desde el entorno activado.

Instalación de dependencias

Instala los paquetes necesarios con pip:

pip install fastapi uvicorn mssql-python pydantic pyjwt

Estructura del proyecto

Organiza tu proyecto con módulos separados para bases de datos, esquemas y operaciones CRUD:

my_api/
├── main.py
├── database.py
├── models.py
├── schemas.py
├── crud.py
├── test_api.py
└── routers/
    └── products.py

Gestión de conexiones de bases de datos

FastAPI utiliza inyección de dependencias para proporcionar recursos como conexiones a bases de datos a los manejadores de rutas. El patrón que se muestra en esta sección crea un gestor de contexto que abre una conexión, cede un cursor y confirma o revierte la transacción y cierra la conexión automáticamente.

Cree database.py

La get_connection_string() función construye la cadena de conexión ODBC a partir de valores de configuración. El gestor de contexto get_db() y el generador get_db_dependency() siguen ambos el mismo patrón: abrir una conexión, devolver un cursor, confirmar la transacción en caso de éxito, deshacerla en caso de error y cerrarla siempre al terminar. FastAPI llama Depends()get_db_dependency() una vez por solicitud y gestiona su ciclo de vida.

# database.py
import mssql_python
from contextlib import contextmanager
from typing import Generator

# Configuration
DATABASE_CONFIG = {
    "server": "<server>.database.windows.net",
    "database": "<database>",
}

def get_connection_string() -> str:
    """Build connection string from config."""
    return (
        f"Server={DATABASE_CONFIG['server']};"
        f"Database={DATABASE_CONFIG['database']};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

@contextmanager
def get_db() -> Generator:
    """Database connection context manager for FastAPI dependency injection."""
    conn = mssql_python.connect(get_connection_string())
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()

def get_db_dependency():
    """FastAPI dependency for database cursor."""
    conn = mssql_python.connect(get_connection_string())
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()

Modelos de Pydantic

Los modelos pydânticos definen las reglas de forma y validación para los datos de petición y respuesta. FastAPI utiliza estos modelos para analizar JSON entrante, validar restricciones de campos y generar automáticamente documentación OpenAPI.

Cree schemas.py

Separar los esquemas en Base, Create, Update, y variantes de respuesta. El Base esquema contiene campos compartidos, Create hereda de él para las operaciones de inserción y Update hace que todos los campos sean opcionales para actualizaciones parciales.

# schemas.py
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from typing import Optional
from datetime import datetime

# Product schemas
class ProductBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    product_number: str = Field(..., min_length=1, max_length=25)
    price: float = Field(..., gt=0)
    color: Optional[str] = Field(None, max_length=50)
    size: Optional[str] = Field(None, max_length=50)
    category_id: Optional[int] = None

class ProductCreate(ProductBase):
    pass

class ProductUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    product_number: Optional[str] = Field(None, min_length=1, max_length=25)
    price: Optional[float] = Field(None, gt=0)
    color: Optional[str] = Field(None, max_length=50)
    size: Optional[str] = Field(None, max_length=50)
    category_id: Optional[int] = None

class Product(ProductBase):
    id: int

    model_config = ConfigDict(from_attributes=True)

# Pagination
class PaginatedResponse(BaseModel):
    items: list
    total: int
    page: int
    page_size: int
    pages: int

Operaciones CRUD

Encapsula las consultas a la base de datos en una clase dedicada para mantener los controladores de ruta ligeros. Cada método estático toma un cursor (inyectado por FastAPI) y gestiona una operación usando consultas parametrizadas (%(name)s marcadores de posición con un diccionario de valores) para evitar la inyección SQL. Esta separación facilita la prueba y reutilización de la lógica de negocio.

Cree crud.py

# crud.py
from typing import Optional, List
from schemas import ProductCreate, ProductUpdate, Product

class ProductCRUD:
    """CRUD operations for products."""
    
    @staticmethod
    def get(cursor, product_id: int) -> Optional[dict]:
        cursor.execute("""
            SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
            FROM SalesLT.Product
            WHERE ProductID = %(id)s
        """, {"id": product_id})
        
        row = cursor.fetchone()
        if row:
            return {
                "id": row.ProductID,
                "name": row.Name,
                "product_number": row.ProductNumber,
                "price": float(row.ListPrice),
                "color": row.Color,
                "size": row.Size
            }
        return None
    
    @staticmethod
    def get_all(cursor, skip: int = 0, limit: int = 100) -> List[dict]:
        cursor.execute("""
            SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
            FROM SalesLT.Product
            ORDER BY ProductID
            OFFSET %(skip)s ROWS
            FETCH NEXT %(limit)s ROWS ONLY
        """, {"skip": skip, "limit": limit})
        
        return [{
            "id": row.ProductID,
            "name": row.Name,
            "product_number": row.ProductNumber,
            "price": float(row.ListPrice),
            "color": row.Color,
            "size": row.Size
        } for row in cursor.fetchall()]
    
    @staticmethod
    def count(cursor) -> int:
        cursor.execute("SELECT COUNT(*) FROM SalesLT.Product")
        return cursor.fetchval()
    
    @staticmethod
    def create(cursor, product: ProductCreate) -> dict:
        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.Size
            VALUES (%(name)s, %(product_number)s, %(price)s, %(color)s, %(size)s, %(category_id)s, 0, GETDATE())
        """, {
            "name": product.name,
            "product_number": product.product_number,
            "price": product.price,
            "color": product.color,
            "size": product.size,
            "category_id": product.category_id
        })
        
        row = cursor.fetchone()
        return {
            "id": row.ProductID,
            "name": row.Name,
            "product_number": row.ProductNumber,
            "price": float(row.ListPrice),
            "color": row.Color,
            "size": row.Size
        }
    
    @staticmethod
    def update(cursor, product_id: int, product: ProductUpdate) -> Optional[dict]:
        # Build dynamic update
        updates = []
        params = {"id": product_id}
        
        if product.name is not None:
            updates.append("Name = %(name)s")
            params["name"] = product.name
        if product.product_number is not None:
            updates.append("ProductNumber = %(product_number)s")
            params["product_number"] = product.product_number
        if product.price is not None:
            updates.append("ListPrice = %(price)s")
            params["price"] = product.price
        if product.category_id is not None:
            updates.append("ProductCategoryID = %(category_id)s")
            params["category_id"] = product.category_id
        
        if not updates:
            return ProductCRUD.get(cursor, product_id)
        
        cursor.execute(f"""
            UPDATE SalesLT.Product SET {', '.join(updates)}
            OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber,
                   INSERTED.ListPrice, INSERTED.Color, INSERTED.Size
            WHERE ProductID = %(id)s
        """, params)
        
        row = cursor.fetchone()
        if row:
            return {
                "id": row.ProductID,
                "name": row.Name,
                "product_number": row.ProductNumber,
                "price": float(row.ListPrice),
                "color": row.Color,
                "size": row.Size
            }
        return None
    
    @staticmethod
    def delete(cursor, product_id: int) -> bool:
        cursor.execute("""
            DELETE FROM SalesLT.Product WHERE ProductID = %(id)s
        """, {"id": product_id})
        return cursor.rowcount > 0
    
    @staticmethod
    def search(cursor, query: str, skip: int = 0, limit: int = 100) -> List[dict]:
        cursor.execute("""
            SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
            FROM SalesLT.Product
            WHERE Name LIKE %(query)s OR ProductNumber LIKE %(query)s
            ORDER BY ProductID
            OFFSET %(skip)s ROWS
            FETCH NEXT %(limit)s ROWS ONLY
        """, {"query": f"%{query}%", "skip": skip, "limit": limit})
        
        return [{
            "id": row.ProductID,
            "name": row.Name,
            "product_number": row.ProductNumber,
            "price": float(row.ListPrice),
            "color": row.Color,
            "size": row.Size
        } for row in cursor.fetchall()]

Aplicación FastAPI

Crear main.py

El módulo principal conecta todo. Cada ruta declara cursor = Depends(get_db_dependency), lo que le indica a FastAPI que llame al generador, pase el cursor generado al manejador y realice la limpieza después. FastAPI también valida los cuerpos de solicitud conforme a tus esquemas de Pydantic antes de que se ejecute el controlador.

# main.py
from fastapi import FastAPI, HTTPException, Depends, Query
from typing import List
from database import get_db_dependency
from schemas import Product, ProductCreate, ProductUpdate, PaginatedResponse
from crud import ProductCRUD

app = FastAPI(
    title="Product API",
    description="REST API for products using mssql-python",
    version="1.0.0"
)

@app.get("/")
def root():
    return {"message": "Product API", "docs": "/docs"}

@app.get("/products", response_model=PaginatedResponse)
def list_products(
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    cursor = Depends(get_db_dependency)
):
    """List all products with pagination."""
    skip = (page - 1) * page_size
    items = ProductCRUD.get_all(cursor, skip=skip, limit=page_size)
    total = ProductCRUD.count(cursor)
    
    return {
        "items": items,
        "total": total,
        "page": page,
        "page_size": page_size,
        "pages": (total + page_size - 1) // page_size
    }

@app.get("/products/{product_id}", response_model=Product)
def get_product(product_id: int, cursor = Depends(get_db_dependency)):
    """Get a specific product by ID."""
    product = ProductCRUD.get(cursor, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

@app.post("/products", response_model=Product, status_code=201)
def create_product(product: ProductCreate, cursor = Depends(get_db_dependency)):
    """Create a new product."""
    return ProductCRUD.create(cursor, product)

@app.put("/products/{product_id}", response_model=Product)
def update_product(
    product_id: int,
    product: ProductUpdate,
    cursor = Depends(get_db_dependency)
):
    """Update an existing product."""
    updated = ProductCRUD.update(cursor, product_id, product)
    if not updated:
        raise HTTPException(status_code=404, detail="Product not found")
    return updated

@app.delete("/products/{product_id}", status_code=204)
def delete_product(product_id: int, cursor = Depends(get_db_dependency)):
    """Delete a product."""
    if not ProductCRUD.delete(cursor, product_id):
        raise HTTPException(status_code=404, detail="Product not found")

@app.get("/products/search/", response_model=List[Product])
def search_products(
    q: str = Query(..., min_length=1),
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    cursor = Depends(get_db_dependency)
):
    """Search products by name or product number."""
    skip = (page - 1) * page_size
    return ProductCRUD.search(cursor, q, skip=skip, limit=page_size)

# Health check endpoint
@app.get("/health")
def health_check(cursor = Depends(get_db_dependency)):
    """Check database connectivity."""
    try:
        cursor.execute("SELECT 1")
        return {"status": "healthy", "database": "connected"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"Database unhealthy: {str(e)}")

Ejecutar la aplicación

uvicorn main:app --reload --host 0.0.0.0 --port 8000

Gestión de errores

FastAPI te permite registrar gestores globales de excepciones para tipos específicos de excepciones. Cuando detectas mssql_python.DatabaseError y mssql_python.IntegrityError, FastAPI devuelve errores JSON estructurados con códigos de estado HTTP apropiados en lugar de respuestas genéricas 500.

Gestor global de excepciones

Añade estos controladores a main.py, justo después de la línea app = FastAPI(...). FastAPI ejecuta el manejador correspondiente cada vez que una ruta eleva ese tipo de excepción, así que no necesitas un try/except bloque en cada ruta.

# main.py
from fastapi import Request
from fastapi.responses import JSONResponse
import mssql_python

@app.exception_handler(mssql_python.DatabaseError)
async def database_exception_handler(request: Request, exc: mssql_python.DatabaseError):
    """Handle database errors globally."""
    return JSONResponse(
        status_code=500,
        content={"detail": "Database error occurred", "type": "database_error"}
    )

@app.exception_handler(mssql_python.IntegrityError)
async def integrity_exception_handler(request: Request, exc: mssql_python.IntegrityError):
    """Handle integrity constraint violations."""
    error_msg = str(exc)
    
    if "UNIQUE" in error_msg:
        return JSONResponse(
            status_code=409,
            content={"detail": "Resource already exists", "type": "duplicate_error"}
        )
    elif "FOREIGN KEY" in error_msg:
        return JSONResponse(
            status_code=400,
            content={"detail": "Referenced resource not found", "type": "reference_error"}
        )
    
    return JSONResponse(
        status_code=400,
        content={"detail": "Data integrity error", "type": "integrity_error"}
    )

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. Llama a mssql_python.pooling() una vez al inicio. Con el pooling activado, conn.close() devuelve get_db_dependency() la conexión al pool en vez de cerrarlo realmente.

Módulo de base de datos mejorado

Activa el pooling llamando mssql_python.pooling() al inicio y configúralo con el tamaño máximo y la configuración de tiempo de espera adecuados:

# database.py with connection pooling
import mssql_python
from contextlib import contextmanager
import os

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

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "Server=<server>.database.windows.net;Database=<database>;"
    "Authentication=ActiveDirectoryDefault;Encrypt=yes"
)

def get_db_dependency():
    """FastAPI dependency with connection pooling."""
    conn = mssql_python.connect(DATABASE_URL)
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()  # Returns to pool

Middleware de autenticación

Puedes combinar el acceso a la base de datos con la autenticación encadenando dependencias de FastAPI. El siguiente ejemplo valida un token portador JWT, busca el registro de persona correspondiente en la base de datos de ejemplo AdventureWorksLT y pone el resultado a disposición de rutas protegidas.

# auth.py
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    cursor = Depends(get_db_dependency)
):
    """Validate JWT and return the matching AdventureWorksLT person."""
    try:
        token = credentials.credentials
        # Replace with a strong secret loaded from environment variables
        payload = jwt.decode(token, "your-secret-key", algorithms=["HS256"])
        person_id = int(payload.get("sub"))
        
        if not person_id:
            raise HTTPException(status_code=401, detail="Invalid token")
        
        cursor.execute("""
            SELECT BusinessEntityID, FirstName, LastName
            FROM Person.Person
            WHERE BusinessEntityID = %(id)s
        """, {"id": person_id})
        
        person = cursor.fetchone()
        if not person:
            raise HTTPException(status_code=401, detail="User not found")
        
        return {
            "id": person.BusinessEntityID,
            "first_name": person.FirstName,
            "last_name": person.LastName
        }
        
    except (TypeError, ValueError):
        raise HTTPException(status_code=401, detail="Invalid token subject")
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

# Protected endpoint
@app.get("/me")
def get_me(current_user: dict = Depends(get_current_user)):
    return current_user

Testing

FastAPI proporciona un TestClient basado en httpx que envía solicitudes a tu aplicación sin iniciar un servidor HTTP real. Escribe pruebas con pytest para verificar rutas, códigos de estado y formas de respuesta.

Antes de ejecutar las pruebas de esta sección, instala las dependencias de prueba:

pip install pytest httpx

Configuración de pruebas

Crea un archivo de prueba que se utilice TestClient para verificar el comportamiento de rutas y los esquemas de respuesta:

# test_api.py
from fastapi.testclient import TestClient
from main import app
import uuid
import pytest

client = TestClient(app)

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

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

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

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

Ejecuta las pruebas desde pytest la raíz del proyecto, el mismo directorio que main.py:

pytest

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.

Configuración de la implementación

Usa el BaseSettings de Pydantic para cargar la configuración a partir de variables de entorno y archivos .env. Este enfoque mantiene los secretos fuera del código fuente y facilita el cambio entre entornos. Instala el paquete de configuración con pip install pydantic-settings.

Variables de entorno

Crea un módulo de configuración que cargue la configuración desde variables de entorno, permitiéndote gestionar secretos y valores específicos de despliegue fuera de tu código:

# config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    database_server: str = "<server>.database.windows.net"
    database_name: str = "<database>"
    pool_size: int = 10

    model_config = SettingsConfigDict(env_file=".env")

settings = Settings()

def get_connection_string() -> str:
    return (
        f"Server={settings.database_server};"
        f"Database={settings.database_name};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

Luego actualiza database.py para importar get_connection_string desde config en lugar de definir su propia copia. Al eliminar la función duplicada, te aseguras de que la app lea los ajustes de conexión de una única fuente.

# database.py
from config import get_connection_string

Contenido relacionado

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