Użyj mssql-python z FastAPI

FastAPI to nowoczesny framework webowy Python do tworzenia API. W połączeniu z mssql-python możesz budować wysokowydajne REST API wspierane przez Microsoft SQL i Azure SQL Database.

Wymagania wstępne

  • Python 3.10 lub nowszy.
  • Pakiety mssql-python, fastapi, uvicorn, pydantic oraz PyJWT. Zainstaluj wszystko za pomocą pip install fastapi uvicorn mssql-python pydantic pyjwt.
  • Zainstaluj jednorazowe wymagania wstępne dotyczące systemu operacyjnego. Użytkownicy Windows mogą pominąć ten krok. Pełne szczegóły dotyczące platformy można znaleźć w artykule Install mssql-python.
    apk add libtool krb5-libs krb5-dev

Tworzenie bazy danych SQL

Stwórz lub połącz się z bazą danych SQL na jednej z następujących platform:

Przykłady w tym artykule korzystają z przykładowej bazy danych AdventureWorksLT , a konkretnie z tabeli SalesLT.Product . Jeśli nie masz zainstalowanego AdventureWorksLT, zobacz przykładowe bazy danych AdventureWorks.

Konfiguracja projektu

Tworzenie środowiska wirtualnego

Stwórz i aktywuj środowisko wirtualne, aby pakiety tego projektu pozostały odizolowane od innych instalacji Python. Ten krok zapobiega również częstemu problemowi instalowania pakietów w jednym interpreterze podczas uruchamiania aplikacji lub testów na innym.

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

Po aktywowaniu środowiska python, pip i pytest wszystkie wskazują na ten sam interpreter. Wykonaj pozostałe polecenia z tego artykułu z aktywowanego środowiska.

Note

W systemie Windows on Arm utwórz środowisko przy użyciu kompilacji Arm64 interpretera Python, aby mssql-python i jego zależności instalowały się z prekompilowanych pakietów wheel. Na komputerze, na którym zainstalowano więcej niż jedną wersję języka Python, py -m venv może wybrać inną wersję lub architekturę, niż się spodziewasz, więc po aktywowaniu sprawdź to za pomocą python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())". Jeśli pip próbuje zbudować cryptography ze źródeł (błąd związany z łańcuchem narzędzi Rust i OpenSSL), najpierw zainstaluj wersję opartą na pakiecie wheel za pomocą pip install --only-binary=:all: cryptography, a następnie zainstaluj pozostałe pakiety.

Instalowanie zależności

Zainstaluj wymagane pakiety za pomocą pip:

pip install fastapi uvicorn mssql-python pydantic pyjwt

struktura projektu

Zorganizuj swój projekt z osobnymi modułami dotyczącymi baz danych, schematów i operacji CRUD:

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

Zarządzanie połączeniami bazy danych

FastAPI wykorzystuje wstrzykiwanie zależności do udostępniania zasobów, takich jak połączenia z bazą danych, programom obsługi tras. Opisany w tej sekcji wzorzec tworzy menedżera kontekstu, który otwiera połączenie, udostępnia kursor oraz automatycznie obsługuje zatwierdzanie transakcji, wycofywanie zmian i zamykanie połączenia.

Stwórz database.py

Funkcja get_connection_string() buduje parametry połączenia ODBC na podstawie wartości konfiguracyjnych. Menedżer kontekstu get_db() i generator get_db_dependency() działają według tego samego schematu: otwierają połączenie, zwracają kursor, zatwierdzają transakcję w przypadku powodzenia, wycofują ją w razie błędu i zawsze zamykają po zakończeniu. FastAPI Depends() wywołuje get_db_dependency() jeden raz dla każdego żądania i zarządza jego cyklem życia.

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

Note

ActiveDirectoryDefault używa DefaultAzureCredential, który testuje kolejno wielu dostawców poświadczeń. Pierwsze połączenie może być wolne, ponieważ SDK przechodzi przez łańcuch, aż znajdzie dostawcę, który działa. W środowisku produkcyjnym, jeśli wiesz, jakiego typu poświadczeń używa środowisko, wskaż go bezpośrednio (na przykład ActiveDirectoryMSI w przypadku tożsamości zarządzanej), aby uniknąć przechodzenia przez łańcuch. Aby uzyskać więcej informacji, zobacz Microsoft Entra authentication (Uwierzytelnianie w usłudze Microsoft Entra).

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

Modele pydantyczne

Modele pydantyczne definiują kształt i reguły walidacji danych żądań i odpowiedzi. FastAPI wykorzystuje te modele do analizy przychodzącego JSON, walidacji ograniczeń polowych oraz automatycznego generowania dokumentacji OpenAPI.

Stwórz schemas.py

Podziel schematy na Base, Create, Update, oraz warianty odpowiedzi. Schemat Base przechowuje pola współdzielone, Create dziedziczy z nich do operacji wstawiania i Update czyni wszystkie pola opcjonalnymi dla częściowych aktualizacji.

# 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

Operacje CRUD

Umieść zapytania do bazy danych w osobnej klasie, aby obsługa tras pozostała prosta i lekka. Każda statyczna metoda przyjmuje kursor (wstrzyknięty przez FastAPI) i wykonuje jedną operację za pomocą parametryzowanych zapytań (%(name)s zastępczych z słownikiem wartości), aby zapobiec wstrzykiwaniu SQL. To rozdzielenie ułatwia testowanie i ponowne wykorzystanie logiki biznesowej.

