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.

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 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

Membuat database SQL

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

Membuat database SQL dalam hitungan menit menggunakan portal Microsoft Azure

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.

Windows
Linux/macOS

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

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

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

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
└── 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 membuka koneksi, menyediakan kursor, dan menggunakan manajer konteks koneksi mssql-python untuk melakukan commit jika berhasil, melakukan rollback jika terjadi pengecualian, dan menutup koneksi.

Buat database.py

get_connection_string()Fungsi tersebut membentuk string koneksi ODBC dari nilai konfigurasi. FastAPI memanggil Depends()get_db_dependency() sekali per permintaan dan mengelola siklus hidupnya.

# 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 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.

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:
        raise HTTPException(status_code=503, detail="Database unavailable")

Jalankan aplikasi

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

Server mendengarkan pada http://localhost:8000. Jaga terminal ini tetap berjalan saat Anda menjalankan API.

Gunakan API

Buka http://localhost:8000/docs di browser. FastAPI menampilkan dokumentasi interaktif untuk setiap rute.

Perluas GET /health, pilih Coba, lalu pilih Eksekusi. Verifikasi bahwa respons memiliki kode 200 status dan melaporkan koneksi database yang sehat.
Perluas GET /products, pilih Coba, atur page_size ke 5, lalu pilih Eksekusi. Respons tersebut berisi lima produk dan detail paginasi.
Salin nilai id dari respons. Perluas GET /products/{product_id}, pilih Coba keluar, masukkan nilai yang disalin untuk product_id, lalu pilih Jalankan.
Perluas GET /products/search/, pilih Coba, masukkan istilah pencarian seperti bike untuk q, lalu pilih Jalankan.

Uji dan terapkan aplikasi

Untuk panduan tentang penanganan kesalahan, pooling koneksi, autentikasi, pengujian, dan deployment, lihat Uji dan sebarkan aplikasi FastAPI dengan mssql-python.

Saran dan Komentar

Apakah halaman ini membantu?

Terakhir diperbarui pada 2026-10-19