Gunakan mssql-python dengan FastAPI

FastAPI adalah kerangka kerja web Python modern untuk membangun API. Dikombinasikan dengan mssql-python, Anda dapat membangun REST API berperforma tinggi yang didukung oleh Microsoft SQL dan Azure SQL Database.

Prasyarat

  • Python 3.10 atau yang lebih baru.
  • Paket mssql-python, fastapi, uvicorn, pydantic, dan PyJWT. Instal semuanya dengan pip install fastapi uvicorn mssql-python pydantic pyjwt.
  • Instal prasyarat khusus sistem operasi satu kali. Pengguna Windows dapat melewati langkah ini. Untuk detail platform lengkap, lihat Menginstal mssql-python.
    apk add libtool krb5-libs krb5-dev

Membuat database SQL

Buat atau sambungkan ke database SQL di salah satu platform berikut:

Contoh dalam artikel ini menggunakan database sampel AdventureWorksLT , khususnya SalesLT.Product tabel. Jika Anda belum menginstal AdventureWorksLT, lihat Database sampel AdventureWorks.

Penyusunan proyek

Membuat lingkungan virtual

Buat dan aktifkan lingkungan virtual sehingga paket proyek ini tetap terisolasi dari instalasi Python lainnya. Langkah ini juga mencegah masalah umum menginstal paket ke satu penerjemah saat menjalankan aplikasi atau pengujian dengan penerjemah lain.

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

Setelah Anda mengaktifkan lingkungan, python, pip, dan pytest semuanya merujuk ke interpreter yang sama. Jalankan perintah yang tersisa dalam artikel ini dari lingkungan yang diaktifkan.

Note

Pada Windows di Arm, buat lingkungan dengan build Arm64 dari Python agar mssql-python dan dependensinya diinstal dari wheel yang sudah dibuat sebelumnya. Pada komputer dengan lebih dari satu versi Python, py -m venv mungkin memilih versi atau arsitektur yang berbeda dari yang Anda harapkan, jadi verifikasi dengan python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())" setelah Anda mengaktifkan. Jika pip mencoba mengompilasi cryptography dari kode sumber (galat pada toolchain Rust dan OpenSSL), instal terlebih dahulu versi berbasis wheel dengan pip install --only-binary=:all: cryptography, lalu instal sisanya.

Pasang dependensi

Instal paket yang diperlukan dengan pip:

pip install fastapi uvicorn mssql-python pydantic pyjwt

Struktur proyek

Atur proyek Anda dengan modul terpisah untuk operasi database, skema, dan CRUD:

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

Manajemen koneksi database

FastAPI menggunakan injeksi dependensi untuk menyediakan sumber daya seperti koneksi basis data kepada handler rute. Pola di bagian ini membuat context manager yang membuka koneksi, mengembalikan kursor, dan menangani commit/rollback/close secara otomatis.

Buat database.py

get_connection_string()Fungsi tersebut membentuk string koneksi ODBC dari nilai konfigurasi. Manajer konteks get_db() dan generator get_db_dependency() sama-sama mengikuti pola yang sama: membuka koneksi, menghasilkan kursor, mengonfirmasi transaksi jika berhasil, membatalkan transaksi jika terjadi kesalahan, dan selalu menutup koneksi setelah selesai. FastAPI memanggil Depends()get_db_dependency() sekali per permintaan dan mengelola siklus hidupnya.

# database.py
import mssql_python
from contextlib import contextmanager
from typing 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 menggunakan DefaultAzureCredential, yang mencoba beberapa penyedia kredensial secara berurutan. Koneksi pertama bisa lambat karena SDK menelusuri rantai hingga menemukan penyedia yang berfungsi. Dalam lingkungan produksi, jika Anda mengetahui jenis kredensial yang digunakan oleh lingkungan Anda, tentukan secara langsung (misalnya, ActiveDirectoryMSI untuk identitas terkelola) agar terhindar dari penelusuran berantai. Untuk informasi selengkapnya, lihat Autentikasi Microsoft Entra.

@contextmanager
def get_db() -> Generator:
    """Database connection context manager for FastAPI dependency injection."""
    conn = mssql_python.connect(get_connection_string())
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()

def get_db_dependency():
    """FastAPI dependency for database cursor."""
    conn = mssql_python.connect(get_connection_string())
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()

Model Pydantic

Model Pydantic menentukan aturan bentuk dan validasi untuk data permintaan dan respons. FastAPI menggunakan model ini untuk mengurai JSON yang masuk, memvalidasi batasan bidang, dan menghasilkan dokumentasi OpenAPI secara otomatis.

Buat schemas.py

Pisahkan skema menjadi Base, Create, Update, dan varian respons. Skema Base berisi bidang umum, Create mewarisinya untuk operasi penyisipan, dan Update menjadikan semua bidang opsional untuk pembaruan parsial.

