mssql-python'u FastAPI ile kullanın

FastAPI, API oluşturmak için modern bir Python web çerçevesidir. mssql-python ile birleştiğinde, Microsoft SQL ve Azure SQL Veritabanı destekli yüksek performanslı REST API'leri oluşturabilirsiniz.

Prerequisites

Python 3.10 veya üzeri.

Tek seferlik işletim sistemine özgü önkoşulları yükleyin. Windows kullanıcıları bu adımı atlayabilir. Platformun tam detayları için mssql-python'u Install sayfasına bakınız.

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 veritabanı oluşturma

Aşağıdaki platformlardan birinde bir SQL veritabanı oluşturun veya bağlanın:

Azure portalını kullanarak dakikalar içinde SQL veritabanı oluşturma

Bu makaledeki örnekler AdventureWorksLT örnek veritabanını, özellikle SalesLT.Product tabloyu kullanır. Eğer AdventureWorksLT'niz yüklü değilse, AdventureWorks örnek veritabanlarına bakabilirsiniz.

Proje kurulumu

Sanal ortam oluşturma

Bu projenin paketlerinin diğer Python kurulumlarından izole kalması için sanal bir ortam oluşturun ve etkinleştirin. Bu adım, uygulamanızı çalıştırırken paketleri bir yorumlayıcıya yükleme veya başka bir uygulamada test yapma sorununu da önler.

Windows
Linux/macOS

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

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

Ortamı etkinleştirdikten sonra, python, pip ve pytest üçü de aynı yorumlayıcıyı işaret eder. Bu makaledeki kalan komutları etkinleştirilmiş ortamdan çalıştırın.

Note

Windows on Arm’de, ortamı Python’ın Arm64 derlemesiyle oluşturun; böylece mssql-python ve bağımlılıkları önceden oluşturulmuş wheel paketlerinden yüklenir. Birden fazla Python sürümü bulunan bir makinede, py -m venv beklediğinizden farklı bir sürümü veya mimariyi seçebilir; bu yüzden etkinleştirdikten sonra python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())" ile doğrulayın. Eğer pip, cryptography paketini kaynaktan derlemeye çalışırsa (bu bir Rust ve OpenSSL araç zinciri hatasıdır), önce pip install --only-binary=:all: cryptography ile wheel tabanlı bir sürüm yükleyin, ardından geri kalanını yükleyin.

Bağımlılıkları yükleme

Gerekli paketleri pip ile kurun:

pip install fastapi uvicorn mssql-python pydantic

Proje yapısı

Projenizi veritabanı, şemalar ve CRUD işlemleri için ayrı modüllerle organize edin:

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

Veritabanı bağlantı yönetimi

FastAPI, yönlendirme işleyicileri için veritabanı bağlantıları gibi kaynaklar sağlamak için bağımlılık enjeksiyonu kullanır. Bu bölümdeki örüntü bir bağlantı açar, bir imleç döndürür ve başarılı olduğunda değişiklikleri kaydetmek, bir özel durum oluştuğunda geri almak ve bağlantıyı kapatmak için mssql-python bağlantısının bağlam yöneticisini kullanır.

database.py oluştur

Fonksiyon, get_connection_string() yapılandırma değerlerinden ODBC bağlantı dizesi'i oluşturur. FastAPI Depends() her istek için bir kez çağrı get_db_dependency() yapar ve yaşam döngüsünü yönetir.

# 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 DefaultAzureCredential, kullanır ve bu da birden fazla kimlik doğrulama sağlayıcısını sırayla dener. İlk bağlantı yavaş olabilir çünkü SDK çalışan bir sağlayıcı bulana kadar zincirde yürür. Üretimde, ortamınızın hangi kimlik bilgisi türünü kullandığını biliyorsanız, zincirde dolaşmayı önlemek için bunu doğrudan belirtin (örneğin, yönetilen kimlik için ActiveDirectoryMSI). Daha fazla bilgi için bkz . Microsoft Entra kimlik doğrulaması.

Pydantik modeller

Pydantik modeller, talep ve yanıt verileri için şekil ve doğrulama kurallarını tanımlar. FastAPI, gelen JSON'u ayrıştırmak, alan kısıtlamalarını doğrulamak ve OpenAPI dokümantasyonunu otomatik olarak oluşturmak için bu modelleri kullanır.

schemas.py oluştur

Şemaları , Base, Create, ve yanıt varyantlarına ayırınUpdate. Şema Base paylaşılan alanları tutar, Create ekleme işlemleri için ondan miras alır ve Update kısmi güncellemeler için tüm alanları isteğe bağlı hale getirir.

# 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 operasyonları

Rota yöneticilerini ince tutmak için veritabanı sorgularını özel bir sınıfa kapsülleyin. Her statik yöntem, FastAPI tarafından enjekte edilen bir imleç alır ve SQL enjeksiyonunu önlemek için parametreli sorgular (%(name)s değer sözlüğüne sahip yer tutucular) kullanarak bir işlem uygular. Bu ayrım, iş mantığını test etmeyi ve yeniden kullanmayı kolaylaştırır.

crud.py oluşturun

# 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 uygulaması

main.py oluştur

Ana modül her şeyi birbirine bağlıyor. Her rota, cursor = Depends(get_db_dependency) bildiriminde bulunur; bu, FastAPI'ye jeneratörü çağırmasını, üretilen imleci işleyiciye iletmesini ve ardından temizliği yapmasını söyler. FastAPI ayrıca istek gövdelerini Pydantic şemalarınıza göre handler çalışmadan önce doğrular.

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

Uygulamayı çalıştırma

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

Sunucu, http://localhost:8000 üzerinde dinliyor. API'yi çalıştırırken bu terminali çalışır tutun.

API'yi çalıştırın

Tarayıcıda açın http://localhost:8000/docs . FastAPI her rota için etkileşimli dokümantasyon sunar.

GET /health seçeneğini genişletin, Deneyin ve sonra Execute seçin. Yanıtın durum kodu 200 olup olmadığını ve sağlıklı bir veritabanı bağlantısı rapor ettiğini doğrulayın.
GET /products öğesini genişletin, Try it out seçeneğini seçin, page_size değerini 5 olarak ayarlayın ve ardından Execute seçeneğini seçin. Yanıtta beş ürün ve sayfalama ayrıntıları yer alıyor.
Yanıttan bir id değer kopyalayın. GET /products/{product_id}'yi genişletin, Try it seçin, kopyalanan değeri girin product_idve ardından Execute seçin.
GET /products/search/ seçeneğini genişletin, Try it out seçeneğini belirleyin, bike için q gibi bir arama terimi girin ve ardından Execute seçeneğini belirleyin.

Uygulamayı test et ve dağıt

Hata işleme, bağlantı havuzu, kimlik doğrulama, test ve dağıtım konularında rehberlik için mssql-python ile FastAPI uygulamalarını test et ve dağıt bölümünü inceleyebilirsiniz.

Geri Bildirim

Bu sayfayı yararlı buldunuz mu?

Son güncelleştirme: 2026-10-19