Jak przeprowadzić testy jednostkowe dla API i zaimplementować filtrację w FastAPI?

Aby przeprowadzić testy jednostkowe dla naszego API, musimy zacząć od zdefiniowania odpowiednich zasobów testowych w module conftest.py, który będzie używany do wszystkich testów w obrębie głównego folderu projektu. Na początku tworzmy plik testowy i definiujemy listę zadań testowych oraz nazwę pliku CSV, który posłuży do testów. Przykład:

python
Copy code
TEST_DATABASE_FILE = "test_tasks.csv"
TEST_TASKS_CSV = [
    { "id": "1", "title": "Test Task One", "description": "Test Description One", "status": "Incomplete" },
    { "id": "2", "title": "Test Task Two", "description": "Test Description Two", "status": "Ongoing" }
]
TEST_TASKS = [
    {**task_json, "id": int(task_json["id"])} for task_json in TEST_TASKS_CSV
]

Kolejnym krokiem jest stworzenie fixture, który będzie używany we wszystkich testach. Dzięki atrybutowi autouse=True, fixture będzie uruchamiany przed każdą funkcją testową. Takie rozwiązanie zapewnia, że przed każdym testem baza danych testowa będzie odpowiednio przygotowana:

python
Copy code
import csv
import os
from pathlib import Path
from unittest.mock import patch
import pytest

@pytest.fixture(autouse=True)
def create_test_database():
    database_file_location = str(Path(__file__).parent / TEST_DATABASE_FILE)
    with patch("operations.DATABASE_FILENAME", database_file_location) as csv_test:
        with open(database_file_location, mode="w", newline="") as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=["id", "title", "description", "status"])
            writer.writeheader()
            writer.writerows(TEST_TASKS_CSV)
        yield csv_test
        os.remove(database_file_location)

Teraz możemy przejść do definiowania testów jednostkowych. W module test_main.py tworzymy testowego klienta FastAPI:

python
Copy code
from main import app
from fastapi.testclient import TestClient

client = TestClient(app)

Testowanie punktów końcowych API

Zaczynamy od testowania endpointu GET /tasks, który ma za zadanie zwrócić wszystkie zadania z bazy danych:

python
Copy code
from conftest import TEST_TASKS

def test_endpoint_read_all_tasks():
    response = client.get("/tasks")
    assert response.status_code == 200
    assert response.json() == TEST_TASKS

Następnie przechodzimy do testowania endpointu GET /tasks/{task_id}, który pozwala na pobranie konkretnego zadania na podstawie jego ID:

python
Copy code
def test_endpoint_get_task():
    response = client.get("/task/1")
    assert response.status_code == 200
    assert response.json() == TEST_TASKS[0]
    response = client.get("/task/5")
    assert response.status_code == 404

Kolejny test dotyczy endpointu POST /task, który dodaje nowe zadanie do bazy danych:

python
Copy code
from operations import read_all_tasks

def test_endpoint_create_task():
    task = {"title": "To Define", "description": "will be done", "status": "Ready"}
    response = client.post("/task", json=task)
    assert response.status_code == 200
    assert response.json() == {**task, "id": 3}
    assert len(read_all_tasks()) == 3

W analogiczny sposób testujemy endpoint PUT /tasks/{task_id} do modyfikacji zadania:

python
Copy code
from operations import read_task

def test_endpoint_modify_task():
    updated_fields = {"status": "Finished"}
    response = client.put("/task/2", json=updated_fields)
    assert response.status_code == 200
    assert response.json() == {**TEST_TASKS[1], **updated_fields}
    response = client.put("/task/3", json=updated_fields)
    assert response.status_code == 404

Na koniec testujemy endpoint DELETE /tasks/{task_id}, który ma na celu usunięcie zadania:

python
Copy code
def test_endpoint_delete_task():
    response = client.delete("/task/2")
    assert response.status_code == 200
    expected_response = TEST_TASKS[1]
    del expected_response["id"]
    assert response.json() == expected_response
    assert read_task(2) is None

Uruchamianie testów

Po napisaniu testów, możemy je uruchomić za pomocą narzędzia pytest. Wystarczy przejść do katalogu głównego projektu i uruchomić polecenie:

bash
Copy code
$ pytest .

Jeśli wszystko zostało poprawnie zaimplementowane, pytest wyświetli komunikat o uzyskaniu 100% punktów w konsoli.

Dodanie funkcjonalności filtrowania i wyszukiwania