Stwórz 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()]

Aplikacja FastAPI

Stwórz main.py

Główny moduł łączy wszystko razem. Każda trasa deklaruje cursor = Depends(get_db_dependency), co nakazuje FastAPI wywołać generator, przekazać wybrany kursor do handlera i następnie wyczyścić. FastAPI również sprawdza poprawność treści żądania na podstawie schematów Pydantic, zanim zostanie uruchomiona funkcja obsługująca.

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

Uruchamianie aplikacji

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

Obsługa błędów

FastAPI pozwala rejestrować globalne obsługiwacze wyjątków dla konkretnych typów wyjątków. Gdy złapiesz mssql_python.DatabaseError i mssql_python.IntegrityError, FastAPI zwraca ustrukturyzowane błędy JSON z odpowiednimi kodami statusu HTTP zamiast ogólnych odpowiedzi 500.

Globalna obsługa wyjątków

Dodaj te handlery do main.py, zaraz po linii app = FastAPI(...) . FastAPI uruchamia dopasowany handler za każdym razem, gdy trasa podnosi ten typ wyjątku, więc nie potrzebujesz bloku try/except w każdej trasie.

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

Note

Usunięcie produktu, do którego inne wiersze nadal się odwołują, podnosi mssql_python.IntegrityError ograniczenie klucza obcego, a obsługiwacz zwraca 400 zamiast usuwać wiersz. W przykładzie AdventureWorksLT większość produktów w SalesLT.Product jest używana przez SalesLT.SalesOrderDetail, więc DELETE zgodnie z założeniem kończy się dla nich niepowodzeniem. Aby przetestować skuteczne usunięcie, stwórz produkt z tym POST /products i usuń go lub usuń najpierw wiersze referencyjne.

Buforowanie połączeń

Bez puli połączeń każde żądanie otwiera i zamyka połączenie TCP z Microsoft SQL, co zwiększa opóźnienia. Pula połączeń utrzymuje zestaw bezczynnych połączeń gotowych do ponownego użycia. Wywołaj mssql_python.pooling() raz podczas uruchamiania. Po włączonym poolingu conn.close() in get_db_dependency() zwraca połączenie z pulą zamiast ją faktycznie zamykać.

Rozszerzony moduł bazy danych

Włącz mechanizm puli, wywołując mssql_python.pooling() podczas uruchamiania, i skonfiguruj go za pomocą odpowiednich ustawień maksymalnego rozmiaru i limitu czasu:

# 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

Oprogramowanie pośrednicze uwierzytelniania

Możesz łączyć dostęp do bazy danych z uwierzytelnianiem, łącząc zależności FastAPI. Poniższy przykład weryfikuje token JWT typu bearer, wyszukuje pasujący rekord osoby w przykładowej bazie danych AdventureWorksLT i udostępnia wynik chronionym ścieżkom.

# 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 udostępnia TestClientklienta testowego opartego na httpx, który wysyła żądania do twojej aplikacji bez uruchamiania prawdziwego serwera HTTP. Pisz testy przy użyciu pytest, aby weryfikować trasy, kody stanu i struktury odpowiedzi.

Przed uruchomieniem testów w tej sekcji zainstaluj zależności testowe:

pip install pytest httpx

Note

Jeśli grasz na najnowszym Starlette lub konfigurujesz nowe środowisko, wolej httpx2 niż httpx. Najnowsze wersje Starlette używają httpx2 zamiast TestClient i emitują ostrzeżenie o wycofaniu, gdy zainstalowane jest tylko httpx. Zainstaluj go za pomocą polecenia pip install pytest httpx2.

Konfiguracja testu

Utwórz plik testowy, który używa TestClient do weryfikacji działania trasy i schematów odpowiedzi:

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

Uruchom testy za pomocą pytest z katalogu głównego projektu, w tym samym katalogu co main.py:

pytest

Te testy działają na działającej bazie danych, a nie na mockach, więc test_create_product wstawia prawdziwy wiersz do SalesLT.Product. W AdventureWorksLT zarówno Name, jak i ProductNumber mają ograniczenia unikalności, więc test generuje dla każdego z nich unikalną wartość przy każdym uruchomieniu. Jeśli zamiast tego zakodujesz te wartości na stałe, test kończy się konfliktem przy drugim uruchomieniu, chyba że najpierw usuniesz ten wiersz.

Konfiguracja wdrożenia

Użyj elementu BaseSettings biblioteki Pydantic do wczytywania konfiguracji ze zmiennych środowiskowych i z plików .env. Takie podejście chroni tajemnice z dala od kodu źródłowego i ułatwia przełączanie się między środowiskami. Zainstaluj pakiet ustawień za pomocą pip install pydantic-settings.

Zmienne środowiskowe

Stwórz moduł ustawień, który ładuje konfigurację ze zmiennych środowiskowych, pozwalając zarządzać sekretami i wartościami specyficznymi dla wdrożenia poza kodem:

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

Następnie zaktualizuj database.py, tak aby importować get_connection_string z config zamiast definiować własną kopię. Usuwając zduplikowaną funkcję, zapewniasz, że aplikacja odczytuje ustawienia połączenia z jednego źródła.

# database.py
from config import get_connection_string

Treści powiązane

  • Last updated on