Użyj mssql-python z FastAPI

FastAPI to nowoczesny framework webowy Python do tworzenia API. W połączeniu z mssql-python możesz budować wysokowydajne REST API wspierane przez Microsoft SQL i Azure SQL Database.

Wymagania wstępne

• Python 3.10 lub nowszy.
• Zainstaluj jednorazowe wymagania wstępne dotyczące systemu operacyjnego. Użytkownicy Windows mogą pominąć ten krok. Pełne szczegóły dotyczące platformy można znaleźć w artykule Install mssql-python.
  • Alpejski
  • Debian/Ubuntu
  • RHEL
  • SUSE
  • openSUSE
  • macOS

  apk add libtool krb5-libs krb5-dev
  

Tworzenie bazy danych SQL

Stwórz lub połącz się z bazą danych SQL na jednej z następujących platform:

Przykłady w tym artykule korzystają z przykładowej bazy danych AdventureWorksLT , a konkretnie z tabeli SalesLT.Product . Jeśli nie masz zainstalowanego AdventureWorksLT, zobacz przykładowe bazy danych AdventureWorks.

Konfiguracja projektu

Tworzenie środowiska wirtualnego

Stwórz i aktywuj środowisko wirtualne, aby pakiety tego projektu pozostały odizolowane od innych instalacji Python. Ten krok zapobiega również częstemu problemowi instalowania pakietów w jednym interpreterze podczas uruchamiania aplikacji lub testów na innym.

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

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

Po aktywowaniu środowiska python, pip i pytest wszystkie wskazują na ten sam interpreter. Wykonaj pozostałe polecenia z tego artykułu z aktywowanego środowiska.

Instalowanie zależności

Zainstaluj wymagane pakiety za pomocą pip:

pip install fastapi uvicorn mssql-python pydantic

struktura projektu

Zorganizuj swój projekt z osobnymi modułami dotyczącymi baz danych, schematów i operacji CRUD:

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

Zarządzanie połączeniami bazy danych

FastAPI wykorzystuje wstrzykiwanie zależności do udostępniania zasobów, takich jak połączenia z bazą danych, programom obsługi tras. Wzór w tej sekcji otwiera połączenie, wyświetla kursor i używa menedżera kontekstu połączenia mssql-python, aby zatwierdzić sukces, cofnąć wyjątek i zamknąć połączenie.

Stwórz database.py

Funkcja get_connection_string() buduje parametry połączenia ODBC na podstawie wartości konfiguracyjnych. FastAPI Depends() wywołuje get_db_dependency() jeden raz dla każdego żądania i zarządza jego cyklem życia.

# 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

Modele pydantyczne

Modele pydantyczne definiują kształt i reguły walidacji danych żądań i odpowiedzi. FastAPI wykorzystuje te modele do analizy przychodzącego JSON, walidacji ograniczeń polowych oraz automatycznego generowania dokumentacji OpenAPI.

Stwórz schemas.py

Podziel schematy na Base, Create, Update, oraz warianty odpowiedzi. Schemat Base przechowuje pola współdzielone, Create dziedziczy z nich do operacji wstawiania i Update czyni wszystkie pola opcjonalnymi dla częściowych aktualizacji.

# 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

Operacje CRUD

Umieść zapytania do bazy danych w osobnej klasie, aby obsługa tras pozostała prosta i lekka. Każda statyczna metoda przyjmuje kursor (wstrzyknięty przez FastAPI) i wykonuje jedną operację za pomocą parametryzowanych zapytań (%(name)s zastępczych z słownikiem wartości), aby zapobiec wstrzykiwaniu SQL. To rozdzielenie ułatwia testowanie i ponowne wykorzystanie logiki biznesowej.

Stwórz 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()]

Aplikacja FastAPI

Stwórz main.py

Główny moduł łączy wszystko razem. Każda trasa deklaruje cursor = Depends(get_db_dependency), co nakazuje FastAPI wywołać generator, przekazać wybrany kursor do handlera i następnie wyczyścić. FastAPI również sprawdza poprawność treści żądania na podstawie schematów Pydantic, zanim zostanie uruchomiona funkcja obsługująca.

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

Uruchamianie aplikacji

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

Serwer nasłuchuje na http://localhost:8000. Utrzymuj ten terminal włączony, podczas gdy ćwiczysz API.

Ćwicz API

Otwórz http://localhost:8000/docs w przeglądarce. FastAPI wyświetla interaktywną dokumentację dla każdej trasy.

1. Rozwiń GET /health, wybierz Wypróbuj, a następnie wybierz Wykonaj. Sprawdź, czy odpowiedź ma kod 200 statusu i raportuje prawidłowe połączenie z bazą danych.
2. Rozwiń GET /products, wybierz Spróbuj, ustaw page_size na 5, a następnie wybierz Wykonaj. Odpowiedź zawiera pięć produktów oraz szczegóły dotyczące paginacji.
3. Skopiuj id wartość z odpowiedzi. Rozwiń GET /products/{product_id}, wybierz Spróbuj, wpisz skopiowaną wartość dla product_id, a następnie wybierz Wykonaj.
4. Rozwiń GET /products/search/, wybierz Wypróbuj, wpisz hasło wyszukiwania, np. bike for q, a następnie wybierz Wykonaj.

Przetestuj i wdrażaj aplikację

Aby uzyskać wskazówki dotyczące obsługi błędów, pulowania połączeń, uwierzytelniania, testowania i wdrażania, zobacz: Testuj i wdrażaj aplikacje FastAPI z mssql-python.

Treści powiązane

• Testuj i wdrażaj aplikacje FastAPI za pomocą mssql-python
• Zarządzanie połączeniami za pomocą mssql-python
• Pulowanie połączeń z mssql-python
• Obsługa błędów i kody SQLSTATE dla mssql-python
• Microsoft Entra authentication with mssql-python