Jak skonfigurować środowisko testowe i pisać testy jednostkowe w FastAPI?

Po skonfigurowaniu podstawowej aplikacji FastAPI można przejść do integracji testów, co jest niezbędnym krokiem, aby upewnić się, że aplikacja działa zgodnie z oczekiwaniami. Testowanie w FastAPI opiera się na frameworku pytest, który oferuje elastyczność, łatwość użycia oraz integrację z asynchronicznymi aplikacjami, takimi jak te tworzone przy pomocy FastAPI. Ważnym krokiem w tym procesie jest odpowiednia konfiguracja środowiska testowego i pisanie testów jednostkowych.

Konfiguracja środowiska testowego

css
Copy code
protoapp/
    ├── protoapp/
    │   └── main.py
    ├── tests/
    │   └── test_main.py
    └── pytest.ini

W pliku pytest.ini należy dodać konfigurację dla pytest, która zapewni, że odpowiednie katalogi zostaną dodane do ścieżki PYTHONPATH, co umożliwi prawidłowe działanie testów. Konfiguracja ta powinna wyglądać następująco:

ini
Copy code
[pytest]

pythonpath = . protoapp

Następnie w module testowym test_main.py należy napisać test dla stworzonego wcześniej punktu końcowego /home. W tym celu użyjemy klienta AsyncClient z biblioteki httpx, który będzie komunikował się z aplikacją FastAPI. Oto jak może wyglądać przykładowy test:

python
Copy code
import pytest

from httpx import ASGITransport, AsyncClient
from protoapp.main import app
@pytest.mark.asyncio
async def test_read_main():
    client = AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    )
    response = await client.get("/home")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

Ten test weryfikuje, czy punkt końcowy /home zwraca oczekiwaną odpowiedź w formie JSON i czy status odpowiedzi to 200 OK.

Aby sprawdzić, czy środowisko testowe zostało skonfigurowane poprawnie, wystarczy uruchomić polecenie:

bash
Copy code
$ pytest --collect-only

Zobaczymy wtedy informację o wykrytych testach, na przykład:

vbnet
Copy code
configfile: pytest.ini

plugins: anyio-4.2.0, asyncio-0.23.5, cov-4.1.0
asyncio: mode=Mode.STRICT
collected 1 item

Pisanie i uruchamianie testów jednostkowych

Po skonfigurowaniu środowiska testowego możemy przejść do pisania testów jednostkowych, które są kluczowe do weryfikowania działania poszczególnych części aplikacji w izolacji. Testy jednostkowe sprawdzają, czy konkretne fragmenty aplikacji działają zgodnie z oczekiwaniami.

python
Copy code
import pytest
from fastapi.testclient import TestClient
from protoapp.main import app
@pytest.fixture(scope="function")
def test_client():
    client = TestClient(app)
    yield client

Dzięki temu, fixture test_client będzie dostępna dla wszystkich testów i pozwoli na wygodne tworzenie testów w sposób bardziej kompaktowy. Teraz możemy napisać test w pliku test_main.py, który będzie wykorzystywał naszą fixture:

python
Copy code
def test_read_main_client(test_client):

    response = test_client.get("/home")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

Test ten jest prostszy i szybszy do napisania dzięki wykorzystaniu TestClient, który zapewnia synchronizację i automatycznie zarządza połączeniem z aplikacją. Aby uruchomić testy, wystarczy ponownie wykonać polecenie:

bash
Copy code
$ pytest

W terminalu powinna pojawić się informacja o przeprowadzonych testach i ich wynikach.

Testowanie punktów końcowych API

Aby zacząć, należy zainstalować odpowiednią bibliotekę SQLAlchemy:

bash
Copy code
$ pip install "sqlalchemy>=2.0.0"

Następnie, w module database.py utwórzmy konfigurację bazy danych, definiując klasę bazową oraz mapowanie tabeli:

python
Copy code
from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped

from sqlalchemy import create_engine, Integer, String
class Base(DeclarativeBase):
    pass
class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String, index=True)
    color: Mapped[str] = mapped_column(String)

Kolejnym krokiem jest konfiguracja silnika bazy danych oraz sesji:

python
Copy code
DATABASE_URL = "sqlite:///./production.db"

