Gebruik mssql-python met FastAPI

FastAPI is een modern Python-webframework voor het bouwen van API's. In combinatie met mssql-python kun je high-performance REST API's bouwen die worden ondersteund door Microsoft SQL en Azure SQL Database.

Prerequisites

• Python 3.10 of hoger.
• De mssql-python, fastapi, uvicorn, , pydantic, en PyJWT pakketten. Installeer alles met pip install fastapi uvicorn mssql-python pydantic pyjwt.
• Installeer eenmalige vereisten voor het besturingssysteem. Windows-gebruikers kunnen deze stap overslaan. Voor volledige platformdetails, zie Install mssql-python.
  • Alpen
  • Debian/Ubuntu
  • RHEL
  • SUSE
  • openSUSE
  • macOS

  apk add libtool krb5-libs krb5-dev
  

Een SQL-database maken

Maak een SQL-database aan of maak verbinding met een van de volgende platforms:

De voorbeelden in dit artikel gebruiken de voorbeelddatabase van AdventureWorksLT , specifiek de SalesLT.Product tabel. Als je AdventureWorksLT niet hebt geïnstalleerd, zie dan de voorbeelddatabases van AdventureWorks.

Projectopstelling

Een virtuele omgeving maken

Maak een virtuele omgeving aan en activeer deze zodat de pakketten van dit project geïsoleerd blijven van andere Python-installaties. Deze stap voorkomt ook het veelvoorkomende probleem van het installeren van pakketten in één interpreter tijdens het uitvoeren van je app of tests met een andere.

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

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

Nadat je de omgeving hebt geactiveerd, verwijzen python, pip en pytest allemaal naar dezelfde interpreter. Voer de resterende commando's uit dit artikel vanuit de geactiveerde omgeving.

Afhankelijkheden installeren

Installeer de benodigde pakketten met pip:

pip install fastapi uvicorn mssql-python pydantic pyjwt

Projectstructuur

Organiseer je project met aparte modules voor database, schema's en CRUD-operaties:

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

Databaseverbindingsbeheer

FastAPI gebruikt afhankelijkheidsinjectie om middelen zoals databaseverbindingen aan routehandlers te leveren. Het patroon in deze sectie maakt een contextmanager aan die een verbinding opent, een cursor oplevert en automatisch commit/rollback/close afhandelt.

Maak het bestand database.py

De get_connection_string() functie bouwt de ODBC-verbindingsreeks op basis van configuratiewaarden. De get_db()-contextmanager en de get_db_dependency()-generator volgen beide hetzelfde patroon: een verbinding openen, een cursor teruggeven, de transactie bij succes vastleggen, bij een fout de transactie terugdraaien, en na afloop altijd sluiten. FastAPI roept Depends()get_db_dependency() één keer per verzoek aan en beheert zijn levenscyclus.

# 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()

Pydantische modellen

Pydantische modellen definiëren de vorm- en validatieregels voor verzoek- en responsgegevens. FastAPI gebruikt deze modellen om binnenkomende JSON te analyseren, veldbeperkingen te valideren en OpenAPI-documentatie automatisch te genereren.

Maak het bestand schemas.py

Splits schema’s op in Base, Create, Update en antwoordvarianten. Het Base schema bevat gedeelde velden, Create erft ervan voor invoegoperaties en Update maakt alle velden optioneel voor gedeeltelijke updates.

# 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

CRUD-bewerkingen

Breng databasequery's onder in een aparte klasse om route-handlers compact te houden. Elke statische methode neemt een cursor (geïnjecteerd door FastAPI) en verwerkt één bewerking met geparametriseerde queries (%(name)s placeholders met een woordenboek van waarden) om SQL-injectie te voorkomen. Deze scheiding maakt de bedrijfslogica gemakkelijker te testen en hergebruiken.

Maak crud.py aan

# 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()]

FastAPI-toepassing

Maak het bestand main.py aan

