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

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

Prerequisites

Python 3.10 以降。
オペレーティングシステム固有の 1 回限りの前提条件をインストールします。 Windowsユーザーはこのステップをスキップできます。プラットフォームの詳細は「Install mssql-python」をご覧ください。
- アルパイン
- Debian/Ubuntu
- RHEL
- SUSE
- openSUSE
- macOS
```
apk add libtool krb5-libs krb5-dev
```
```
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 データベースを作成する

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

Azure portal を使用して数分で SQL データベースを作成する

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

プロジェクトの設定

仮想環境を作成する

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

Windows
Linux または macOS

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

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

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

Note

Windows on Arm では、Arm64 版の Python を使って環境を作成し、mssql-python とその依存関係がビルド済みホイールからインストールされるようにします。複数のPythonバージョンがあるマシンでは、py -m venv予想と異なるバージョンやアーキテクチャを選択することがあるので、アクティベート後に python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())" で確認してください。もし pip ソースから cryptography ビルドしようとした場合(RustやOpenSSLのツールチェーンエラー)、まず pip install --only-binary=:all: cryptographyでホイールバック版をインストールし、その後残りをインストールしてください。

依存関係のインストール

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

pip install fastapi uvicorn mssql-python pydantic

プロジェクト構造

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

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

データベース接続管理

FastAPIは依存性注入を用いて、データベース接続などのリソースをルーティングハンドラに提供します。このセクションのパターンは、接続を開き、カーソルを生成し、mssql-pythonの接続コンテキストマネージャーを使って成功時にコミットし、例外でロールバックし、接続を閉じます。

database.py 作成

get_connection_string()関数は設定値からODBC 接続文字列を構築します。 FastAPI はリクエストごとに Depends()get_db_dependency() を1回呼び出し、そのライフサイクルを管理します。

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

Note

ActiveDirectoryDefault は DefaultAzureCredential を使用します。これは、複数の認証情報プロバイダーを順番に試行します。最初の接続は遅くなることがあります。なぜならSDKが動作するプロバイダーを見つけるまでチェーンを歩くからです。本番環境では、環境がどの認証情報タイプを使っているか分かっているなら、チェーンウォークを避けるために直接指定してください(例えばマネージドIDの ActiveDirectoryMSI )。詳細については、Microsoft Entra 認証に関するページを参照してください。

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

ピダンティックモデル

ピダンティックモデルは、要求データと応答データの形状と検証ルールを定義します。 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:
        raise HTTPException(status_code=503, detail="Database unavailable")

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

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

アプリケーションのテストとデプロイ

補助記事を使って応募書類を完成させてください:

エラー処理

この補随記事ではデータベース例外処理について扱っています。

グローバル例外ハンドラ

「データベースエラーの処理」を参照してください。

コネクションプーリング

関連記事では接続プールの構成について説明しています。

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

詳細は「接続プーリングの設定」を参照してください。

認証ミドルウェア

「認証依存関係を追加する」を参照してください。

Testing

この関連記事では統合テストについて扱っています。

テストセットアップ

「 Test the application」を参照してください。

デプロイの構成

伴随記事では展開設定と運用について扱っています。

環境変数

「デプロイメント設定の設定」と「デプロイメントチェックリスト」をご覧ください。

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-09-23