mssql-pythonとFastAPIを組み合わせて使う

FastAPIはAPI構築のための最新のPythonウェブフレームワークです。 mssql-pythonと組み合わせることで、Microsoft SQLやAzure SQL Databaseでサポートされた高性能なREST APIを構築できます。

前提条件

• Python 3.10 以降。
• mssql-python、fastapi、uvicorn、pydantic、PyJWTパッケージです。 すべて pip install fastapi uvicorn mssql-python pydantic pyjwtでインストールしてください。
• オペレーティング システム固有の 1 回限りの前提条件をインストールします。 Windowsユーザーはこのステップをスキップできます。 プラットフォームの詳細は 「Install mssql-python」をご覧ください。
  • アルパイン
  • Debian/Ubuntu
  • RHEL
  • SUSE
  • openSUSE
  • macOS

  apk add libtool krb5-libs krb5-dev
  

SQL データベースを作成する

以下のいずれかのプラットフォームでSQLデータベースを作成または接続してください:

この記事の例は AdventureWorksLT のサンプルデータベース、特に SalesLT.Product テーブルを使用しています。 AdventureWorksLTをインストールしていない場合は、 AdventureWorksのサンプルデータベースをご覧ください。

プロジェクトの設定

仮想環境を作成する

このプロジェクトのパッケージが他のPythonインストールから隔離されるよう、仮想環境を作成・有効化してください。 このステップはまた、アプリやテストを別のインタプリタで実行中にパッケージをインストールするという一般的な問題を防いでいます。

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

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

環境を起動した後、 python、 pip、 pytest はすべて同じインタープリターに解析されます。 この記事の残りのコマンドは起動済みの環境から実行してください。

依存関係のインストール

pipで必要なパッケージをインストールしてください:

pip install fastapi uvicorn mssql-python pydantic pyjwt

プロジェクト構造

データベース、スキーマ、CRUD操作用の別々のモジュールでプロジェクトを整理してください:

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

データベース接続管理

FastAPIは依存性注入を用いて、データベース接続などのリソースをルーティングハンドラに提供します。 このセクションのパターンは、接続を開き、カーソルを生成し、コミット/ロールバック/クローズを自動的に処理するコンテキストマネージャーを作成します。

database.py 作成

get_connection_string()関数は設定値からODBC 接続文字列を構築します。 get_db()コンテキストマネージャーとget_db_dependency()ジェネレーターは同じパターンに従っています:接続を開き、カーソルを表示し、成功したらコミットし、エラーでロールバックし、完了すると必ず閉じます。 FastAPI はリクエストごとに Depends()get_db_dependency() を1回呼び出し、そのライフサイクルを管理します。

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

@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()

ピダンティックモデル

ピダンティックモデルは、要求データと応答データの形状と検証ルールを定義します。 FastAPIはこれらのモデルを使って、入ってくるJSONを解析し、フィールド制約を検証し、OpenAPIのドキュメントを自動的に生成します。

schemas.py を作る

スキーマを Base、 Create、 Update、応答のバリエーションに分けてください。 Baseスキーマは共有フィールドを保持し、挿入操作のためにCreate継承し、Update部分的な更新に関してはすべてのフィールドを任意にします。

# 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操作

データベースクエリを専用クラスにカプセル化し、ルートハンドラを薄く保つこと。 各静的メソッドはカーソル(FastAPIで注入)を受け取り、パラメータ 化されたクエリ (値の辞書付きプレースホルダー%(name)s )を使って1つの操作を処理し、SQLインジェネルを防ぎます。 この分離により、ビジネスロジックのテストや再利用が容易になります。

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アプリケーション

main.py を作る

メインモジュールがすべてをつなぎ合わせています。 各ルートは cursor = Depends(get_db_dependency)を宣言し、FastAPIにジェネレーターを呼び出し、得られたカーソルをハンドラーに渡し、その後クリーンアップするよう指示します。 FastAPIはハンドラを実行する前にリクエストボディをPydanticスキーマに対して検証します。

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

アプリケーションを実行する

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

エラー処理

FastAPIでは特定の例外タイプに対してグローバル例外ハンドラを登録できます。 mssql_python.DatabaseErrorやmssql_python.IntegrityErrorをキャッチすると、FastAPIは一般的な500回のレスポンスではなく、適切なHTTPステータスコード付きの構造化JSONエラーを返します。

グローバル例外ハンドラ

これらのハンドラーをmain.pyラインのすぐ後にapp = FastAPI(...)に加えてください。 FastAPIはルートがその例外タイプを発生させるたびにマッチングハンドラーを実行するため、すべてのルートに try/except ブロックは必要ありません。

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

コネクションプーリング

接続プーリングがなければ、各リクエストはMicrosoft SQLへのTCP接続を開閉し、遅延が発生します。 接続プーリングは、再利用できる状態で待機中の接続をまとめて維持します。 起動時に一度 mssql_python.pooling() 電話してください。 プーリングを有効にすると、get_db_dependency() 内の conn.close() は、実際には接続を閉じずに、接続をプールに戻します。

強化データベースモジュール

起動時に mssql_python.pooling() を呼び出してプーリングを有効にし、適切な最大サイズとタイムアウト設定を設定してください:

# 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

認証ミドルウェア

FastAPI依存関係を連鎖することで、データベースアクセスと認証を組み合わせることができます。 以下の例では、JWTベアラートークンを検証し、AdventureWorksLTサンプルデータベースでマッチングするpersonレコードを検索し、その結果を保護されたルートに提供します。

# 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は、実際のHTTPサーバーを起動せずにアプリケーションにリクエストを送信するTestClientに基づくhttpxを提供します。 ルート、ステータスコード、レスポンス形状を検証するために pytest テストを書きます。

このセクションのテストを実行する前に、テスト依存関係をインストールしてください:

pip install pytest httpx

テストセットアップ

ルートの挙動やレスポンススキーマを検証するために TestClient を使ったテストファイルを作成します:

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

プロジェクトのルートから pytest を使ってテストを実行してください。 main.pyと同じディレクトリです:

pytest

これらのテストはモックではなくライブデータベースに対して実行されるため、test_create_productSalesLT.Productに実際の行を挿入します。 AdventureWorksLTでは、 Name も ProductNumber もそれぞれ固有の制約があるため、テストは各ランでそれぞれ固有の値を生成します。 もしそれらの値をハードコードすると、最初にその行を削除しない限り、2回目の実行時に競合が発生してテストは失敗します。

デプロイの構成

Pydanticの BaseSettings を使って環境変数や .env ファイルから設定を読み込みます。 この方法はソースコードの秘密を守り、環境間の切り替えを容易にします。 pip install pydantic-settingsで設定パッケージをインストールしてください。

環境変数

環境変数から設定を読み込む設定モジュールを作成し、コード外で秘密やデプロイ固有の値を管理できるようにします:

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

その後、database.py が独自のコピーを定義するのではなく、config から get_connection_string をインポートするように更新します。 重複した機能を削除することで、アプリが単一のソースから接続設定を読み取ることが保証されます。

# database.py
from config import get_connection_string

関連するコンテンツ

• 接続管理
• 接続プール
• エラー処理
• Microsoft Entra 認証