De hoofdmodule verbindt alles met elkaar. Elke route definieert cursor = Depends(get_db_dependency), waarmee FastAPI wordt verteld de generator aan te roepen, de opgeleverde cursor aan de handler door te geven en deze daarna op te ruimen. FastAPI valideert ook verzoeklichamen tegen je Pydantic-schema's voordat de handler draait.

# 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)}")

De toepassing uitvoeren

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

Foutafhandeling

FastAPI laat je globale uitzonderingshandlers registreren voor specifieke uitzonderingtypes. Wanneer je mssql_python.DatabaseError en mssql_python.IntegrityError opvangt, retourneert FastAPI gestructureerde JSON-fouten met de juiste HTTP-statuscodes in plaats van algemene 500-antwoorden.

Globale uitzonderingshandler

Voeg deze handlers toe aan main.py, direct na de app = FastAPI(...) lijn. FastAPI draait de matching handler telkens wanneer een route dat exceptiontype opneemt, dus je hebt niet in elke route een try/except blok nodig.

# 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"}
    )

Groepsgewijze verbindingen

Zonder connection pooling opent en sluit elk verzoek een TCP-verbinding met Microsoft SQL, wat de latentie toevoegt. Verbindingspooling houdt een verzameling inactieve verbindingen gereed voor hergebruik. Roep mssql_python.pooling() één keer aan bij het opstarten. Met pooling ingeschakeld conn.close() geeft in get_db_dependency() de verbinding terug naar de pool in plaats van deze daadwerkelijk te sluiten.

Verbeterde databasemodule

Schakel pooling in door bij het opstarten aan te roepen mssql_python.pooling() en configureer deze met de juiste maximale grootte en timeout-instellingen:

# 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 voor authenticatie

Je kunt database-toegang combineren met authenticatie door FastAPI-afhankelijkheden aan elkaar te koppelen. Het volgende voorbeeld valideert een JWT-dragertoken, zoekt het overeenkomstige persoonrecord op in de AdventureWorksLT-voorbeelddatabase en maakt het resultaat beschikbaar voor beschermde routes.

# 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 biedt een TestClient, gebouwd op httpx, die verzoeken naar je applicatie stuurt zonder een echte HTTP-server te starten. Schrijf tests met pytest om routes, statuscodes en de structuur van responses te verifiëren.

Installeer voordat je de tests in deze sectie uitvoert, de testafhankelijkheden:

pip install pytest httpx

Testopstelling

Maak een testbestand aan dat wordt gebruikt TestClient om routegedrag en responsschema's te verifiëren:

# 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"

Voer de tests uit met pytest vanaf de projectroot, dezelfde map als main.py:

pytest

Deze tests worden uitgevoerd op je live-database in plaats van op mockobjecten, dus voegt test_create_product een echte rij toe aan SalesLT.Product. In AdventureWorksLT hebben beide Name en ProductNumber unieke beperkingen, dus de test genereert bij elke run een unieke waarde voor elk. Als je die waarden hardcodeert, faalt de test met een conflict bij de tweede run, tenzij je eerst de rij verwijdert.

Implementatieconfiguratie

Gebruik Pydantic's BaseSettings om configuraties te laden vanuit omgevingsvariabelen en .env bestanden. Deze aanpak houdt geheimen buiten de broncode en maakt het gemakkelijk om tussen omgevingen te wisselen. Installeer het instellingenpakket met pip install pydantic-settings.

Omgevingsvariabelen

Maak een instellingenmodule die configuratie laadt vanuit omgevingsvariabelen, zodat je secrets en implementatie-specifieke waarden buiten je code kunt beheren:

# 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"
    )

Werk database.py daarna bij om get_connection_string uit config te importeren in plaats van zijn eigen kopie te definiëren. Door de gedupliceerde functie te verwijderen, zorg je ervoor dat de app de verbindingsinstellingen van één bron leest.

# database.py
from config import get_connection_string

Verwante onderwerpen

• Verbindingsbeheer
• Groepsgewijze verbindingen
• Foutafhandeling
• Microsoft Entra-authenticatie