# 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

Operasi CRUD

Enkapsulasikan kueri basis data dalam kelas khusus agar handler rute tetap ringkas. Setiap metode statis mengambil kursor (disuntikkan oleh FastAPI) dan menangani satu operasi menggunakan kueri berparameter (%(name)s placeholder dengan kamus nilai) untuk mencegah injeksi SQL. Pemisahan ini membuat logika bisnis lebih mudah diuji dan digunakan kembali.

Buat 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()]

Aplikasi FastAPI

Buat main.py

Modul utama menyambungkan semuanya. Setiap rute mendeklarasikan cursor = Depends(get_db_dependency), yang memberi tahu FastAPI untuk memanggil generator, meneruskan kursor yang dihasilkan ke handler, dan membersihkan setelahnya. FastAPI juga memvalidasi isi permintaan terhadap skema Pydantic Anda sebelum handler berjalan.

# 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 as e:
        raise HTTPException(status_code=503, detail=f"Database unhealthy: {str(e)}")

Jalankan aplikasi

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

Penanganan kesalahan

FastAPI memungkinkan Anda mendaftarkan penanganan pengecualian global untuk jenis pengecualian tertentu. Saat Anda menangkap mssql_python.DatabaseError dan mssql_python.IntegrityError, FastAPI mengembalikan kesalahan JSON terstruktur dengan kode status HTTP yang sesuai, bukan respons 500 generik.

Penangan pengecualian global

Tambahkan penangan ini ke main.py, tepat setelah baris app = FastAPI(...). FastAPI menjalankan handler yang cocok setiap kali rute memunculkan jenis pengecualian itu, jadi Anda tidak memerlukan try/except blok di setiap rute.

# main.py
from fastapi import Request
from fastapi.responses import JSONResponse
import mssql_python

@app.exception_handler(mssql_python.DatabaseError)
async def database_exception_handler(request: Request, exc: mssql_python.DatabaseError):
    """Handle database errors globally."""
    return JSONResponse(
        status_code=500,
        content={"detail": "Database error occurred", "type": "database_error"}
    )

@app.exception_handler(mssql_python.IntegrityError)
async def integrity_exception_handler(request: Request, exc: mssql_python.IntegrityError):
    """Handle integrity constraint violations."""
    error_msg = str(exc)
    
    if "UNIQUE" in error_msg:
        return JSONResponse(
            status_code=409,
            content={"detail": "Resource already exists", "type": "duplicate_error"}
        )
    elif "FOREIGN KEY" in error_msg:
        return JSONResponse(
            status_code=400,
            content={"detail": "Referenced resource not found", "type": "reference_error"}
        )
    
    return JSONResponse(
        status_code=400,
        content={"detail": "Data integrity error", "type": "integrity_error"}
    )

Note

Menghapus produk yang masih direferensikan baris lain akan diangkat mssql_python.IntegrityError dari batasan kunci asing, dan handler mengembalikan 400 alih-alih menghapus baris. Dalam sampel AdventureWorksLT, sebagian besar produk di SalesLT.Product direferensikan oleh SalesLT.SalesOrderDetail, sehingga DELETE gagal untuk produk tersebut sesuai rancangan. Untuk menguji penghapusan yang berhasil, buat produk dengan POST /products dan hapus produk tersebut, atau hapus baris referensi terlebih dahulu.

Pemanfaatan koneksi

Tanpa pengumpulan koneksi, setiap permintaan membuka dan menutup koneksi TCP ke Microsoft SQL, yang menambahkan latensi. Pengumpulan koneksi membuat satu set koneksi menganggur siap untuk digunakan kembali. Panggil mssql_python.pooling() sekali saat memulai. Dengan pooling diaktifkan, conn.close() in get_db_dependency() mengembalikan koneksi ke kumpulan alih-alih benar-benar menutupnya.

Modul database yang disempurnakan

Aktifkan pengumpulan dengan memanggil mssql_python.pooling() saat startup dan konfigurasikan dengan ukuran maksimum dan pengaturan batas waktu yang sesuai:

# database.py with connection pooling
import mssql_python
from contextlib import contextmanager
import os

# Configure pool
mssql_python.pooling(max_size=20, idle_timeout=300)

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "Server=<server>.database.windows.net;Database=<database>;"
    "Authentication=ActiveDirectoryDefault;Encrypt=yes"
)

def get_db_dependency():
    """FastAPI dependency with connection pooling."""
    conn = mssql_python.connect(DATABASE_URL)
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()  # Returns to pool

Middleware autentikasi

