Használj mssql-pythont FastAPI-val

A FastAPI egy modern Python webes keretrendszer API-k építéséhez. Az mssql-pythonnal kombinálva nagy teljesítményű REST API-kat építhetsz Microsoft SQL és Azure SQL Database támogatásával.

Prerequisites

Python 3.10 vagy újabb verzió.

Egyszeri operációs rendszerspecifikus előfeltételek telepítése. A Windows felhasználók ezt a lépést kihagyhatják. A platform teljes részleteiért lásd: Install mssql-python.

apk add libtool krb5-libs krb5-dev libstdc++

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

SQL-adatbázis létrehozása

Létrehozni vagy csatlakozni SQL adatbázishoz az alábbi platformok egyikén:

SQL-adatbázis létrehozása percek alatt az Azure Portal használatával

A cikkben szereplő példák az AdventureWorksLT mintaadatbázist, konkrétan a SalesLT.Product táblát használják. Ha nincs telepítve az AdventureWorksLT, nézd meg az AdventureWorks mintaadatbázisokat.

Projekt beállítás

Virtuális környezet létrehozása

Hozz létre és aktiválj egy virtuális környezetet, hogy a projekt csomagjai elszigeteltek maradjanak a többi Python telepítéstől. Ez a lépés elkerüli azt a gyakori problémát is, hogy csomagokat telepítsenek egy értelmezőbe, miközben az alkalmazásodat vagy egy másiknál teszteket futtatnak.

Windows
Linux/macOS

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

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

A környezet aktiválása után a python, a pip és a pytest mind ugyanarra az értelmezőre mutat. A cikkben a többi parancsot az aktivált környezetből futtasd le.

Note

Windows on Arm rendszeren a környezetet a Python Arm64-es verziójával hozd létre, hogy a(z) mssql-python és annak függőségei előre elkészített wheel csomagokból települhessenek. Egynél több Python-verzióval rendelkező gépen a py -m venv a várttól eltérő verziót vagy architektúrát is kiválaszthat, ezért aktiválás után a python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())" paranccsal ellenőrizd. Ha a(z) pip megpróbálja forrásból felépíteni a(z) cryptography csomagot (Rust- és OpenSSL-eszközlánccal kapcsolatos hiba miatt), először a(z) pip install --only-binary=:all: cryptography paranccsal telepíts egy wheel-alapú verziót, majd a többit.

Függőségek telepítése

Telepítsd a szükséges csomagokat pip-jel:

pip install fastapi uvicorn mssql-python pydantic

A projekt szerkezete

Szervezze meg a projektjét külön modulokkal adatbázisokhoz, sémákhoz és CRUD műveletekhez:

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

Adatbázis-kapcsolatkezelés

A FastAPI függőségi injekciót használ, hogy olyan erőforrásokat, mint például adatbázis-kapcsolatok biztosítson az útvonalkezelők számára. Az ebben a szakaszban bemutatott minta megnyit egy kapcsolatot, visszaad egy kurzort, és az mssql-python kapcsolati kontextuskezelőjét használja arra, hogy siker esetén véglegesítse a tranzakciót, kivétel esetén visszagörgesse azt, majd lezárja a kapcsolatot.

Készíts database.py

A get_connection_string() függvény konfigurációs értékekből építi az ODBC kapcsolati karakterlánc-et. A FastAPI Depends() kérésenként egyszer hívja meg a(z) get_db_dependency() elemet, és kezeli annak életciklusát.

# 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 használ DefaultAzureCredential, amely több hitelesítésszolgáltatót próbál egymás után. Az első kapcsolat lassú lehet, mert az SDK végigjárja a láncot, amíg meg nem talál egy működő szolgáltatót. A termelésben, ha tudod, melyik hitelesítéstípust használja a környezeted, közvetlenül megadd (például ActiveDirectoryMSI menedzselt identitásnál), hogy elkerüld a láncos sétát. További információ: Microsoft Entra-hitelesítés.

Pydantikus modellek

A pydantikus modellek határozzák meg a kérelmek és válaszadatok alakját és validációs szabályait. A FastAPI ezeket a modelleket használja a bejövő JSON elemzésére, a mezőkorlátok validálására és az OpenAPI dokumentáció automatikus generálására.

Készíts schemas.py

Szegezzük a sémákat Base, Create, Update, és válaszvariánsokra. A Base séma megosztott mezőket tart, Create örököl belőle beszúrási műveletekhez, és Update minden mezőt opcionálissá tesz részleges frissítésekhez.

# 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 műveletek

Incapsuláld az adatbázis-lekérdezéseket egy dedikált osztályba, hogy az útvonalkezelők vékonyak legyenek. Minden statikus metódus egy kurzort fogad (amelyet a FastAPI ad át), és paraméterezett lekérdezéseket használva egy műveletet hajt végre (%(name)s helyőrzők egy értékszótárral), az SQL-injektálás megelőzése érdekében. Ez a szétválasztás megkönnyíti az üzleti logika tesztelését és újrahasznosítását.

Készíts 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 alkalmazás

Készíts main.py

A fő modul mindent összeköt. Minden útvonal deklarálja cursor = Depends(get_db_dependency), ami azt mondja a FastAPI-nak, hogy hívja meg a generátort, adja át a megadott kurzort a kezelőnek, majd tisztítsa meg azt. A FastAPI a kérés testeit is érvényesíti a Pydantic sémáidhoz képest, mielőtt a kezelő elindulna.

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

Az alkalmazás futtatása

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

A szerver a(z) http://localhost:8000 címen figyel. Tartsd működésben ezt a terminált, amíg az API-t használod.

Az API gyakorlása

Nyissa meg http://localhost:8000/docs egy böngészőben. A FastAPI minden útvonalhoz interaktív dokumentációt jelenít meg.

Kibővítsd a GET /health gombot, válaszd ki a Próbáld ki, majd válaszd Execute. Ellenőrizd, hogy a válasz állapotkóddal 200 rendelkezik, és egészséges adatbázis-kapcsolatot jelent.
Bontsd ki a GET /products elemet, válaszd a Try it out lehetőséget, állítsd a(z) page_size értékét erre: 5, majd válaszd az Execute lehetőséget. A válasz öt terméket és a lapozási adatokat tartalmazza.
Másolj egy id értéket a válaszból. Kibővítsd a GET /products/{product_id} gombot, válaszd ki a Próbáld ki, írd be a másolt értéket , product_idmajd válaszd az Execute.
Bontsd ki a GET /products/search/ elemet, válaszd a Próbáld ki lehetőséget, írj be egy keresőkifejezést, például a(z) bike kifejezést a(z) q elemhez, majd válaszd az Execute lehetőséget.

Tesztelni és telepíteni az alkalmazást

Hibakezeléshez, kapcsolati poolinghez, hitelesítéshez, teszteléshez és telepítéshez vonatkozó útmutatásért lásd: Tesztelj és telepítsd a FastAPI alkalmazásokat mssql-python-val.

Visszajelzés

Hasznosnak találta ezt az oldalt?

Utolsó frissítés időpontja: 2026-10-19