Jak stworzyć prosty benchmark WebSocket i optymalizować jego wydajność w FastAPI?

Aby stworzyć prosty benchmark dla WebSocket w aplikacjach FastAPI, należy zacząć od stworzenia skryptu, który zrealizuje kluczowe zadania: uruchomi serwer, utworzy odpowiednią liczbę klientów WebSocket i wymieni z nimi wiadomości. Zaczniemy od przygotowania skryptu, który pozwoli przeprowadzić testy wydajnościowe.

Pierwszym krokiem jest zdefiniowanie funkcji uruchamiającej serwer FastAPI. W tym celu użyjemy uvicorn – narzędzia do uruchamiania aplikacji ASGI. W skrypcie, zamiast standardowego polecenia uruchamiania serwera z linii poleceń, będziemy posługiwać się funkcją run_server, która uruchomi serwer wewnętrznie. Przykładowa implementacja tej funkcji wygląda następująco:

python
import uvicorn
from app.main import app

def run_server():
    uvicorn.run(app)

Ta funkcja uruchamia serwer FastAPI w sposób podobny do uruchamiania go z terminala, ale w ramach samego skryptu.

Kolejnym krokiem jest stworzenie funkcji, która pozwoli na połączenie odpowiedniej liczby klientów WebSocket z naszym punktem końcowym i wymianę wiadomości. W tym celu posłużymy się biblioteką websockets i asyncio, które umożliwiają realizowanie asynchronicznych połączeń. Funkcja connect_client utworzy klientów, którzy połączą się z serwerem WebSocket i wyślą kilka wiadomości, symulując tym samym interakcje w aplikacji czatu. Warto dodać pewien czas opóźnienia pomiędzy wiadomościami, aby uzyskać bardziej realistyczne zachowanie:

python
import asyncio
from websockets import connect

async def connect_client(n: int, n_messages: int = 3):
    async with connect(f"ws://localhost:8000/chatroom/user{n}") as client:
        for _ in range(n_messages):
            await client.send(f"Hello World from user{n}")
            await asyncio.sleep(n * 0.1)  # Opóźnienie dla realistycznego zachowania
        await asyncio.sleep(2)

Dzięki użyciu asyncio jesteśmy w stanie uruchomić wiele połączeń równocześnie, co pozwala na symulowanie wysokiego obciążenia serwera przez jednoczesne zapytania do punktu końcowego. Warto dodać, że w rzeczywistych aplikacjach opóźnienie w wysyłaniu wiadomości jest również ważne, ponieważ użytkownicy nie wysyłają ich natychmiast po sobie, lecz z pewnym czasem opóźnienia.

Następnie musimy połączyć wszystko w jedną funkcję main, która uruchomi serwer w osobnym procesie i zainicjuje połączenia od klientów. Wykorzystujemy tutaj procesy z modułu multiprocessing, aby uruchomić serwer w tle i równocześnie nawiązywać połączenia z WebSocket:

python
import multiprocessing

async def main(n_clients: int = 10):
    p = multiprocessing.Process(target=run_server)
    p.start()
    await asyncio.sleep(1)  # Dajemy czas na uruchomienie serwera
    connections = [connect_client(n) for n in range(n_clients)]
    await asyncio.gather(*connections)
    await asyncio.sleep(1)
    p.terminate()  # Zakończenie procesu serwera

Aby uruchomić cały skrypt, wystarczy wywołać go jak zwykły skrypt Pythona z terminala:

bash
$ python benchmark_websocket.py

Warto zauważyć, że domyślnie parametr n_clients ustawia liczbę klientów na 10. Możemy zwiększyć tę liczbę, ale z każdym kolejnym klientem wydajność naszego komputera może zostać wystawiona na próbę. Gdy liczba klientów przekroczy granicę obsługiwaną przez maszynę, pojawią się błędy związane z połączeniami WebSocket.

Optymalizacja wydajności WebSocket

Benchmarking to tylko pierwszy krok w kierunku optymalizacji wydajności aplikacji WebSocket. Po przeprowadzeniu prostych testów obciążeniowych warto podjąć działania, które poprawią działanie aplikacji i zapobiegną błędom. Oto kilka kluczowych działań, które warto podjąć:

Testowanie jednostkowe WebSocket z TestClient: FastAPI udostępnia narzędzie TestClient, które pozwala testować połączenia WebSocket. Dzięki niemu możemy sprawdzić, czy endpoint WebSocket działa zgodnie z oczekiwaniami, a także upewnić się, że nie zachodzą w nim nieoczekiwane zmiany w trakcie rozwoju aplikacji.
Obsługa błędów: Warto zaimplementować mechanizmy obsługi błędów, aby w przypadku problemów z połączeniami WebSocket aplikacja nie ulegała awarii. Użycie bloków try/except pozwoli na łapanie wyjątków związanych z błędami połączenia, a także umożliwi automatyczne przechwytywanie i obsługę błędów związanych z rozłączeniami, np. przy użyciu konstrukcji async for zamiast nieskończonej pętli while True.
Zarządzanie połączeniami za pomocą puli połączeń: W aplikacjach obsługujących wielu klientów, takich jak czaty, użycie menedżerów pul połączeń znacząco poprawia wydajność oraz ułatwia utrzymanie kodu. Pozwala to na efektywne zarządzanie otwartymi połączeniami i optymalizowanie zużycia zasobów.

Po przeprowadzeniu testów wydajnościowych i optymalizacji, warto wdrożyć dodatkowe mechanizmy ochrony i autoryzacji. Przykładem może być zabezpieczenie połączeń WebSocket przy użyciu OAuth2, co zapewni, że tylko autoryzowani użytkownicy będą mogli korzystać z endpointów WebSocket. Warto dodać, że FastAPI nie wspiera natywnie klasy OAuth2PasswordBearer dla WebSocket, więc trzeba stworzyć dedykowaną klasę, która pozwoli na weryfikację tokenów w nagłówkach WebSocket, co przedstawiono w poniższym przykładzie:

python
from fastapi import WebSocket, WebSocketException, status
from fastapi.security import OAuth2PasswordBearer

class OAuth2WebSocketPasswordBearer(OAuth2PasswordBearer):
    async def __call__(self, websocket: WebSocket) -> str:
        authorization: str = websocket.headers.get("authorization")
        if not authorization:
            raise WebSocketException(code=status.HTTP_401_UNAUTHORIZED, reason="Not authenticated")
        scheme, param = authorization.split()
        if scheme.lower() != "bearer":
            raise WebSocketException(code=status.HTTP_403_FORBIDDEN, reason="Invalid authentication credentials")
        return param

Taka klasa umożliwia sprawdzenie tokena OAuth2 w nagłówkach WebSocket i zabezpieczenie połączenia przed nieautoryzowanym dostępem. Po implementacji tego mechanizmu użytkownicy muszą przejść proces autoryzacji przed uzyskaniem dostępu do chronionych zasobów WebSocket.

Jak rozwijać middleware w FastAPI do modyfikacji żądań i odpowiedzi?

Middleware w aplikacjach webowych stanowi potężne narzędzie umożliwiające przetwarzanie żądań i odpowiedzi. Tworzenie własnych middleware pozwala na modyfikację tych elementów przed ich dalszym przetwarzaniem, co daje deweloperom ogromną elastyczność i kontrolę nad aplikacją. W tym rozdziale skupimy się na rozwoju middleware w ASGI (Asynchronous Server Gateway Interface) w kontekście FastAPI, które pozwala na dynamiczne przetwarzanie treści żądań oraz odpowiedzi. Dzięki temu możemy dodawać lub zmieniać nagłówki odpowiedzi, modyfikować treść ciał żądań, a także wprowadzać dodatkowe warstwy bezpieczeństwa.

Zanim przejdziemy do szczegółów implementacji, warto zaznaczyć, że middleware w FastAPI bazuje na bibliotece Starlette, co otwiera dodatkowe możliwości w zakresie tworzenia bardziej zaawansowanych funkcji. Dla osób, które po raz pierwszy spotykają się z tym zagadnieniem, rekomendujemy zapoznanie się z dokumentacją ASGI, dostępną na stronach Starlette oraz ASGI. Wspomniana dokumentacja zawiera niezbędne informacje dotyczące tworzenia czystego middleware w oparciu o ASGI, co może być przydatne w bardziej złożonych projektach.

Tworzenie middleware do modyfikacji ciała żądania

Podstawowym zadaniem middleware może być modyfikacja treści ciała żądania, na przykład przez jego haszowanie przed wysłaniem do endpointu. Zaczniemy od utworzenia odpowiedniego modułu w naszej aplikacji. Na przykład w pliku request_middleware.py zdefiniujemy klasę HashBodyContentMiddleware, która będzie odpowiadać za haszowanie treści ciała żądania, jeżeli jest ono dostępne.

Aby rozpocząć, w module middleware zaimportujemy wymagane elementy z biblioteki Starlette oraz funkcję sha1 z hashlib, która będzie wykorzystywana do generowania skrótu treści. Przykładowy kod wyglądałby następująco:

python
from starlette.types import ASGIApp, Scope, Receive, Send, Message
from hashlib import sha1

