Используйте mssql-python с FastAPI

FastAPI — это современный веб-фреймворк на Python для создания API. В сочетании с mssql-python вы можете создавать высокопроизводительные REST API, поддерживаемые Microsoft SQL и База данных SQL Azure.

Необходимые условия

Python 3.10 или более поздней версии.
Установите единовременные предварительные условия для операционной системы. Пользователи Windows могут пропустить этот шаг. Для полной информации о платформе см. Установить 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-базе данных на одной из следующих платформ:

Создание базы данных SQL в минутах с помощью портала Azure

В примерах этой статьи используется пример базы данных 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 создайте среду, используя сборку Python для Arm64, чтобы mssql-python и его зависимости устанавливались из предварительно собранных wheel-пакетов. На компьютере, где установлено несколько версий Python, py -m venv может выбрать не ту версию или архитектуру, которую вы ожидаете, поэтому после активации проверьте это с помощью python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())". Если pip пытается собрать cryptography из исходников (ошибка в цепочке инструментов Rust и OpenSSL), сначала установите версию из wheel-пакета с помощью 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() один раз на запрос и управляет его жизненным циклом.

# 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 идёт по цепочке, пока не найдёт работающего провайдера. В продакшене, если вы знаете, какой тип учетных данных использует ваша среда, укажите его напрямую (например, 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 плейсхолдеров со словарём значений), чтобы предотвратить 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

Протестировать и развернуть приложение

Используйте сопутствующую статью, чтобы завершить заявление:

Обработка ошибок

Сопутствующая статья посвящена обработке исключений в базе данных.

Глобальный обработчик исключений

См. Обработка ошибок базы данных.

Пулинг соединений

В сопутствующей статье рассматривается конфигурация пула соединений.

Расширенный модуль базы данных

См. Настройка пула соединений.

Промежуточное программное обеспечение аутентификации

См. Добавить зависимости аутентификации.

Тестирование

Сопутствующая статья посвящена интеграционному тестированию.

Тестовая настройка

См. Тестировать заявку.

Конфигурация развертывания

Сопутствующая статья охватывает конфигурацию развертывания и операции.

Переменные среды

См. Настройки развертывания и контрольный список развертывания.

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-09-23