Anda dapat menggabungkan akses database dengan autentikasi dengan merantai dependensi FastAPI. Contoh berikut memvalidasi token pembawa JWT, mencari rekaman orang yang cocok dalam database sampel AdventureWorksLT, dan membuat hasilnya tersedia untuk rute yang dilindungi.

# auth.py
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    cursor = Depends(get_db_dependency)
):
    """Validate JWT and return the matching AdventureWorksLT person."""
    try:
        token = credentials.credentials
        # Replace with a strong secret loaded from environment variables
        payload = jwt.decode(token, "your-secret-key", algorithms=["HS256"])
        person_id = int(payload.get("sub"))
        
        if not person_id:
            raise HTTPException(status_code=401, detail="Invalid token")
        
        cursor.execute("""
            SELECT BusinessEntityID, FirstName, LastName
            FROM Person.Person
            WHERE BusinessEntityID = %(id)s
        """, {"id": person_id})
        
        person = cursor.fetchone()
        if not person:
            raise HTTPException(status_code=401, detail="User not found")
        
        return {
            "id": person.BusinessEntityID,
            "first_name": person.FirstName,
            "last_name": person.LastName
        }
        
    except (TypeError, ValueError):
        raise HTTPException(status_code=401, detail="Invalid token subject")
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

# Protected endpoint
@app.get("/me")
def get_me(current_user: dict = Depends(get_current_user)):
    return current_user

Testing

FastAPI menyediakan TestClient yang dibangun di atas httpx yang mengirim permintaan ke aplikasi Anda tanpa memulai server HTTP sungguhan. Tulis pengujian dengan pytest untuk memverifikasi rute, kode status, dan bentuk respons.

Sebelum menjalankan pengujian di bagian ini, instal dependensi pengujian:

pip install pytest httpx

Note

Jika Anda memakai Starlette versi terbaru atau sedang menyiapkan lingkungan baru, sebaiknya gunakan httpx2 daripada httpx. Versi Starlette terbaru menggunakan httpx2 untuk TestClient dan mengeluarkan peringatan deprekasi saat hanya httpx yang diinstal. Instal itu dengan pip install pytest httpx2.

Penyiapan pengujian

Buat file pengujian yang digunakan TestClient untuk memverifikasi perilaku rute dan skema respons:

# test_api.py
from fastapi.testclient import TestClient
from main import app
import uuid
import pytest

client = TestClient(app)

def test_list_products():
    response = client.get("/products")
    assert response.status_code == 200
    data = response.json()
    assert "items" in data
    assert "total" in data

def test_create_product():
    suffix = uuid.uuid4().hex[:8]
    name = f"Test Product {suffix}"
    product_data = {
        "name": name,
        "product_number": f"TEST-{suffix}",
        "price": 19.99,
        "color": "Red",
        "size": "M",
        "category_id": 1
    }
    response = client.post("/products", json=product_data)
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == name
    assert data["price"] == 19.99

def test_get_product_not_found():
    response = client.get("/products/99999")
    assert response.status_code == 404

def test_health_check():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

Jalankan pengujian dengan pytest dari root proyek, direktori yang sama dengan:main.py

pytest

Pengujian ini dijalankan pada database aktif Anda, bukan mock, jadi test_create_product menyisipkan baris sungguhan ke SalesLT.Product. Di AdventureWorksLT, keduanya Name dan ProductNumber memiliki batasan unik, sehingga pengujian menghasilkan nilai unik untuk masing-masing pada setiap eksekusi. Jika Anda menetapkan nilai-nilai tersebut secara hardcode sebagai gantinya, pengujian akan gagal karena konflik pada eksekusi kedua kecuali Anda menghapus baris tersebut terlebih dahulu.

Konfigurasi penerapan

Gunakan BaseSettings milik Pydantic untuk memuat konfigurasi dari variabel lingkungan dan file .env. Pendekatan ini menjauhkan rahasia dari kode sumber dan memudahkan untuk beralih antar lingkungan. Instal paket pengaturan dengan pip install pydantic-settings.

Variabel lingkungan

Buat modul pengaturan yang memuat konfigurasi dari variabel lingkungan, memungkinkan Anda mengelola rahasia dan nilai khusus penyebaran di luar kode Anda:

# config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    database_server: str = "<server>.database.windows.net"
    database_name: str = "<database>"
    pool_size: int = 10

    model_config = SettingsConfigDict(env_file=".env")

settings = Settings()

def get_connection_string() -> str:
    return (
        f"Server={settings.database_server};"
        f"Database={settings.database_name};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

Kemudian, perbarui database.py untuk mengimpor get_connection_string dari config alih-alih mendefinisikan salinannya sendiri. Dengan menghapus fungsi duplikat, Anda memastikan aplikasi membaca pengaturan koneksi dari satu sumber.

# database.py
from config import get_connection_string

Konten terkait

  • Last updated on