class HashBodyContentMiddleware:
    def __init__(self, app: ASGIApp, allowed_paths: list[str]):
        self.app = app
        self.allowed_paths = allowed_paths

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        if scope["type"] != "http" or scope["path"] not in self.allowed_paths:
            await self.app(scope, receive, send)
            return

        async def receive_with_new_body() -> Message:
            message = await receive()
            assert message["type"] == "http.request"
            body = message.get("body", b"")
            message["body"] = f'"{sha1(body).hexdigest()}"'.encode()
            return message

        await self.app(scope, receive_with_new_body, send)

W powyższym kodzie sprawdzamy, czy żądanie jest HTTP oraz czy ścieżka żądania znajduje się w dozwolonych ścieżkach, zanim rozpoczniemy modyfikację ciała. Funkcja receive_with_new_body umożliwia nam przechwycenie oryginalnego ciała żądania, jego haszowanie oraz dalsze przekazanie go do aplikacji. W tym przykładzie haszujemy treść za pomocą funkcji sha1, ale może to być dowolna funkcja kryptograficzna lub algorytm, który zdecydujemy się zastosować w zależności od potrzeb aplikacji.

Po zdefiniowaniu middleware w naszym projekcie, możemy je łatwo dodać do instancji aplikacji FastAPI, używając metody add_middleware. Oto jak można to zrobić:

python
app.add_middleware(
    HashBodyContentMiddleware,
    allowed_paths=["/send"],
)

Dzięki temu middleware zostanie uruchomione tylko dla ścieżek, które są określone w parametrze allowed_paths, co pozwala na elastyczne zarządzanie tym, które żądania będą podlegały modyfikacjom.

Testowanie i implementacja w aplikacji

Po zaimplementowaniu i dodaniu middleware do aplikacji FastAPI możemy przeprowadzić testy. W tym celu uruchamiamy serwer za pomocą komendy uvicorn:

bash
$ uvicorn main:app

Następnie możemy odwiedzić dokumentację interaktywną pod adresem http://localhost:8000/docs, aby przetestować endpoint POST /send, który będzie modyfikowany przez nasze middleware. Warto spróbować wysłać na przykład treść "hello middleware", a w odpowiedzi powinna pojawić się jej skrócona wersja w postaci hasha.

Rozwój middleware do modyfikacji odpowiedzi

Po opanowaniu podstaw modyfikacji żądań, możemy przejść do modyfikacji odpowiedzi. Middleware, które będzie odpowiadać za modyfikację nagłówków odpowiedzi, może dodać dodatkowe informacje, takie jak nagłówki związane z bezpieczeństwem, personalizację treści odpowiedzi, a także inne elementy, które chcielibyśmy dodać do każdej odpowiedzi w naszej aplikacji. Tego typu middleware może wyglądać podobnie do wcześniejszego, jednak jego celem będzie manipulacja odpowiedziami po ich przetworzeniu przez endpointy.

Kluczowe informacje

Warto zwrócić uwagę, że middleware w ASGI działa w sposób asynchroniczny, co oznacza, że operacje takie jak odbieranie i modyfikowanie ciał żądań mogą wymagać odpowiedniego zarządzania asynchronicznością. Każda operacja w ramach middleware powinna być przeprowadzona w sposób nieblokujący, aby nie wpłynąć na ogólną wydajność aplikacji.

Ponadto, podczas tworzenia middleware należy pamiętać o zabezpieczeniach. Choć nasze middleware do haszowania ciał żądań jest prostym przykładem, w bardziej zaawansowanych scenariuszach może być konieczne zastosowanie dodatkowych warstw ochrony przed atakami typu XSS (cross-site scripting), SQL injection oraz innymi formami ataków, które mogą próbować wykorzystać luki w przetwarzaniu żądań.

W miarę jak aplikacja rośnie, warto również zastanowić się nad testowaniem i monitorowaniem działania middleware. Dzięki odpowiednim narzędziom można łatwo sprawdzić, jak działa middleware w różnych warunkach i w jaki sposób wpływa na wydajność oraz bezpieczeństwo aplikacji.

Jak zaimplementować webhooki w aplikacji FastAPI?

Implementacja webhooków w aplikacjach opartych na FastAPI jest jednym z kluczowych elementów umożliwiających komunikację z systemami zewnętrznymi. Webhooki to mechanizm umożliwiający otrzymywanie powiadomień o zdarzeniach w czasie rzeczywistym, które są wysyłane przez zewnętrzne aplikacje do naszego serwera. Dzięki nim można zbudować interaktywne aplikacje reagujące na wydarzenia, co sprawia, że proces integracji z innymi systemami staje się znacznie łatwiejszy i bardziej efektywny.

Aby zaimplementować webhooki w aplikacji FastAPI, zaczynamy od stworzenia middleware, które będzie obsługiwać przychodzące powiadomienia. Kluczowym krokiem jest zaimportowanie odpowiedniego middleware do aplikacji. Można to zrobić w pliku main.py, dodając kod:

python
from middleware.webhook import WebhookSenderMiddleWare

app.add_middleware(WebhookSenderMiddleWare)

Dzięki temu aplikacja będzie mogła obsługiwać żądania POST skierowane do endpointu webhooka. Następnie, aby poprawnie udokumentować działanie webhooka, warto dodać odpowiednią funkcję do dokumentacji OpenAPI, co ułatwi użytkownikom API zrozumienie sposobu działania webhooków w naszej aplikacji.

python
@app.webhooks.post("/fastapi-webhook")
def fastapi_webhook(event: Event):
    """Webhook przyjmujący zdarzenia z systemów zewnętrznych"""
    # Przetwarzanie zdarzenia

W tym przypadku, zdarzenie (obiekt Event) jest przekazywane jako parametr funkcji. Może ono zawierać informacje o hoście, ścieżce, czasie oraz ciele żądania. Warto zdefiniować strukturę tego zdarzenia, aby dokładnie określić, jakie dane są wymagane:

python
class Event(BaseModel):
    host: str
    path: str
    time: str
    body: str | None = None

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "host": "127.0.0.1",
                    "path": "/send",
                    "time": "2024-05-22T14:24:28.847663",
                    "body": '"body content"'
                }
            ]
        }
    }

W przypadku uruchomienia serwera na lokalnej maszynie za pomocą komendy uvicorn main:app oraz otwarcia dokumentacji pod adresem http://localhost:8000/docs, użytkownicy będą mogli zapoznać się z dokładnym opisem działania endpointu webhooka. W ten sposób możliwe jest zapewnienie, że wszystkie zainteresowane strony będą mogły skutecznie korzystać z funkcjonalności webhooka w aplikacji.

Warto również przetestować webhooki, konfigurując prosty serwer działający na porcie 8080. Może to być zarówno serwer stworzony samodzielnie, jak i gotowy skrypt http_server.py dostępny w repozytorium GitHub. Po uruchomieniu serwera należy dodać jego adres do aplikacji FastAPI przy użyciu endpointu POST /register-webhook-url, wskazując odpowiedni port, na którym nasz serwer nasłuchuje. Po każdym wywołaniu endpointu w aplikacji FastAPI, zdarzenie webhooka zostanie przekazane do serwera, a odpowiednie informacje zostaną wyświetlone w terminalu.

Podstawowa implementacja webhooków w FastAPI jest relatywnie prosta, ale można ją znacznie wzbogacić, wdrażając dodatkowe funkcje i zabezpieczenia. Do najważniejszych z nich należą:

Autentykacja: W celu zapewnienia bezpieczeństwa komunikacji pomiędzy serwerem a webhookiem, warto zaimplementować odpowiednią metodę autentykacji, np. poprzez klucze API lub OAuth.
Mechanizm ponawiania prób: Z racji, że webhooki wykorzystują HTTP, mogą wystąpić problemy z dostarczaniem powiadomień (np. z powodu problemów z siecią). Implementacja mechanizmu ponawiania prób zapewnia, że zdarzenia zostaną dostarczone nawet w przypadku tymczasowych problemów.
Przechowywanie zdarzeń: Aby umożliwić późniejsze odtworzenie lub audyt, warto przechowywać zdarzenia webhooków w bazie danych. Można wykorzystać do tego narzędzie SQLAlchemy, które ułatwia integrację z relacyjnymi bazami danych.
WebSocket dla powiadomień w czasie rzeczywistym: Jeżeli aplikacja wymaga natychmiastowych powiadomień, zamiast klasycznych webhooków HTTP, można wdrożyć WebSocket, co umożliwi natychmiastowe przesyłanie danych do klientów.
Ograniczenia liczby żądań (rate limiting): Aby zapobiec nadużyciom, warto wdrożyć mechanizm limitowania liczby żądań, co pomoże uniknąć przeciążenia serwera.

Webhooki są niezwykle potężnym narzędziem, pozwalającym na łatwą integrację z systemami zewnętrznymi. Ich efektywne wykorzystanie sprawia, że aplikacje stają się bardziej interaktywne i reagują na zdarzenia w czasie rzeczywistym. Warto zapoznać się z dodatkowymi zasobami, które oferują bardziej zaawansowane techniki pracy z webhookami, a także zgłębić dokumentację FastAPI, aby w pełni wykorzystać potencjał tej technologii.

Hogyan szereljük össze és optimalizáljuk a Raygun Pen elektronikai egységét?
Miért Muriseay más, mint bármi, amit valaha láttál?
Hogyan vásároljunk okosan: Tippek a tudatos ruhavásárláshoz
Milyen szerepe van a mesékben és folklórban a hősöknek és a csodás lényeknek?