Notatka
Dostęp do tej strony wymaga autoryzacji. Może spróbować zalogować się lub zmienić katalogi.
Dostęp do tej strony wymaga autoryzacji. Możesz spróbować zmienić katalogi.
Po zbudowaniu aplikacji FastAPI z mssql-python, skonfiguruj ją do wdrożenia, ponownego użycia połączenia, obsługi błędów, uwierzytelniania i automatycznego testowania.
Wymagania wstępne
Całkowicie użyj mssql-python z FastAPI lub posiadaj odpowiednią aplikację FastAPI korzystającą z przykładowej bazy danych AdventureWorksLT. Zależność uwierzytelniania opisana w tym artykule odpytuje
SalesLT.Customer.Instaluj zależności produkcyjne i testowe:
pip install pydantic-settings pyjwt pytest httpx
Konfigurowanie ustawień wdrażania
Użyj ustawień Pydantic do ładowania wartości specyficznych dla wdrożenia ze zmiennych środowiskowych. Takie podejście chroni tajemnice przed kodem źródłowym i nadaje każdemu środowisku własną bazę danych, pulę oraz konfigurację uwierzytelniania.
Utwórz config.py:
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_server: str
database_name: str
pool_size: int = 20
pool_idle_timeout: int = 300
jwt_secret: str
settings = Settings()
def get_connection_string() -> str:
return (
f"Server={settings.database_server};"
f"Database={settings.database_name};"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes"
)
Ustaw DATABASE_SERVER, DATABASE_NAME oraz JWT_SECRET w środowisku wdrożeniowym. Ustawienia Pydantic automatycznie odczytują nazwy zmiennych środowiskowych wielkimi literami.
Note
ActiveDirectoryDefault próbuje użyć kolejnych dostawców poświadczeń. W produkcji określ tryb uwierzytelniania dla wdrożonej tożsamości, na przykład ActiveDirectoryMSI dla tożsamości zarządzanej, aby uniknąć przechodzenia przez łańcuch poświadczeń. Aby uzyskać dostępne tryby, zobacz Microsoft Entra authentication with mssql-python.
Konfigurowanie puli połączeń
MSSQL-Python domyślnie umożliwia pulowanie połączeń. Skonfiguruj pulę raz, zanim aplikacja utworzy pierwsze połączenie. Dostosuj rozmiar puli do przewidywanego współbieżnego obciążenia bazy danych przez aplikację oraz do warstwy usługi bazy danych.
Zaktualizuj database.py, aby używał ustawień wdrożenia:
from collections.abc import Generator
import mssql_python
from config import get_connection_string, settings
mssql_python.pooling(
max_size=settings.pool_size,
idle_timeout=settings.pool_idle_timeout,
)
def get_db_dependency() -> Generator:
with mssql_python.connect(get_connection_string()) as conn:
with conn.cursor() as cursor:
yield cursor
Menedżer kontekstu połączenia zatwierdza transakcję po pomyślnym przetworzeniu żądania, wycofuje ją, gdy podczas przetwarzania żądania zostanie zgłoszony wyjątek, i zamyka połączenie. Zamknięcie połączenia powoduje jego zwrot do puli. Informacje o kluczach puli, ustalaniu rozmiaru, izolacji tożsamości oraz wskazówki dotyczące wyczerpania puli znajdziesz w artykule Pula połączeń w mssql-python.
Obsługa błędów bazy danych
Rejestruj obsługiwacze wyjątków, aby awarie bazy danych zwracały spójne odpowiedzi bez ujawniania szczegółów połączenia, zapytań czy tekstu błędu serwera.
Dodaj programy obsługi po app = FastAPI(...) w main.py:
import mssql_python
from fastapi import Request
from fastapi.responses import JSONResponse
@app.exception_handler(mssql_python.IntegrityError)
async def integrity_exception_handler(
request: Request,
exc: mssql_python.IntegrityError,
):
return JSONResponse(
status_code=409,
content={
"detail": "The request conflicts with existing data.",
"type": "integrity_error",
},
)
@app.exception_handler(mssql_python.DatabaseError)
async def database_exception_handler(
request: Request,
exc: mssql_python.DatabaseError,
):
return JSONResponse(
status_code=500,
content={
"detail": "A database operation failed.",
"type": "database_error",
},
)
Przed zwróceniem odpowiedzi zarejestruj wyjątek przez chroniony potok telemetryczny aplikacji. Dla hierarchii wyjątków i obsługi SQLSTATE zobacz Obsługa błędów i kody SQLSTATE dla mssql-python.
Dodaj zależności uwierzytelniania
Łańcuch zależności FastAPI, aby zweryfikować JSON Web Token (JWT), załadować dopasowanego klienta AdventureWorksLT i udostępnić go na chronionych trasach. Zweryfikuj token przed nawiązaniem połączenia z bazą danych, aby nieprawidłowy token nie korzystał z połączenia pulowego.
Utwórz auth.py:
import jwt
from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from config import settings
from database import get_db_dependency
security = HTTPBearer()
def get_customer_id(
credentials: HTTPAuthorizationCredentials = Depends(security),
) -> int:
try:
payload = jwt.decode(
credentials.credentials,
settings.jwt_secret,
algorithms=["HS256"],
)
customer_id = int(payload["sub"])
except (KeyError, 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")
return customer_id
def get_current_customer(
customer_id: int = Depends(get_customer_id),
cursor = Depends(get_db_dependency),
):
cursor.execute(
"""
SELECT CustomerID, FirstName, LastName
FROM SalesLT.Customer
WHERE CustomerID = %(id)s
""",
{"id": customer_id},
)
customer = cursor.fetchone()
if customer is None:
raise HTTPException(status_code=401, detail="Customer not found")
return {
"id": customer.CustomerID,
"first_name": customer.FirstName,
"last_name": customer.LastName,
}
Importuj zależność i dodaj chronioną trasę do main.py:
from auth import get_current_customer
@app.get("/me")
def get_me(current_customer: dict = Depends(get_current_customer)):
return current_customer
Użyj dostawcy tożsamości do wydawania i wymiany kluczy do podpisywania. Dla HS256 ustaw JWT_SECRET na co najmniej 32 losowe bajty. Nie przechowuj sekretu podpisu produkcyjnego w repozytorium ani w obrazie.
Testowanie aplikacji
TestClient FastAPI wysyła żądania do aplikacji bez uruchamiania serwera HTTP. Poniższe testy integracyjne wykorzystują skonfigurowaną bazę danych.
Utwórz test_api.py:
import uuid
from fastapi.testclient import TestClient
from main import app
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]
response = client.post(
"/products",
json={
"name": f"Test Product {suffix}",
"product_number": f"TEST-{suffix}",
"price": 19.99,
"color": "Red",
"size": "M",
"category_id": 1,
},
)
assert response.status_code == 201
data = response.json()
assert data["product_number"] == f"TEST-{suffix}"
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"
Uruchom testy z korzenia projektu:
pytest
Testy te wykorzystują skonfigurowaną bazę danych i test_create_product wstawiają wiersz do SalesLT.Product. Używaj dedykowanej bazy testowej i resetuj jej dane między testami.
Lista kontrolna wdrożenia
- Ustaw
DATABASE_SERVER,DATABASE_NAMEiJWT_SECRETza pomocą magazynów sekretów i konfiguracji platformy wdrożeniowej. - Używaj dedykowanej tożsamości Microsoft Entra z minimalnymi wymaganymi uprawnieniami do bazy danych.
- Ustaw rozmiar puli poniżej limitu łączeń bazy danych i pozostawij pojemność na dostęp administracyjny i inne obciążenia.
- Uruchom testy integracji bazy danych na izolowanej bazie testowej.
- Konfiguruj chronioną telemetrię pod kątem wyjątków bazy danych, opóźnień żądań oraz wyczerpania puli.
- Uruchom Uvicorn bez
--reloadw środowiskach wdrożonych.