Filtracja danych w API jest niezbędną funkcjonalnością. W tym przypadku rozszerzymy nasz endpoint GET /tasks, aby umożliwić filtrowanie zadań na podstawie statusu i tytułu. Dzięki temu użytkownik będzie mógł precyzyjniej określić, które zadania go interesują.

Zmodyfikowany endpoint GET /tasks wygląda następująco:

python
Copy code
@app.get("/tasks", response_model=list[TaskWithID])
def get_tasks(status: Optional[str] = None, title: Optional[str] = None):
    tasks = read_all_tasks()
    if status:
        tasks = [task for task in tasks if task.status == status]
    if title:
        tasks = [task for task in tasks if task.title == title]
    return tasks

Oprócz tego wprowadzimy nowy endpoint GET /tasks/search, który umożliwia wyszukiwanie zadań na podstawie słowa kluczowego w tytule lub opisie:

python
Copy code
@app.get("/tasks/search", response_model=list[TaskWithID])
def search_tasks(keyword: str):
    tasks = read_all_tasks()
    filtered_tasks = [task for task in tasks if keyword.lower() in (task.title + task.description).lower()]
    return filtered_tasks

Te zmiany poprawiają funkcjonalność naszego API, umożliwiając użytkownikom łatwiejsze wyszukiwanie i filtrowanie zadań na podstawie różnych kryteriów.

Jak wykorzystać techniki debugowania w aplikacjach FastAPI?

Debugowanie kodu to kluczowa część procesu tworzenia oprogramowania, która pozwala na szybkie identyfikowanie i usuwanie błędów. Bez względu na to, czy pracujesz z prostymi skryptami, czy bardziej złożonymi aplikacjami internetowymi, znajomość odpowiednich narzędzi debugujących może znacząco ułatwić twoją pracę. W przypadku aplikacji opartych na FastAPI, często wykorzystywanymi narzędziami są breakpointy, PDB, VS Code i PyCharm.

Breakpointy to punkty zatrzymania w kodzie, które pozwalają na zatrzymanie wykonania programu i sprawdzenie stanu zmiennych oraz wywołań funkcji. Mogą być one powiązane z określonymi warunkami, które decydują o ich aktywacji lub pominięciu. Dzięki tej funkcji możesz dokładnie zobaczyć, co dzieje się w aplikacji w określonym momencie jej działania. W Pythonie, począwszy od wersji 3.7, możesz używać funkcji breakpoint(), aby wprowadzić przerwę w kodzie. Po osiągnięciu tej linii kodu, wykonanie programu przechodzi w tryb debugowania, a z terminala możesz uruchamiać różne polecenia do badania stanu aplikacji.

Pierwszym krokiem w debugowaniu aplikacji FastAPI jest utworzenie prostego skryptu, który uruchamia serwer. Skrypt taki pozwala na łatwiejsze zarządzanie procesem uruchamiania i testowania aplikacji, szczególnie w kontekście integracji z różnymi narzędziami debugującymi. Przykład skryptu run_server.py wygląda następująco:

python
Copy code
import uvicorn
from protoapp.main import app

if __name__ == "__main__":
    uvicorn.run(app)

Ten skrypt uruchamia aplikację FastAPI za pomocą serwera Uvicorn, co pozwala na łatwe uruchomienie aplikacji w trybie debugowania. Można go uruchomić jak zwykły skrypt Python: python run_server.py. Po uruchomieniu aplikacji, w przeglądarce można sprawdzić, czy dokumentacja API została poprawnie wygenerowana, odwiedzając adres localhost:8000/docs.

W przypadku debugowania za pomocą PDB, wystarczy dodać funkcję breakpoint() w miejscu, w którym chcemy zatrzymać program. Po dotarciu do tej linii, program przechodzi w tryb debugowania. Możemy wtedy korzystać z różnych poleceń, takich jak help (aby wyświetlić dostępne komendy), list (aby wyświetlić kod w określonym zakresie), czy args (aby zobaczyć argumenty funkcji). Jeżeli chcemy, aby PDB uruchamiał się automatycznie w przypadku, gdy aplikacja zakończy działanie w sposób nieoczekiwany, możemy uruchomić PDB w trybie post-mortem:

bash
Copy code
python -m pdb run_server.py

Dzięki temu debugger zostanie aktywowany, a my będziemy mogli przeanalizować przyczyny błędu.

Kolejną funkcjonalnością, którą warto wykorzystać, jest tryb przeładowania serwera Uvicorn. Aby włączyć automatyczne przeładowanie serwera, wystarczy dodać parametr reload=True w funkcji uruchamiającej serwer w skrypcie run_server.py. Umożliwia to szybkie testowanie zmian w aplikacji bez konieczności jej ponownego uruchamiania:

python
Copy code
import uvicorn

if __name__ == "__main__":
    uvicorn.run("protoapp.main:app", reload=True)

Dzięki temu wprowadzone zmiany w kodzie będą natychmiast widoczne w aplikacji, co znacznie przyspiesza proces debugowania. Jednakże, należy pamiętać, że w przypadku przeładowania serwera Uvicorn, tryb post-mortem debugowania nie jest obsługiwany.

W przypadku korzystania z popularnych edytorów kodu, takich jak VS Code czy PyCharm, debugowanie staje się jeszcze prostsze. VS Code oferuje integrację z debugpy, które jest wbudowanym debugerem wtyczki Python. Aby skonfigurować środowisko debugowania, należy w pliku .vscode/launch.json określić odpowiednie ustawienia, takie jak typ debugera oraz program do uruchomienia:

json
Copy code
{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Python Debugger FastAPI server",
            "type": "debugpy",
            "request": "launch",
            "program": "run_server.py",
            "console": "integratedTerminal"
        }
    ]
}

VS Code pozwala także na debugowanie testów jednostkowych, jeżeli w pliku konfiguracyjnym wskazany zostanie odpowiedni cel, np. testy. Dzięki tej funkcji, możesz debugować testy aplikacji i śledzić, w którym miejscu kodu występuje błąd.

Z kolei PyCharm oferuje zaawansowane opcje konfiguracji uruchamiania i debugowania, umożliwiając tworzenie różnych profili startowych. Może to być szczególnie pomocne w bardziej złożonych aplikacjach, które wymagają specyficznych ustawień środowiska lub zmiennych. Dzięki konfiguracjom w PyCharmie, można wygodnie przełączać się pomiędzy różnymi wersjami interpretera Pythona lub korzystać z dedykowanych zmiennych środowiskowych, dostosowanych do konkretnego przypadku testowego lub produkcyjnego.

Pamiętaj, że prawidłowe debugowanie jest nie tylko o używaniu odpowiednich narzędzi, ale także o zrozumieniu procesu pracy aplikacji. Warto dbać o dokumentację i strukturyzację kodu, aby łatwiej było znaleźć problematyczne miejsca. Używanie testów jednostkowych, odpowiedniego logowania oraz systemów monitorowania stanu aplikacji w czasie rzeczywistym może również pomóc w szybszym wykrywaniu i eliminowaniu błędów.

Jak zintegrować FastAPI z gRPC?

Aby zintegrować FastAPI z gRPC, należy przejść przez kilka kluczowych etapów, zaczynając od podstawowej konfiguracji serwera gRPC, a następnie stworzenia bramy (gateway), która umożliwi połączenie z serwisem gRPC poprzez FastAPI. Proces ten nie jest trudny, ale wymaga precyzyjnego postępowania oraz znajomości kilku technologii. Poniżej przedstawiam szczegóły, jak przejść przez ten proces.

Na początku musimy stworzyć plik .proto, który będzie zawierał definicje usług i wiadomości. W naszym przypadku używamy wersji proto3. Plik ten zawiera definicję metody serwera oraz komunikaty, które będą przesyłane między klientem a serwerem. Oto przykład pliku .proto:

proto
Copy code
syntax = "proto3";
service GrpcServer {
  rpc GetServerResponse(Message) returns (MessageResponse) {}
}

message Message {
  string message = 1;
}

message MessageResponse {
  string message = 1;
  bool received = 2;
}

Z tego pliku .proto możemy automatycznie wygenerować kod Pythona do integracji z serwisem i klientem gRPC, używając kompilatora proto. Aby to zrobić, uruchamiamy następującą komendę:

bash
Copy code
$ python -m grpc_tools.protoc --proto_path=. ./grpcserver.proto --python_out=. --grpc_python_out=.

W wyniku jej wykonania powstają dwa pliki: grpcserver_pb2.py i grpcserver_pb2_grpc.py. Pierwszy z nich zawiera definicje wiadomości (np. Message, MessageResponse), a drugi zawiera klasy potrzebne do budowy serwera oraz stuby, które będą wykorzystywane przez klienta.

Po wygenerowaniu plików, czas na napisanie serwera gRPC. Oto przykładowa implementacja:

python
Copy code
from grpcserver_pb2 import MessageResponse
from grpcserver_pb2_grpc import GrpcServerServicer
import logging

