Använd mssql-python med FastAPI

FastAPI är ett modernt Python-webbramverk för att bygga API:er. I kombination med mssql-python kan du bygga högpresterande REST-API:er som backas av Microsoft SQL och Azure SQL Database.

Förutsättningar

  • Python 3.10 eller senare.
  • paketen mssql-python, fastapi, uvicorn, pydantic och PyJWT. Installera allt med pip install fastapi uvicorn mssql-python pydantic pyjwt.
  • Installera operativsystemsspecifika engångsförutsättningar. Windows-användare kan hoppa över detta steg. För fullständiga plattformsdetaljer, se Installera mssql-python.
    apk add libtool krb5-libs krb5-dev

Skapa en SQL-databas

Skapa eller koppla till en SQL-databas på en av följande plattformar:

Exemplen i denna artikel använder AdventureWorksLT :s exempeldatabas, specifikt tabellen SalesLT.Product . Om du inte har AdventureWorksLT installerat, se AdventureWorks exempeldatabaser.

Projektinställningar

Skapa en virtuell miljö

Skapa och aktivera en virtuell miljö så att projektets paket förblir isolerade från andra Python-installationer. Detta steg förhindrar också det vanliga problemet att installera paket i en tolk medan din app körs eller testas med en annan.

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

När du har aktiverat miljön pekar python, pip och pytest alla på samma tolk. Kör de återstående kommandona i denna artikel från den aktiverade miljön.

Note

I Windows på Arm skapar du miljön med en Arm64-version av Python så att mssql-python och dess beroenden installeras från förbyggda wheel-paket. På en maskin med mer än en Python-version kan py -m venv välja en annan version eller arkitektur än väntat, så kontrollera med python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())" efter att du har aktiverat. Om du pip försöker bygga cryptography från källkoden (ett Rust- och OpenSSL-verktygskedjafel), installera först en hjulbaserad version med pip install --only-binary=:all: cryptography, och installera sedan resten.

Installera beroenden

Installera de nödvändiga paketen med pip:

pip install fastapi uvicorn mssql-python pydantic pyjwt

Projekt-struktur

Organisera ditt projekt med separata moduler för databas, scheman och CRUD-operationer:

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

Databasanslutningshantering

FastAPI använder beroendeinjektion för att tillhandahålla resurser som databasanslutningar till rutthanterare. Mönstret i detta avsnitt skapar en kontexthanterare som öppnar en anslutning, ger en markör och hanterar commit/rollback/stängning automatiskt.

Skapa database.py

Funktionen get_connection_string() bygger ODBC-reťazec pripojenia från konfigurationsvärden. Kontexthanteraren get_db() och generatorn get_db_dependency() följer båda samma mönster: de öppnar en anslutning, returnerar en markör, verkställer transaktionen vid framgång, rullar tillbaka den vid fel och stänger alltid anslutningen när de är klara. FastAPI anropar Depends()get_db_dependency() en gång per förfrågan och hanterar dess livscykel.

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

Note

ActiveDirectoryDefault använder DefaultAzureCredential, som provar flera leverantörer av autentiseringsuppgifter i följd. Den första anslutningen kan vara långsam eftersom SDK:n går igenom kedjan tills den hittar en fungerande leverantör. I produktion, om du vet vilken typ av behörighet din miljö använder, ange det direkt (till exempel ActiveDirectoryMSI för managed identity) för att undvika kedjevandring. Mer information finns i Microsoft Entra-autentisering.

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

Pydantiska modeller

Pydantiska modeller definierar form- och valideringsregler för begäran- och svarsdata. FastAPI använder dessa modeller för att analysera inkommande JSON, validera fältbegränsningar och automatiskt generera OpenAPI-dokumentation.

Skapa schemas.py

Dela upp scheman i Base, Create, Update, och responsvarianter. Schemat Base innehåller delade fält, Create ärver från dem för insättningsoperationer och Update gör alla fält valfria för partiella uppdateringar.

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

Kapsla in databasfrågor i en dedikerad klass för att hålla rutthanterare tunna. Varje statisk metod tar en markör (injicerad av FastAPI) och hanterar en operation med hjälp av parameteriserade frågor (%(name)s platshållare med en ordbok över värden) för att förhindra SQL-injektion. Denna separation gör affärslogiken lättare att testa och återanvända.

Skapa 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-applikation

