Verwenden Sie mssql-python mit FastAPI

FastAPI ist ein modernes Python-Webframework zum Erstellen von APIs. In Kombination mit mssql-python können Sie leistungsstarke REST-APIs bauen, die von Microsoft SQL und Azure SQL-Datenbank unterstützt werden.

Voraussetzungen

• Python 3.10 oder höher.
• Die mssql-python, fastapi, uvicorn, , pydantic, und PyJWT Pakete. Installieren Sie alle mit pip install fastapi uvicorn mssql-python pydantic pyjwt.
• Installieren Sie die einmaligen betriebsystem-spezifischen Voraussetzungen. Windows-Nutzer können diesen Schritt überspringen. Für vollständige Plattformdetails siehe Install mssql-python.
  • Alpin
  • Debian/Ubuntu
  • RHEL
  • SUSE
  • openSUSE
  • macOS

  apk add libtool krb5-libs krb5-dev
  

Erstellen einer SQL-Datenbank

Erstellen oder verbinden Sie sich mit einer SQL-Datenbank auf einer der folgenden Plattformen:

Die Beispiele in diesem Artikel verwenden die AdventureWorksLT-Beispieldatenbank , speziell die Tabelle SalesLT.Product . Wenn Sie AdventureWorksLT nicht installiert haben, sehen Sie sich die AdventureWorks-Beispieldatenbanken an.

Projektkonfiguration

Erstellen einer virtuellen Umgebung

Erstellen und aktivieren Sie eine virtuelle Umgebung, damit die Pakete dieses Projekts von anderen Python-Installationen isoliert bleiben. Dieser Schritt verhindert auch das häufige Problem, Pakete in einen Interpreter zu installieren, während die eigene App läuft oder mit einem anderen getestet wird.

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

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

Nachdem du die Umgebung aktiviert hast, verweisen python, pip und pytest alle auf denselben Interpreter. Führe die übrigen Befehle in diesem Artikel aus der aktivierten Umgebung aus.

Abhängigkeiten installieren

Installiere die erforderlichen Pakete mit pip:

pip install fastapi uvicorn mssql-python pydantic pyjwt

Projektstruktur

Organisieren Sie Ihr Projekt mit separaten Modulen für Datenbank, Schemata und CRUD-Operationen:

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

Datenbank-Verbindungsverwaltung

FastAPI verwendet Abhängigkeitsinjektion, um Ressourcen wie Datenbankverbindungen an Routenhandler bereitzustellen. Das Muster in diesem Abschnitt erstellt einen Kontextmanager, der eine Verbindung öffnet, einen Cursor liefert und automatisch Commit/Rollback/Closing verwaltet.

Erstellen Sie database.py

Die get_connection_string() Funktion erstellt den ODBC-Verbindungszeichenfolge aus Konfigurationswerten. Der get_db() Kontextmanager und der get_db_dependency() Generator folgen beide demselben Muster: eine Verbindung öffnen, einen Cursor zurückgeben, bei Erfolg committen, bei Fehlern ein Rollback durchführen und am Ende immer schließen. FastAPIs Depends() ruft get_db_dependency() einmal pro Anfrage auf und verwaltet seinen Lebenszyklus.

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

Pydantische Modelle definieren die Form- und Validierungsregeln für Anfrage- und Antwortdaten. FastAPI verwendet diese Modelle, um eingehendes JSON zu analysieren, Feldbeschränkungen zu validieren und OpenAPI-Dokumentation automatisch zu generieren.

Erstellen Sie schemas.py

Trenne Schemata in Base, Create, Update, und Antwortvarianten. Das Schema Base enthält gemeinsame Felder, Create erbt davon für Einfügeoperationen, und Update macht alle Felder für partielle Aktualisierungen optional.

# 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-Vorgänge

Lagern Sie Datenbankabfragen in eine eigene Klasse aus, um Route-Handler schlank zu halten. Jede statische Methode nimmt einen Cursor (von FastAPI eingeschleust) und verarbeitet eine Operation mittels parametrisierter Abfragen (%(name)s Platzhalter mit einem Wörterbuch von Werten), um SQL-Injektion zu verhindern. Diese Trennung macht die Geschäftslogik leichter zu testen und wiederzuverwenden.

Erstellen Sie 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()]

FastAPI-Anwendung

Erstellen Sie main.py