engine = create_engine(DATABASE_URL)
Base.metadata.create_all(bind=engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Dzięki temu będziemy mieli dostęp do sesji bazy danych, z której możemy korzystać w aplikacji. Następnie, w pliku main.py należy zaimplementować odpowiednie punkty końcowe, które będą dodawać i odczytywać dane z bazy.

Ważnym aspektem przy testowaniu integracyjnym jest zapewnienie, aby testy bazowały na jak najbardziej rzeczywistym środowisku, w którym system działa. Oznacza to, że w testach integracyjnych należy korzystać z pełnych komponentów systemu, takich jak baza danych, zewnętrzne API czy inne usługi, z których korzysta aplikacja.

Uwagi

Testowanie FastAPI to nie tylko prosta weryfikacja poprawności punktów końcowych, ale także kompleksowe sprawdzenie działania aplikacji w różnych scenariuszach. Ważne jest, aby testy były napisane w taki sposób, by testować aplikację na każdym poziomie – od testów jednostkowych, przez testy integracyjne, po pełne testy end-to-end. Pamiętaj, że odpowiednia konfiguracja środowiska testowego jest kluczowa do poprawnego przebiegu testów, a także do uzyskania stabilnych i powtarzalnych wyników.

Jak zintegrować FastAPI z Elasticsearch i Redis?

FastAPI to jeden z najnowszych i najszybszych frameworków do budowy aplikacji webowych w Pythonie, szczególnie popularny w przypadku mikroserwisów. Dzięki swojej wydajności i prostocie integracji z bazami danych NoSQL, takimi jak Elasticsearch i Redis, jest świetnym wyborem do budowy systemów wyszukiwania i cache'owania. W tej części skupimy się na integracji FastAPI z Elasticsearch oraz Redis, pokazując krok po kroku, jak skonfigurować zapytania do Elasticsearch oraz jak wprowadzić mechanizm cache'owania z Redis, aby zoptymalizować czas odpowiedzi aplikacji.

Integracja FastAPI z Elasticsearch

Pierwszym krokiem jest stworzenie indeksu w Elasticsearch oraz dodanie danych, które będą przechowywane i przetwarzane w systemie. Po utworzeniu indeksu w Elasticsearch, musimy zaprojektować zapytanie, które pozwoli nam na przetwarzanie danych, takich jak liczba odsłon dla piosenek w różnych krajach. Aby zrealizować tę funkcjonalność, definiujemy funkcję top_ten_songs_query, która zwróci zapytanie na podstawie kraju. Dzięki temu możemy sortować piosenki po liczbie odsłon w danym kraju i wybrać najlepsze dziesięć utworów.

Funkcja zapytania wygląda następująco:

python
Copy code
def top_ten_songs_query(country) -> dict:

    views_field = f"views_per_country.{country}"
    query = {
        "bool": {
            "must": {"match_all": {}},
            "filter": [
                {"exists": {"field": views_field}}
            ],
        }
    }
    sort = {views_field: {"order": "desc"}}
    source = [
        "title", views_field, "album.title", "artist"
    ]
    return {
        "index": "songs_index",
        "query": query,
        "size": 10,
        "sort": sort,
        "source": source,
    }

Po zbudowaniu zapytania tworzymy endpoint w FastAPI, który będzie korzystać z tego zapytania. Endpoint ten zwróci dziesięć najlepszych artystów w danym kraju na podstawie liczby odsłon ich piosenek.

python
Copy code
from fastapi import Depends, HTTPException
from app.db_connection import es_client
@router.get("/top/ten/artists/{country}")
async def top_ten_artist_by_country(
    country: str, es_client=Depends(get_elasticsearch_client)
):
    try:
        response = await es_client.search(*top_ten_songs_query(country))
    except BadRequestError as e:
        logger.error(e)
        raise HTTPException(
            status_code=400, detail="Invalid country"
        )
    return [
        {
            "artist": record.get("key"),
            "views": record.get("views", {}).get("value"),
        }

        for record in response["aggregations"]["top_ten_artists"]["buckets"]

    ]

Integracja FastAPI z Redis

Redis to narzędzie do przechowywania danych w pamięci, które często wykorzystywane jest jako system cache'owania w aplikacjach webowych. Dzięki Redis możemy przechowywać wyniki często wykonywanych zapytań, co znacząco przyspiesza czas odpowiedzi serwera. W tym przypadku, nasz cel to zastosowanie cache'owania dla endpointu, który zwraca dziesięciu najlepszych artystów w danym kraju. Zamiast za każdym razem wykonywać zapytanie do Elasticsearch, możemy przechowywać wynik w Redis przez określony czas, by przy kolejnych zapytaniach po prostu zwrócić dane z cache.

Zaczynamy od skonfigurowania połączenia z Redis w naszej aplikacji. W tym celu dodajemy klienta Redis do modułu połączeń (db_connection.py):

python
Copy code
from redis import asyncio as aioredis
redis_client = aioredis.from_url("redis://localhost")

Następnie definiujemy funkcję, która będzie pingować serwer Redis, aby sprawdzić, czy jest dostępny:

python
Copy code
async def ping_redis_server():

    try:
        await redis_client.ping()
        logger.info("Connected to Redis")
    except Exception as e:
        logger.error(f"Error connecting to Redis: {e}")
        raise e

W końcu integrujemy pingowanie Redis z aplikacją FastAPI, tak jak robiliśmy to wcześniej z MongoDB i Elasticsearch:

python
Copy code
@asynccontextmanager
async def lifespan(app: FastAPI):
    await ping_mongo_db_server()
    await ping_elasticsearch_server()
    await ping_redis_server()
    yield

Optymalizacja zapytań z Redis

Oto jak zmodyfikować nasz endpoint, by wykorzystać Redis:

python
Copy code
@router.get("/top/ten/artists/{country}")
async def top_ten_artist_by_country(

    country: str, es_client=Depends(get_elasticsearch_client), redis_client=Depends(get_redis_client)

):
    redis_key = f"top_ten_artists_{country}"
    cached_result = await redis_client.get(redis_key)
    if cached_result:
        return json.loads(cached_result)
    try:
        response = await es_client.search(*top_ten_songs_query(country))
    except BadRequestError as e:
        logger.error(e)
        raise HTTPException(status_code=400, detail="Invalid country")
    result = [
        {
            "artist": record.get("key"),
            "views": record.get("views", {}).get("value"),
        }
        for record in response["aggregations"]["top_ten_artists"]["buckets"]
    ]

    await redis_client.set(redis_key, json.dumps(result), ex=3600)  # Cache for 1 hour

    return result

Dzięki temu podejściu, jeśli dane dla danego kraju już znajdują się w cache, aplikacja szybko zwróci je z pamięci, co znacząco poprawi wydajność systemu.

Co warto dodać?

Ważnym aspektem, który warto podkreślić, jest zapewnienie odpowiedniej obsługi błędów i walidacji w przypadku integracji z Elasticsearch i Redis. Należy zadbać, by system był odporny na problemy z połączeniem z tymi usługami, oraz aby w razie błędu zapytania do Elasticsearch zwrócić odpowiedni komunikat użytkownikowi. Ponadto, warto ustawić odpowiednie limity czasowe na cache w Redis, by dane nie były przechowywane zbyt długo, a cache nie zawierał przestarzałych informacji.

Jak tworzyć middleware i zależności w FastAPI: Praktyczne podejście

FastAPI, jako framework do tworzenia aplikacji webowych, oferuje potężne mechanizmy do zarządzania zależnościami i middleware, co pozwala na łatwe rozszerzanie funkcjonalności aplikacji oraz lepsze zarządzanie wymaganiami związanymi z bezpieczeństwem, logowaniem czy obsługą błędów. W tej części przedstawimy, jak tworzyć niestandardowe middleware oraz jak wykorzystać zależności w FastAPI, aby poprawić strukturę kodu i elastyczność aplikacji.

W FastAPI, zależności są istotnym elementem, który pozwala na wstrzykiwanie funkcji i obiektów do punktów końcowych aplikacji. Dzięki temu możemy unikać powtarzania kodu i centralizować logikę aplikacji w jednym miejscu. FastAPI wspiera różne typy zależności, w tym te oparte na klasach, które umożliwiają bardziej zorganizowane i skalowalne podejście do zarządzania danymi.

Zależności oparte na klasach

Załóżmy, że chcemy utworzyć punkt końcowy, który wymaga trzech parametrów: zakresu czasowego, kategorii oraz kodu. Zamiast przekazywać te parametry oddzielnie, możemy stworzyć klasę, która będzie je grupować, a następnie wykorzystać ją jako zależność w naszym punkcie końcowym:

python
Copy code
from pydantic import BaseModel

from fastapi import Depends
class FilterParams(BaseModel):
    time_range: str
    category: str
    code: str
def get_filter_params(params: FilterParams = Depends()):
    return params

Teraz, w naszym punkcie końcowym, wystarczy użyć tej zależności, aby uzyskać wszystkie wymagane parametry w jednym obiekcie:

python
Copy code
from fastapi import FastAPI
app = FastAPI()
@app.get("/items")

def get_items(filters: FilterParams = Depends(get_filter_params)):

    return filters

Dzięki temu nasz kod staje się bardziej modularny i łatwiejszy do rozbudowy. Możemy także dodać dodatkowe walidacje w klasach, co pozwala na łatwiejsze zarządzanie danymi w aplikacji.

Middleware w FastAPI

Middleware to komponent API, który pozwala na przechwytywanie i modyfikowanie przychodzących żądań oraz wychodzących odpowiedzi. Jest to potężne narzędzie do implementowania funkcji takich jak uwierzytelnianie, logowanie czy obsługa błędów.

Aby stworzyć niestandardowe middleware w FastAPI, musimy stworzyć klasę, która dziedziczy po BaseHTTPMiddleware z biblioteki Starlette. Klasa ta musi implementować metodę dispatch, która przechwytuje żądanie, przetwarza je, a następnie przekazuje do następnego etapu w cyklu życia żądania.

Przykład prostego middleware, które loguje informacje o kliencie:

python
Copy code
import logging

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
logger = logging.getLogger("uvicorn.error")
class ClientInfoMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        host_client = request.client.host
        requested_path = request.url.path
        method = request.method

        logger.info(f"host client {host_client} requested {method} {requested_path} endpoint")

        response = await call_next(request)
        return response

Aby dodać middleware do aplikacji FastAPI, wystarczy użyć metody add_middleware:

python
Copy code
from fastapi import FastAPI
from app.middleware import ClientInfoMiddleware
app = FastAPI()
app.add_middleware(ClientInfoMiddleware)

Po uruchomieniu serwera i wykonaniu zapytania, zobaczymy w terminalu logi, które zawierają informacje o kliencie, ścieżce żądania i metodzie HTTP. Tego rodzaju middleware może być wykorzystywane do celów analitycznych, bezpieczeństwa czy audytu.

Zastosowanie middleware w praktyce

1. Uwierzytelnianie i autoryzacja: Przechwytywanie żądań, aby zweryfikować, czy użytkownik jest zalogowany lub ma odpowiednie uprawnienia.

2. Logowanie: Monitorowanie aktywności użytkowników i zapisywanie informacji o wykonywanych operacjach.

3. Obsługa błędów: Centralne zarządzanie błędami, co pozwala na spójne generowanie odpowiedzi w przypadku wystąpienia problemów.

Tworząc middleware, warto pamiętać o jego wydajności i modularności. Wraz ze wzrostem liczby zależności i operacji w aplikacji, kluczowe staje się odpowiednie zarządzanie kodem, aby nie wprowadzać nadmiernych opóźnień czy złożoności.

Międzynarodowość i lokalizacja w FastAPI

Międzynarodowość (i18n) i lokalizacja (l10n) są kluczowe, gdy aplikacja ma obsługiwać użytkowników z różnych regionów i kultur. i18n to proces projektowania oprogramowania w sposób umożliwiający jego łatwą adaptację do różnych języków i kultur, podczas gdy l10n to dostosowanie aplikacji do specyficznych rynków (np. dostosowanie walut, jednostek miar).

Przykładem może być obsługa dwóch języków: angielskiego i francuskiego. W pierwszym kroku definiujemy języki, które chcemy wspierać:

python
Copy code
SUPPORTED_LOCALES = ["en_US", "fr_FR"]

Następnie, na podstawie nagłówka Accept-Language, możemy określić, który język będzie używany przez aplikację:

python
Copy code
from fastapi import Request

def get_language(request: Request):
    accept_language = request.headers.get('Accept-Language', 'en_US')
    if 'fr' in accept_language:
        return 'fr_FR'
    return 'en_US'

Dzięki temu nasza aplikacja może automatycznie dostarczać treści w odpowiednim języku, co jest kluczowe dla globalnych aplikacji.