Skapa main.py

Huvudmodulen kopplar ihop allt. Varje rutt deklarerar cursor = Depends(get_db_dependency), vilket säger åt FastAPI att anropa generatorn, skicka den yieldade markören till hanteraren och sedan rensa upp. FastAPI validerar också förfrågningskroppar mot dina Pydantiska scheman innan hanteraren körs.

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

Starta programmet

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

Felhantering

FastAPI låter dig registrera globala undantaghanterare för specifika undantagstyper. När du fångar mssql_python.DatabaseError och mssql_python.IntegrityError, returnerar FastAPI strukturerade JSON-fel med lämpliga HTTP-statuskoder istället för generiska 500-svar.

Global undantagshantering

Lägg till dessa hanterare till main.py, precis efter app = FastAPI(...) linjen. FastAPI kör matchningshanteraren varje gång en rutt tar upp den undantagstypen, så du behöver inte ett try/except block i varje rutt.

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

Note

Att ta bort en produkt som andra rader fortfarande refererar till höjer mssql_python.IntegrityError från främmande nyckelbegränsningen, och hanteraren returnerar 400 istället för att ta bort raden. I exempeldatabasen AdventureWorksLT refereras de flesta produkter i SalesLT.Product av SalesLT.SalesOrderDetail, så DELETE misslyckas avsiktligt för dem. För att testa en lyckad borttagning, skapa en produkt med POST /products och ta bort den, eller ta bort referensraderna först.

Anslutningspoolning

Utan anslutningspooling öppnar och stänger varje förfrågan en TCP-anslutning till Microsoft SQL, vilket ökar latensen. Anslutningspoolning innebär att en uppsättning lediga anslutningar hålls redo för återanvändning. Anropa mssql_python.pooling() en gång vid start. När poolning är aktiverat returneras anslutningen till poolen av conn.close() i get_db_dependency() i stället för att faktiskt stängas.

Förbättrad databasmodul

Aktivera pooling genom att anropa mssql_python.pooling() vid uppstart och konfigurera med lämpliga maxstorleks- och timeout-inställningar:

# 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

Autentiseringsmellanvara

Du kan kombinera databasåtkomst med autentisering genom att kedja FastAPI-beroenden. Följande exempel validerar en JWT-bärartoken, slår upp matchande personpost i AdventureWorksLT:s exempeldatabas och gör resultatet tillgängligt för skyddade rutter.

# 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 tillhandahåller en TestClient som bygger på httpx och skickar begäranden till din applikation utan att starta en riktig HTTP-server. Skriv tester med pytest för att verifiera rutter, statuskoder och svarsformer.

Innan du kör testerna i detta avsnitt, installera testberoenden:

pip install pytest httpx

Note

Om du använder den senaste versionen av Starlette eller sätter upp en ny miljö, använd httpx2 i stället för httpx. Nyare versioner av Starlette använder httpx2 för TestClient och visar en utfasningsvarning när endast httpx är installerad. Installera den med pip install pytest httpx2.

Testkonfiguration

Skapa en testfil som används TestClient för att verifiera ruttbeteende och svarsscheman:

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

Kör testerna med pytest från projektets rot, samma katalog som main.py:

pytest

Dessa tester körs mot din aktiva databas i stället för mockobjekt, så test_create_product lägger in en verklig rad i SalesLT.Product. I AdventureWorksLT har både Name och ProductNumber unika begränsningar, så testet genererar ett unikt värde för varje körning. Om du hårdkodar dessa värden istället misslyckas testet med en konflikt vid andra körningen om du inte tar bort raden först.

Distributionskonfiguration

Använd Pydantics BaseSettings för att ladda konfiguration från miljövariabler och .env filer. Detta tillvägagångssätt håller hemligheter borta från källkoden och gör det enkelt att byta mellan miljöer. Installera inställningspaketet med pip install pydantic-settings.

Miljövariabler

Skapa en inställningsmodul som laddar konfiguration från miljövariabler, vilket låter dig hantera hemligheter och distributionsspecifika värden utanför din kod:

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

Uppdatera database.py sedan för att importera get_connection_string från config istället för att definiera en egen kopia. Genom att ta bort den duplicerade funktionen säkerställer du att appen läser anslutningsinställningar från en enda källa.

# database.py
from config import get_connection_string

Relaterat innehåll

  • Last updated on