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.
  • Installieren Sie die einmaligen betriebsystem-spezifischen Voraussetzungen. Windows-Nutzer können diesen Schritt überspringen. Für vollständige Plattformdetails siehe Install mssql-python.
    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

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.

Note

Erstellen Sie unter Windows auf Arm die Umgebung mit einem Arm64-Build von Python, damit mssql-python und seine Abhängigkeiten aus vorgefertigten Wheels installiert werden. Auf einem Rechner mit mehr als einer Python-Version könnte py -m venv eine andere Version oder Architektur auswählen als erwartet. Überprüfen Sie dies daher mit python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())", nachdem Sie es aktiviert haben. Wenn pip versucht, cryptography aus dem Quellcode zu bauen (ein Fehler in der Rust- und OpenSSL-Toolchain), installieren Sie zuerst mit pip install --only-binary=:all: cryptography eine auf einem Wheel basierende Version und dann den Rest.

Abhängigkeiten installieren

Installiere die erforderlichen Pakete mit pip:

pip install fastapi uvicorn mssql-python pydantic

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

Datenbank-Verbindungsverwaltung

FastAPI verwendet Abhängigkeitsinjektion, um Ressourcen wie Datenbankverbindungen an Routenhandler bereitzustellen. Das Muster in diesem Abschnitt öffnet eine Verbindung, stellt einen Cursor bereit und verwendet den mssql-python-Verbindungskontext-Manager, um bei Erfolg die Änderungen zu übernehmen, bei einer Ausnahme ein Rollback auszuführen und die Verbindung zu schließen.

Erstellen Sie database.py

Die get_connection_string() Funktion erstellt den ODBC-Verbindungszeichenfolge aus Konfigurationswerten. FastAPIs Depends() ruft get_db_dependency() einmal pro Anfrage auf und verwaltet seinen Lebenszyklus.

# 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 verwendet DefaultAzureCredential, das mehrere Anmeldeinformationsanbieter nacheinander ausprobiert. Die erste Verbindung kann langsam sein, weil das SDK die Kette durchläuft, bis es einen funktionierenden Anbieter findet. In der Produktion gilt: Wenn du weißt, welchen Zugangsdatentyp deine Umgebung verwendet, gib ihn direkt an (zum Beispiel ActiveDirectoryMSI für eine verwaltete Identität), um den Chain Walk zu vermeiden. Weitere Informationen finden Sie unter Microsoft Entra-Authentifizierung.

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

Ausführen der Anwendung

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

Teste und deploye die Anwendung

Verwenden Sie den Begleitartikel, um die Bewerbung zu beenden:

Fehlerbehandlung

Der Begleitartikel behandelt die Behandlung von Datenbank-Ausnahmen.

Globaler Ausnahmehandler

Siehe Datenbankfehler verwalten.

Verbindungspooling

Der begleitende Artikel behandelt die Konfiguration von Verbindungspools.

Erweitertes Datenbankmodul

Siehe Connection Pooling konfigurieren.

Authentifizierungsmiddleware

Siehe Authentifizierungsabhängigkeiten hinzufügen.

Testing

Der Begleitartikel behandelt Integrationstests.

Testaufbau

Siehe Die Anwendung testen.

Bereitstellungskonfiguration

Der begleitende Artikel behandelt die Konfiguration und den Betrieb des Einsatzes.

Umgebungsvariablen

Siehe Bereitstellungseinstellungen konfigurieren und die Bereitstellungscheckliste.

Verwandte Inhalte

  • Last updated on