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.

Installeer eenmalige vereisten voor het besturingssysteem. Windows-gebruikers kunnen deze stap overslaan. Voor volledige platformdetails, zie Install mssql-python.

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

Een SQL-database maken

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

Binnen enkele minuten een SQL-database maken met behulp van Azure Portal

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.

Windows
Linux/macOS

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.

Note

Op Windows on Arm maak je de omgeving met een Arm64-build van Python so mssql-python en de bijbehorende afhankelijkheden worden geïnstalleerd vanaf vooraf gebouwde wielen. Op een machine met meer dan één Python-versie kan py -m venv een andere versie of architectuur selecteren dan je verwacht, dus controleer dit met python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())" nadat je hebt geactiveerd. Als pip probeert cryptography vanuit broncode te bouwen (een fout met de Rust- en OpenSSL-toolchain), installeer dan eerst een versie op basis van een wheel met pip install --only-binary=:all: cryptography en installeer daarna de rest.

Afhankelijkheden installeren

Installeer de benodigde pakketten met pip:

pip install fastapi uvicorn mssql-python pydantic

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
└── routers/
    └── products.py

Databaseverbindingsbeheer

FastAPI gebruikt afhankelijkheidsinjectie om middelen zoals databaseverbindingen aan routehandlers te leveren. Het patroon in deze sectie opent een verbinding, levert een cursor op en gebruikt de mssql-python verbindingscontextmanager om te committen bij succes, een uitzondering terug te rollen en de verbinding te sluiten.

Maak het bestand database.py

De get_connection_string() functie bouwt de ODBC-verbindingsreeks op basis van configuratiewaarden. FastAPI roept Depends()get_db_dependency() één keer per verzoek aan en beheert zijn levenscyclus.

# 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 gebruikt DefaultAzureCredential, dat meerdere credentialproviders achter elkaar probeert. De eerste verbinding kan traag zijn omdat de SDK de keten doorloopt totdat hij een werkende provider vindt. In productie, als je weet welk type inloggegevens je omgeving gebruikt, specificeer het dan direct (bijvoorbeeld ActiveDirectoryMSI voor managed identity) om de chain walk te voorkomen. Zie Microsoft Entra-verificatie voor meer informatie.

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

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

De toepassing uitvoeren

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

Test en implementeer de applicatie

Gebruik het bijbehorende artikel om de aanvraag af te ronden:

Foutafhandeling

Het bijbehorende artikel behandelt de afhandeling van database-uitzonderingen.

Globale uitzonderingshandler

Zie Databasefouten behandelen.

Groepsgewijze verbindingen

Het bijbehorende artikel behandelt de configuratie van de verbindingspool.

Verbeterde databasemodule

Zie Verbindingspooling configureren.

middleware voor authenticatie

Zie Toevoegen van authenticatieafhankelijkheden.

Testing

Het bijbehorende artikel behandelt integratietesten.

Testopstelling

Zie Test de applicatie.

Implementatieconfiguratie

Het bijbehorende artikel behandelt de configuratie en operaties van de inzet.

Omgevingsvariabelen

Zie Configuratie-implementatieinstellingen en de implementatiechecklist.

Feedback

Is deze pagina nuttig?

Last updated on 2026-09-23