Jak wykonać operacje CRUD w SQLAlchemy z FastAPI?

Po skonfigurowaniu bazy danych SQL za pomocą FastAPI, kolejnym kluczowym krokiem jest tworzenie modeli bazy danych. Ten proces jest fundamentem sposobu, w jaki aplikacja wchodzi w interakcję z bazą danych. Modele bazy danych w SQLAlchemy to w zasadzie klasy Pythona, które reprezentują tabele w bazie danych SQL. Dają one wysokopoziomowy, obiektowy interfejs do manipulowania rekordami bazy danych tak, jakby były zwykłymi obiektami Pythona. W tej części omówimy, jak skonfigurować punkty końcowe CRUD, które pozwolą na interakcję z bazą danych.

Wszystkie operacje CRUD są niezbędne w każdym systemie opartym na bazach danych, ponieważ umożliwiają one tworzenie, odczyt, aktualizację i usuwanie danych. Każda z tych operacji wymaga odpowiednio przygotowanego punktu końcowego w aplikacji, który będzie obsługiwał komunikację z bazą danych.

Tworzenie nowego użytkownika

Aby dodać nowego użytkownika, używamy zapytania POST. W pliku main.py należy zdefiniować punkt końcowy, który odbierze dane użytkownika, utworzy nową instancję klasy User w żądaniu oraz doda ją do bazy danych. W przykładzie poniżej, przy pomocy UserBody, dane użytkownika są przesyłane do aplikacji:

python
Copy code
class UserBody(BaseModel):
    name: str
    email: str

@app.post("/user")
def add_new_user(user: UserBody, db: Session = Depends(get_db)):
    new_user = User(name=user.name, email=user.email)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

W kilka linii kodu utworzono punkt końcowy, który dodaje nowego użytkownika do bazy danych.

Odczyt użytkownika

Aby pobrać konkretnego użytkownika, używamy zapytania GET. Punkt końcowy sprawdza, czy użytkownik o podanym identyfikatorze istnieje w bazie danych, a jeśli nie, zwraca odpowiedź z kodem błędu 404.

python
Copy code
@app.get("/user")
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

W przypadku braku użytkownika, aplikacja zwraca odpowiedź o błędzie 404.

Aktualizacja użytkownika

Aktualizacja rekordu w bazie danych za pomocą API może być realizowana różnymi metodami, takimi jak PUT, PATCH czy POST. Choć każda z tych metod ma swoje teoretyczne różnice, praktycznie wybór metody zależy od preferencji programisty. W tym przykładzie używamy zapytania POST, rozbudowując punkt końcowy /user o parametr user_id.