class Service(GrpcServerServicer):
    async def GetServerResponse(self, request, context):
        message = request.message
        logging.info(f"Received message: {message}")
        result = {
            "message": f"Hello I am up and running, received: {message}",
            "received": True
        }
        return MessageResponse(**result)

Po zdefiniowaniu serwera, należy go uruchomić na porcie 50051:

python
Copy code
import grpc
from grpcserver_pb2_grpc import add_GrpcServerServicer_to_server
import asyncio
import logging

async def serve():
    server = grpc.aio.server()
    add_GrpcServerServicer_to_server(Service(), server)
    server.add_insecure_port("[::]:50051")
    logging.info("Starting server on port 50051")
    await server.start()
    await server.wait_for_termination()

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(serve())

Po uruchomieniu serwera, zobaczymy na konsoli komunikat informujący, że serwer działa na porcie 50051. Następnie możemy przejść do stworzenia bramy REST przy użyciu FastAPI, która będzie pośredniczyć w przesyłaniu danych między klientem a serwerem gRPC.

Aby utworzyć aplikację FastAPI, tworzymy nowy folder app, a w nim plik main.py z następującym kodem:

python
Copy code
from fastapi import FastAPI
import grpc
from grpcserver_pb2 import Message
from grpcserver_pb2_grpc import GrpcServerStub
from pydantic import BaseModel

app = FastAPI()

# Definicja schematu odpowiedzi
class GRPCResponse(BaseModel):
    message: str
    received: bool

# Inicjalizacja kanału gRPC
grpc_channel = grpc.aio.insecure_channel("localhost:50051")

@app.get("/grpc")
async def call_grpc(message: str) -> GRPCResponse:
    async with grpc_channel as channel:
        grpc_stub = GrpcServerStub(channel)
        response = await grpc_stub.GetServerResponse(Message(message=message))
        return response

W tym przykładzie aplikacja FastAPI nasłuchuje na endpoint /grpc, który przyjmuje parametr message. Wiadomość ta jest przesyłana do serwera gRPC, który odpowiada komunikatem, który następnie jest zwracany do klienta FastAPI.

Po uruchomieniu aplikacji za pomocą komendy:

bash
Copy code
$ uvicorn app.main:app

Możemy przejść do interaktywnej dokumentacji pod adresem http://localhost:8000/docs, gdzie znajdziemy nasz endpoint /grpc. Wysyłając zapytanie na ten endpoint, otrzymamy odpowiedź od serwera gRPC.

Stworzyliśmy w ten sposób podstawową bramę REST-gRPC z wykorzystaniem FastAPI. Jednak to tylko początek. Warto pamiętać, że gRPC oferuje różne typy RPC, takie jak streaming, które pozwalają na przesyłanie danych w czasie rzeczywistym między klientem a serwerem. Również dostępne są dwukierunkowe połączenia, które umożliwiają wymianę wiadomości pomiędzy serwerem a klientem w obu kierunkach.

W przypadku chęci dalszego zgłębiania tematu, warto zapoznać się z oficjalną dokumentacją gRPC i protokołów Buffer:

• gRPC Python Tutorial

• Protocol Buffer Python Generated Code

Te zasoby dostarczą dodatkowych informacji na temat implementacji gRPC w Pythonie oraz jego integracji z różnymi bibliotekami, takimi jak FastAPI.

Jak stworzyć aplikację AI do diagnozowania chorób przy użyciu FastAPI i integracji z modelem ML

W dzisiejszych czasach integracja sztucznej inteligencji z aplikacjami webowymi stała się niemal niezbędna w wielu dziedzinach, w tym w medycynie. Dzięki bibliotece FastAPI, która jest szybka i elastyczna, możemy stworzyć aplikację, która pozwoli na diagnozowanie chorób na podstawie objawów podanych przez użytkownika. Przyjrzyjmy się, jak można to zrobić.

Aby rozpocząć, musimy stworzyć endpoint, który przyjmie objawy jako parametry i na ich podstawie zwróci diagnozę. Ponieważ mamy do czynienia z dużą liczbą objawów, z których użytkownik może wybierać, ważne jest, aby parametry były dynamicznie tworzone i ograniczone do kilku najczęstszych objawów. Możemy to osiągnąć przy pomocy Pydantic, używając funkcji create_model do generowania modelu, który zaakceptuje objawy jako parametry. W tym przypadku, do testów, ograniczamy się do pierwszych dziesięciu objawów.

W kodzie Python wygląda to tak:

python
Copy code
from pydantic import create_model
from app.utils import symptoms_list

