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.

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

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

Skapa en SQL-databas

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

Skapa en SQL-databas på några minuter med hjälp av Azure-portalen

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.

Windows
Linux/macOS

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

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

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

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

Databasanslutningshantering

FastAPI använder beroendeinjektion för att tillhandahålla resurser som databasanslutningar till rutthanterare. Mönstret i det här avsnittet öppnar en anslutning, returnerar en markör och använder mssql-python:s kontexthanterare för anslutningar för att genomföra transaktionen vid lyckat resultat, återställa den vid ett undantag och stänga anslutningen.

Skapa database.py

Funktionen get_connection_string() bygger ODBC-reťazec pripojenia från konfigurationsvärden. FastAPI anropar Depends()get_db_dependency() en gång per förfrågan och hanterar dess livscykel.

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

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

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.

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

Starta programmet

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

Servern lyssnar på http://localhost:8000. Låt denna terminal vara igång medan du använder API:et.

Använd API:et

Öppna http://localhost:8000/docs i en webbläsare. FastAPI visar interaktiv dokumentation för varje rutt.

Expandera GET /health, välj Try it out och välj sedan Execute. Verifiera att svaret har statuskod 200 och rapporterar en frisk databasanslutning.
Expandera GET /products, välj Try it out, ställ page_size in på 5, och välj sedan Kör. Svaret innehåller fem produkter och paginationsinformation.
Kopiera ett id värde från svaret. Expandera GET /products/{product_id}, välj Try it out, ange det kopierade värdet för product_id, och välj sedan Kör.
Expandera GET /products/search/, välj Try it out, skriv in en sökterm som bike för q, och välj sedan Execute.

Testa och distribuera applikationen

För vägledning om felhantering, anslutningspooling, autentisering, testning och distribution, se Test och distribuera FastAPI-applikationer med mssql-python.

Feedback

Var den här sidan till hjälp?

Last updated on 2026-09-30