python
Copy code
@app.post("/user/{user_id}")
def update_user(user_id: int, user: UserBody, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    db_user.name = user.name
    db_user.email = user.email
    db.commit()
    db.refresh(db_user)
    return db_user

Dzięki temu punktowi końcowemu możemy zaktualizować dane użytkownika w bazie danych, bez konieczności tworzenia oddzielnych punktów końcowych dla różnych metod HTTP.

Usuwanie użytkownika

Aby usunąć użytkownika, używamy zapytania DELETE. W tym przypadku, podobnie jak w przypadku odczytu, sprawdzamy, czy użytkownik istnieje, a jeśli tak, usuwamy go z bazy danych:

python
Copy code
@app.delete("/user")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    db.delete(db_user)
    db.commit()
    return {"detail": "User deleted"}

Operacja ta usuwa użytkownika z bazy danych i zwraca odpowiedź informującą o jego usunięciu.

Testowanie CRUD

Po wdrożeniu wszystkich operacji CRUD możesz uruchomić serwer aplikacji za pomocą komendy:

bash
Copy code
$ uvicorn main:app

Po uruchomieniu serwera, odwiedź interaktywną dokumentację pod adresem http://localhost:8000/docs, gdzie będziesz mógł testować punkty końcowe CRUD, tworzyć, odczytywać, aktualizować i usuwać użytkowników.

Opanowanie operacji CRUD w FastAPI stanowi istotny krok w budowaniu dynamicznych aplikacji webowych opartych na danych. Dzięki znajomości integracji punktów końcowych FastAPI z modelami SQLAlchemy, masz solidne podstawy do tworzenia skomplikowanych i wydajnych aplikacji.

Rozszerzenie CRUD w aplikacji

Warto również pamiętać o dodatkowych aspektach, które mogą wpływać na wydajność aplikacji w kontekście operacji CRUD. Po pierwsze, warto zwrócić uwagę na odpowiednie indeksowanie tabel w bazie danych, co znacznie przyspiesza operacje odczytu, szczególnie przy dużej ilości danych. Kolejnym ważnym aspektem jest walidacja danych, którą można rozbudować, by zapewnić integralność danych w aplikacji. Należy także pamiętać o obsłudze transakcji, zwłaszcza przy operacjach aktualizacji i usuwania, by zapewnić, że dane pozostaną w spójnym stanie nawet w przypadku błędów lub awarii systemu. Na koniec, aby zapewnić bezpieczeństwo aplikacji, warto zaimplementować odpowiednie mechanizmy autoryzacji i uwierzytelniania, by operacje CRUD były dostępne tylko dla uprawnionych użytkowników.

Jak skutecznie zarządzać połączeniami WebSocket w aplikacji FastAPI?

WebSocket to potężna technologia umożliwiająca dwukierunkową, ciągłą wymianę danych między klientem a serwerem w czasie rzeczywistym. W kontekście aplikacji FastAPI, WebSockety stanowią kluczowy element w budowie aplikacji opartych na komunikacji w czasie rzeczywistym, takich jak czaty, powiadomienia czy gry online. W tej części przedstawimy podstawy pracy z połączeniami WebSocket, a także jak zarządzać nimi w FastAPI.

Podstawowy proces nawiązywania połączenia WebSocket w FastAPI jest prosty. Serwer rozpoczyna połączenie, czekając na akceptację od klienta, a następnie nawiązuje pełną dwukierunkową komunikację. Po utworzeniu połączenia, serwer może wysłać powitalną wiadomość do klienta, a także odbierać dane wysyłane przez niego w czasie rzeczywistym. Typowy przykład implementacji wygląda następująco:

python
Copy code
@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_text("Welcome to the chat room!")
    while True:
        data = await websocket.receive_text()
        logger.info(f"Message received: {data}")
        await websocket.send_text("Message received!")

Po uruchomieniu serwera, klient łączy się z endpointem /ws, co inicjuje połączenie WebSocket. Serwer w odpowiedzi wysyła powitaną wiadomość "Welcome to the chat room!" oraz czeka na wiadomości od klienta. Każda wiadomość otrzymana od klienta jest rejestrowana na konsoli serwera, a następnie odsyłana jest odpowiedź potwierdzająca odbiór. Dzięki temu, obie strony mogą komunikować się na bieżąco, bez konieczności ponownego nawiązywania połączenia.

Jednak ta podstawowa konfiguracja jest zaledwie początkiem. Kluczowym elementem w zarządzaniu połączeniami WebSocket jest odpowiednia obsługa rozłączeń oraz sytuacji, gdy połączenie powinno zostać przerwane. W tym przypadku, mamy do czynienia z dwoma głównymi scenariuszami:

1. Rozłączenie po stronie klienta:
   Jeśli klient zakończy połączenie (np. zamknie aplikację), serwer powinien odpowiednio zareagować, aby uniknąć błędów i niekontrolowanego zamknięcia połączenia. Można to zrobić, stosując blok try-except, aby wychwycić wyjątek WebSocketDisconnect, który jest rzucany, gdy połączenie zostaje zakończone przez klienta.

python
Copy code
from fastapi.websockets import WebSocketDisconnect

@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_text("Welcome to the chat room!")
    try:
        while True:
            data = await websocket.receive_text()
            logger.info(f"Message received: {data}")
            await websocket.send_text("Message received!")
    except WebSocketDisconnect:
        logger.warning("Connection closed by the client")

W tym przykładzie, po wykryciu zakończenia połączenia przez klienta, serwer wyświetli odpowiedni komunikat ostrzegawczy i pozwoli na zamknięcie połączenia bez wywoływania błędów.

2. Rozłączenie po stronie serwera:
   Czasami może być konieczne, by serwer zakończył połączenie na podstawie określonych warunków, takich jak odbiór specjalnej wiadomości. Na przykład, serwer może zamknąć połączenie, jeśli otrzyma wiadomość "disconnect". Tego rodzaju obsługę można dodać w funkcji endpointu, monitorując odbierane wiadomości:

python
Copy code
@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_text("Welcome to the chat room!")
    while True:
        data = await websocket.receive_text()
        if data == "disconnect":
            await websocket.send_text("Closing connection")
            await websocket.close()
            break
        logger.info(f"Message received: {data}")
        await websocket.send_text("Message received!")

Dzięki temu, kiedy serwer otrzyma wiadomość "disconnect", natychmiast zakończy połączenie, wysyłając komunikat do klienta i zamykając połączenie.

Obie te techniki pozwalają na bardziej zaawansowaną kontrolę nad połączeniami WebSocket, a także umożliwiają implementację logiki specyficznej dla aplikacji. Jednak równie istotne jest zapewnienie, że komunikacja WebSocket jest bezpieczna i odpornie zarządza nieoczekiwanymi błędami. Należy pamiętać, że każdy błąd, który wystąpi w procesie komunikacji, może prowadzić do utraty połączenia lub jego nieprawidłowego zamknięcia. Odpowiednia obsługa błędów i monitorowanie stanu połączeń to kluczowe elementy w budowie stabilnych aplikacji WebSocket.

Warto również zauważyć, że FastAPI umożliwia wykorzystanie dodatkowych funkcji do walidacji danych, które przechodzą przez połączenie WebSocket. Możemy na przykład używać metod send_json() oraz receive_json() do obsługi wiadomości w formacie JSON, co pozwala na łatwiejsze zarządzanie danymi o strukturze bardziej złożonej niż proste teksty.

Wszystkie te elementy składają się na bardziej zaawansowaną i stabilną architekturę aplikacji opartej na WebSocket, umożliwiającą skuteczną i bezpieczną komunikację w czasie rzeczywistym.