Použijte mssql-python s FastAPI

FastAPI je moderní Python webový framework určený pro tvorbu API. Ve spojení s mssql-python můžete vytvářet vysoce výkonná REST API podporovaná Microsoft SQL a Azure SQL Database.

Předpoklady

Python 3.10 nebo novější.

Nainstalujte požadavky specifické pro jednorázový operační systém. Uživatelé Windows mohou tento krok přeskočit. Pro úplné podrobnosti o platformě viz Instalace 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

Vytvoření databáze SQL

Vytvořte nebo se připojte k SQL databázi na jedné z následujících platforem:

Vytvoření databáze SQL v řádu minut pomocí webu Azure Portal

Příklady v tomto článku využívají databázi AdventureWorksLT , konkrétně tabulku SalesLT.Product . Pokud nemáte AdventureWorksLT nainstalovaný, podívejte se na ukázkové databáze AdventureWorks.

Nastavení projektu

Vytvoření virtuálního prostředí

Vytvořte a aktivujte virtuální prostředí, aby balíčky tohoto projektu zůstaly izolované od ostatních instalací Python. Tento krok také zabraňuje běžnému problému instalace balíčků do jednoho interpreteru při spuštění aplikace nebo testů s jiným.

Windows
Linux/macOS

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

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

Po aktivaci prostředí python, pip a pytest všechny odkazují na stejný interpret. Spusť zbývající příkazy v tomto článku z aktivovaného prostředí.

Note

Na Windows na ARM vytvořte prostředí pomocí Arm64 verze Python, takže mssql-python jeho závislosti instalujte z předpřipravených koleček. Na počítači s více než jednou verzí Pythonu může py -m venv vybrat jinou verzi nebo architekturu, než očekáváte, proto to po aktivaci ověřte pomocí python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())". Pokud se pip pokusí sestavit cryptography ze zdroje (chyba v sadě nástrojů Rust a OpenSSL), nejprve pomocí pip install --only-binary=:all: cryptography nainstalujte verzi z wheel balíčku a poté nainstalujte zbytek.

Nainstalujte závislosti

Nainstalujte požadované balíčky pomocí pip:

pip install fastapi uvicorn mssql-python pydantic

Struktura projektu

Organizujte svůj projekt do samostatných modulů pro databáze, schémata a operace CRUD:

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

Správa připojení k databázi

FastAPI využívá injekci závislostí k poskytování zdrojů, jako jsou databázová připojení, pro směrovací handlery. Vzor v této části otevře spojení, poskytne kurzor a pomocí správce kontextu připojení v mssql-python při úspěchu potvrdí transakci, při výjimce provede její vrácení zpět a nakonec spojení uzavře.

Vytvořte database.py

Funkce get_connection_string() vytváří ODBC připojovací řetězec z konfiguračních hodnot. FastAPI volá Depends()get_db_dependency() jednou na požadavek a spravuje jeho životní cyklus.

# 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 používá DefaultAzureCredential, která zkouší více poskytovatelů přihlašovacích údajů postupně. První spojení může být pomalé, protože SDK prochází řetězec, dokud nenajde funkčního poskytovatele. V produkci, pokud víte, jaký typ přihlašovacích údajů vaše prostředí používá, zadejte ho přímo (například ActiveDirectoryMSI pro spravovanou identitu), abyste se vyhnuli tzv. chain walk. Další informace naleznete v tématu ověřování Microsoft Entra.

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

Pydantické modely

Pydantické modely definují tvar a validační pravidla pro data požadavků a odpovědí. FastAPI používá tyto modely k analýze příchozího JSON, ověřování polních omezení a automatickému generování dokumentace OpenAPI.

Vytvořte schemas.py

Rozdělte schémata na Base, Create, Update a varianty odpovědí. Schéma Base obsahuje sdílená pole, Create z něj dědí pro operace vložení a Update označuje všechna pole jako volitelná pro částečné aktualizace.

# 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

operace CRUD

Zapouzdřete databázové dotazy do specializované třídy, aby obslužné funkce tras zůstaly jednoduché. Každá statická metoda vezme kurzor (injektovaný FastAPI) a provede jednu operaci pomocí parametrizovaných dotazů (%(name)s zástupců se slovníkem hodnot), aby zabránila SQL injekci. Toto oddělení usnadňuje testování a opětovné použití obchodní logiky.

Vytvořte 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()]

Aplikace FastAPI

Vytvořte main.py

Hlavní modul vše zapojuje dohromady. Každá trasa deklaruje cursor = Depends(get_db_dependency), což říká FastAPI, aby zavolal generátor, předal výsledný kurzor obslužníkovi a následně ho vyčistil. FastAPI také ověřuje těla požadavků podle vašich schémat Pydantic ještě před spuštěním obslužné funkce.

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

Spuštění aplikace

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

Otestujte a nasaděte aplikaci

Použijte doprovodný článek k dokončení žádosti:

Zpracování chyb

Doprovodný článek se zabývá zpracováním výjimek v databázi.

Globální obsluha výjimek

Viz Vyřizování chyb v databázi.

Sdílení připojení

Doprovodný článek se zabývá konfigurací spojovacího poolu.

Rozšířený databázový modul

Viz Konfigurace sdružování připojení.

middleware pro ověřování

Viz Přidat autentizační závislosti.

Testing

Doprovodný článek se zabývá integračním testováním.

Testovací nastavení

Viz Otestujte aplikaci.

Konfigurace nasazení

Doprovodný článek se zabývá konfigurací nasazení a provozem.

Proměnné prostředí

Viz Nastavit nastavení nasazení a kontrolní seznam nasazení.

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-09-23