Notitie
Voor toegang tot deze pagina is autorisatie vereist. U kunt proberen u aan te melden of de directory te wijzigen.
Voor toegang tot deze pagina is autorisatie vereist. U kunt proberen de mappen te wijzigen.
Nadat je een FastAPI-applicatie met mssql-python hebt gebouwd, configureer je deze voor deployment, hergebruik van verbindingen, foutafhandeling, authenticatie en geautomatiseerd testen.
Prerequisites
Voltooi Gebruik mssql-python met FastAPI, of heb een gelijkwaardige FastAPI-applicatie die de AdventureWorksLT voorbeelddatabase gebruikt. De verificatieafhankelijkheid in dit artikel bevraagt
SalesLT.Customer.Installeer de productie- en testafhankelijkheden:
pip install pydantic-settings pyjwt pytest httpx
Implementatie-instellingen configureren
Gebruik Pydantic Settings om implementatie-specifieke waarden van omgevingsvariabelen te laden. Deze aanpak houdt geheimen buiten de broncode en geeft elke omgeving een eigen database, pool en authenticatieconfiguratie.
Maak 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"
)
Stel DATABASE_SERVER, DATABASE_NAME, en JWT_SECRET in de deployment-omgeving. Pydantic Settings leest automatisch de hoofdletters van de omgevingsvariabelen.
Note
ActiveDirectoryDefault Probeert meerdere credentialproviders achter elkaar. In productie specificeer je de authenticatiemodus voor de geïmplementeerde identiteit, bijvoorbeeld ActiveDirectoryMSI voor beheerde identiteit, om te voorkomen dat je de inlogketen loopt. Voor beschikbare modi, zie Microsoft Entra-authenticatie met mssql-python.
Groepsgewijze verbindingen configureren
MSSQL-python maakt standaard verbindingspooling mogelijk. Configureer de pool één keer, voordat de applicatie zijn eerste verbinding maakt. Schaal de pool op voor het verwachte gelijktijdige databasewerk van de applicatie en de databaseservicelaag.
Werk database.py bij om de implementatie-instellingen te gebruiken:
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
De verbindingscontextmanager voert na succesvolle verwerking van het verzoek een commit door, rolt de transactie terug wanneer er tijdens de verwerking van het verzoek een uitzondering wordt opgegooid, en sluit de verbinding. Door de verbinding te sluiten, wordt deze teruggegeven aan de pool. Voor poolsleutels, grootte, identiteitsisolatie en uitputtingsrichtlijnen, zie Connection pooling met mssql-python.
Databasefouten behandelen
Registreer uitzonderingshandlers zodat databasefouten consistente antwoorden teruggeven zonder verbindingsdetails, queries of serverfouttekst bloot te stellen.
Voeg de handlers toe in app = FastAPI(...)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",
},
)
Registreer de uitzondering via de beschermde telemetrie-pijplijn van je applicatie voordat je de respons retourneert. Voor de uitzonderingshiërarchie en SQLSTATE-afhandeling, zie Foutbehandeling en SQLSTATE-codes voor mssql-python.
Voeg authenticatieafhankelijkheden toe
Koppel FastAPI-afhankelijkheden om een JSON Web Token (JWT) te valideren, laad de bijbehorende AdventureWorksLT-klant en maak die klant beschikbaar voor beschermde routes. Valideer de token voordat je een databaseverbinding verkrijgt, zodat een ongeldig token geen poolverbinding gebruikt.
Maak 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,
}
Importeer de afhankelijkheid en voeg een beschermde route toe aan main.py:
from auth import get_current_customer
@app.get("/me")
def get_me(current_customer: dict = Depends(get_current_customer)):
return current_customer
Gebruik een identiteitsprovider om signeringssleutels uit te geven en periodiek te vervangen. Voor HS256 stel JWT_SECRET je in op minstens 32 willekeurige bytes. Sla geen signinggeheim van een productie op in de repository of in een afbeelding.
De toepassing testen
FastAPI stuurt TestClient verzoeken naar de applicatie zonder een HTTP-server te starten. De volgende integratietests gebruiken de geconfigureerde database.
Maak 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"
Voer de tests uit vanaf de projectroot:
pytest
Deze tests gebruiken de geconfigureerde database en test_create_product voegen een rij in SalesLT.Product. Gebruik een speciale testdatabase en reset de data tussen testruns door.
Controlelijst voor implementatie
- Stel
DATABASE_SERVER,DATABASE_NAMEenJWT_SECRETin via de secret- en configuratieopslag van het deploymentplatform. - Gebruik een speciale Microsoft Entra-identiteit met de minimaal vereiste database-rechten.
- Stel de poolgrootte in onder de verbindingslimiet van de database en laat capaciteit over voor administratieve toegang en andere workloads.
- Voer database-integratietests uit tegen een geïsoleerde testdatabase.
- Configureer beschermde telemetrie voor database-uitzonderingen, verzoeklatentie en pooluitputting.
- Speel Uvicorn zonder
--reloadin ingezette omgevingen.