Das Hauptmodul verbindet alles miteinander. Jede Route deklariert cursor = Depends(get_db_dependency), was FastAPI anweist, den Generator aufzurufen, den ergebenen Cursor an den Handler weiterzugeben und anschließend aufzuräumen. FastAPI validiert außerdem Anfrageinhalte anhand deiner Pydantic-Schemata, bevor der Handler ausgeführt wird.

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

Ausführen der Anwendung

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

Fehlerbehandlung

FastAPI ermöglicht es, globale Ausnahmehandler für bestimmte Ausnahmetypen zu registrieren. Wenn Sie mssql_python.DatabaseError und mssql_python.IntegrityError abfangen, gibt FastAPI strukturierte JSON-Fehler mit entsprechenden HTTP-Statuscodes statt generischer 500-Antworten zurück.

Globaler Ausnahmehandler

Füge diese Handler zu main.py hinzu, direkt nach der Zeile app = FastAPI(...). FastAPI führt den Matching-Handler aus, wann immer eine Route diesen Ausnahmetyp aufruft, sodass du nicht in jeder Route einen Block try/except brauchst.

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

Verbindungspooling

Ohne Connection Pooling öffnet und schließt jede Anfrage eine TCP-Verbindung zu Microsoft SQL, was die Latenz erhöht. Connection Pooling hält eine Reihe von Leerlaufverbindungen bereit zur Wiederverwendung. Rufen Sie mssql_python.pooling() beim Start einmal auf. Bei aktiviertem Pooling gibt conn.close() in get_db_dependency() die Verbindung an den Pool zurück, anstatt sie tatsächlich zu schließen.

Erweitertes Datenbankmodul

Aktivieren Sie das Pooling, indem Sie beim Start aufrufen mssql_python.pooling() und konfigurieren Sie es mit den entsprechenden maximalen Größen- und Timeout-Einstellungen:

# 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

Authentifizierungsmiddleware

Man kann Datenbankzugriff mit Authentifizierung kombinieren, indem man FastAPI-Abhängigkeiten aneinanderreiht. Das folgende Beispiel validiert ein JWT-Trägertoken, sucht den passenden Personeneintrag in der AdventureWorksLT-Beispieldatenbank nach und stellt das Ergebnis geschützten Routen zur Verfügung.

# 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

Testen

FastAPI bietet ein auf httpx basierendes TestClient, das Anfragen an Ihre Anwendung sendet, ohne einen echten HTTP-Server zu starten. Schreibe Tests mit pytest, um Routen, Statuscodes und Antwortstrukturen zu überprüfen.

Bevor Sie die Tests in diesem Abschnitt ausführen, installieren Sie die Testabhängigkeiten:

pip install pytest httpx

Testaufbau

Erstellen Sie eine Testdatei, die zur Überprüfung von Routenverhalten und Antwortschemata verwendet wird TestClient :

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

Führen Sie die Tests im Stammverzeichnis des Projekts mit pytest aus, also in demselben Verzeichnis wie main.py:

pytest

Diese Tests werden gegen deine Live-Datenbank und nicht gegen Mocks ausgeführt, daher fügt test_create_product eine echte Zeile in SalesLT.Product ein. In AdventureWorksLT haben beide Name und ProductNumber einzigartige Einschränkungen, sodass der Test für jeden bei jedem Durchlauf einen einzigartigen Wert erzeugt. Wenn du diese Werte stattdessen fest kodierst, scheitert der Test mit einem Konflikt beim zweiten Durchlauf, es sei denn, du löschst zuerst die Zeile.

Bereitstellungskonfiguration

Verwenden Sie Pydantics BaseSettings, um Konfiguration aus Umgebungsvariablen und .env-Dateien zu laden. Dieser Ansatz hält Geheimnisse aus dem Quellcode heraus und erleichtert den Wechsel zwischen den Umgebungen. Installiere das Einstellungspaket mit pip install pydantic-settings.

Umgebungsvariablen

Erstellen Sie ein Einstellungsmodul, das Konfigurationen aus Umgebungsvariablen lädt und es Ihnen ermöglicht, Geheimnisse und deployment-spezifische Werte außerhalb Ihres Codes zu verwalten:

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

Aktualisieren Sie dann database.py, sodass get_connection_string aus config importiert wird, anstatt eine eigene Kopie davon zu definieren. Indem Sie die duplizierte Funktion entfernen, stellen Sie sicher, dass die App die Verbindungseinstellungen von einer einzigen Quelle liest.

# database.py
from config import get_connection_string

Verwandte Inhalte

• Verbindungsverwaltung
• Verbindungspooling
• Fehlerbehandlung
• Microsoft Entra-Authentifizierung