query_parameters = {
    symp: (bool, False)
    for symp in symptoms_list[:10]
}

Symptoms = create_model("Symptoms", **query_parameters)

Dzięki temu mamy gotowy model, który możemy wykorzystać do stworzenia endpointu, który przyjmuje objawy jako dane wejściowe. Endpoint, który przyjmuje parametry objawów, a następnie wykonuje diagnozę przy pomocy wbudowanego modelu ML, będzie wyglądał tak:

python
Copy code
@app.get("/diagnosis")
async def get_diagnosis(
    symptoms: Annotated[Symptoms, Depends()],
):
    array = [
        int(value)
        for _, value in symptoms.model_dump().items()
    ]
    array.extend(
        [0] * (len(symptoms_list) - len(array))
    )
    diseases = ml_model["doctor"].predict([array])
    return {
        "diseases": [disease for disease in diseases]
    }

Testowanie aplikacji jest równie łatwe. Po uruchomieniu serwera przy użyciu uvicorn, możemy przejść do interaktywnej dokumentacji w przeglądarce (http://localhost:8000/docs), gdzie zobaczymy dostępny endpoint /diagnosis. Będziemy mogli wybrać objawy, które nas interesują, a aplikacja zwróci wynik diagnozy oparty na danych z modelu ML.

Taki system może być wykorzystywany w różnych kontekstach. Możemy zastosować tę samą metodę integracji modeli do różnych endpointów w aplikacji. Możliwe jest również połączenie kilku różnych modeli w jednej aplikacji, co daje dużą elastyczność w tworzeniu bardziej zaawansowanych systemów AI.

Integracja FastAPI z modelami ML nie kończy się jednak na jednej aplikacji. Można również połączyć FastAPI z innymi platformami, takimi jak Cohere, oferującymi potężne modele językowe. Dzięki nim aplikacje mogą wykonywać zadania związane z przetwarzaniem języka naturalnego, takie jak generowanie tekstów, odpowiadanie na pytania czy rozumienie kontekstu rozmów.

W kontekście pracy z Cohere, przed rozpoczęciem warto założyć konto na platformie i uzyskać klucz API. Po uzyskaniu dostępu do API, można stworzyć aplikację, która np. pełni funkcję chatbota doradzającego w kwestiach kulinarnych. W tym przypadku model AI działa na podstawie zadanych pytań użytkownika, udzielając odpowiedzi i sugerując przepisy kulinarne, np. na dania kuchni włoskiej.

Oto jak krok po kroku stworzyć chatbota z wykorzystaniem Cohere:

1. Stwórz plik .env w katalogu głównym projektu, w którym zapiszesz swój klucz API:

   text
   Copy code
   COHERE_API_KEY="twój-klucz-api"
   

2. Zainstaluj wymagane biblioteki, takie jak FastAPI, Uvicorn, Cohere, oraz Python-dotenv:

   bash
   Copy code
   pip install fastapi uvicorn cohere python-dotenv
   

3. Stwórz funkcję, która będzie generować odpowiedzi chatbota, korzystając z modelu Cohere:

python
Copy code
from cohere import AsyncClient
from cohere.core.api_error import ApiError
from fastapi import HTTPException

SYSTEM_MESSAGE = (
    "You are a skilled Italian top chef "
    "expert in Italian cuisine tradition "
    "that suggest the best recipes unveiling "
    "tricks and tips from Grandma's Kitchen "
    "shortly and concisely."
)

client = AsyncClient()

async def generate_chat_completion(user_query=" ", messages=[]):
    try:
        response = await client.chat(
            message=user_query,
            model="command-r-plus",
            preamble=SYSTEM_MESSAGE,
            chat_history=messages,
        )
        messages.append({
            "role": "USER",
            "message": user_query,
        })
        messages.append({
            "role": "CHATBOT",
            "message": response.text,
        })
        return response.text
    except ApiError as e:
        raise HTTPException(status_code=e.status_code, detail=e.body)

4. Stwórz endpoint, który będzie przyjmować zapytania od użytkowników i odpowiadać na nie zgodnie z powyższą logiką.

Integracja FastAPI z modelami AI, zarówno w kontekście diagnozowania chorób, jak i rozumienia i generowania języka naturalnego, otwiera ogromne możliwości w tworzeniu nowoczesnych aplikacji. To narzędzie pozwala na szybkie prototypowanie oraz łatwą integrację z różnymi modelami sztucznej inteligencji, co czyni je bardzo atrakcyjnym wyborem dla wielu nowoczesnych systemów.