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.
Instala los requisitos previos específicos del sistema operativo, que solo hay que instalar una vez. 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
```
```
apt-get install -y libltdl7 libkrb5-3 libgssapi-krb5-2
```
```
dnf install -y libtool-ltdl krb5-libs
```
```
zypper install -y libltdl7 libkrb5-3 libgssapi-krb5-2
```
```
zypper install -y libltdl7
```
```
brew install openssl
```

Creación de una base de datos SQL

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

Creación de una base de datos SQL en cuestión de minutos mediante Azure Portal

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.

Windows
Linux/macOS

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.

Note

En Windows para Arm, cree el entorno con una compilación Arm64 de Python para que mssql-python y sus dependencias se instalen desde wheels precompilados. En una máquina con más de una versión de Python, py -m venv puede que selecciones una versión o arquitectura diferente a la que esperas, así que verifica después python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())" de activarla. Si pip intenta compilar cryptography desde el código fuente (un error de Rust y OpenSSL), instale primero una versión basada en wheel con pip install --only-binary=:all: cryptography, y luego instale el resto.

Instalación de dependencias

Instala los paquetes necesarios con pip:

pip install fastapi uvicorn mssql-python pydantic

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
└── 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 en esta sección abre una conexión, genera un cursor y utiliza el gestor de contexto de conexiones mssql-python para hacer commit en caso de éxito, revertir una excepción y cerrar la conexión.

Cree database.py

La get_connection_string() función construye la cadena de conexión ODBC a partir de valores de configuración. Depends() de FastAPI llama get_db_dependency() una vez por solicitud y gestiona su ciclo de vida..

# database.py
import mssql_python
from collections.abc 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"
    )

Note

ActiveDirectoryDefault utiliza DefaultAzureCredential, que prueba varios proveedores de credenciales de forma secuencial. La primera conexión puede ser lenta porque el SDK recorre la cadena hasta encontrar un proveedor que funcione. En producción, si sabes qué tipo de credencial utiliza tu entorno, especifícala directamente (por ejemplo, ActiveDirectoryMSI para identidad gestionada) para evitar el recorrido en cadena. Para más información, consulte Autenticación de Microsoft Entra.

def get_db_dependency() -> Generator:
    """FastAPI dependency for database cursor."""
    with mssql_python.connect(get_connection_string()) as conn:
        with conn.cursor() as cursor:
            yield cursor

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:
        raise HTTPException(status_code=503, detail="Database unavailable")

Ejecutar la aplicación

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

Prueba y despliega la aplicación

Utiliza el artículo complementario para completar la solicitud:

Gestión de errores

El artículo complementario trata sobre la gestión de excepciones en bases de datos.

Gestor global de excepciones

Ver Gestionar errores de base de datos.

Agrupación de conexiones

El artículo complementario trata sobre la configuración del pool de conexión.

Módulo de base de datos mejorado

Consulta Configurar la agrupación de conexiones.

Middleware de autenticación

Consulta Añadir dependencias de autenticación.

Testing

El artículo complementario trata sobre las pruebas de integración.

Configuración de pruebas

Ver Probar la aplicación.

Configuración de la implementación

El artículo complementario trata sobre la configuración y operaciones de despliegue.

Variables de entorno

Consulta Configurar configuración de despliegue y la lista de comprobación de despliegue.

Comentarios

¿Le resultó útil esta página?

Last updated on 2026-09-23