Flask — Przewodnik po ćwiczeniach

Niniejszy przewodnik omawia wszystkie dziesięć ćwiczeń z katalogu flask-assignments/. Flask to mikro-framework: dostarcza mechanizm routingu oraz kontekst żądania, lecz celowo pozostawia deweloperowi decyzje dotyczące baz danych, walidacji i uwierzytelniania. Dzięki temu każdy komponent jest widoczny i jawny, co czyni go doskonałym narzędziem do nauki.

---

Wymagania wstępne

Wersja Pythona

Flask 3.x wymaga Pythona 3.8 lub nowszego. Sprawdź zainstalowaną wersję:

python --version

Środowiska wirtualne

Dla każdego ćwiczenia należy tworzyć osobne środowisko wirtualne, aby uniknąć konfliktów między zależnościami różnych projektów.

python -m venv venv
source venv/bin/activate   # macOS/Linux
venv\Scripts\activate      # Windows

Narzędzia do testowania

Flask nie generuje interaktywnej dokumentacji API w stylu Swaggera. Należy korzystać z jednego z poniższych narzędzi:

• Przeglądarka — wyłącznie do testowania endpointów GET
• curl — klient HTTP działający w wierszu poleceń; przykłady poleceń znajdują się w plikach README każdego ćwiczenia
• Postman — graficzna aplikacja do komponowania i wysyłania żądań HTTP

Uruchamianie aplikacji Flask

W odróżnieniu od FastAPI (który korzysta z uvicorn), aplikacje Flask uruchamia się bezpośrednio:

python app.py

Wbudowany serwer deweloperski Flask nasłuchuje domyślnie pod adresem http://127.0.0.1:5000.

---

Ćwiczenie 01: Podstawowe API listy zadań

Lokalizacja: flask-assignments/01-basic-todo-api/

Co się nauczysz

• Routing we Flasku z użyciem dekoratora @app.route i parametru methods
• Odczytywanie JSON z treści żądania (request body)
• Zwracanie odpowiedzi JSON
• Dynamiczne parametry URL z konwerterami typów
• Obsługa błędów za pomocą abort

Kluczowe pojęcia

Routing we Flasku łączy wzorce URL z funkcjami Pythona. W przeciwieństwie do FastAPI, Flask używa jednego dekoratora @app.route wraz z listą methods, która określa akceptowane metody HTTP. Trasa bez jawnie podanego parametru methods domyślnie obsługuje wyłącznie metodę GET.

request.json to sparsowana reprezentacja przychodzącego body w formacie JSON. Jeśli żądanie nie zawiera nagłówka Content-Type: application/json lub body nie jest poprawnym JSON-em, request.json zwraca None. W odróżnieniu od FastAPI, który automatycznie waliduje strukturę danych, Flask wymaga ręcznej walidacji po stronie dewelopera.

jsonify() tworzy obiekt Response z nagłówkiem Content-Type: application/json i danymi zserializowanymi do formatu JSON. Zwrócenie zwykłego słownika Pythona z trasy Flask też działa w nowoczesnych wersjach frameworka, jednak jsonify() jest bardziej jawne i pozwala na precyzyjne ustawienie kodu statusu HTTP: return jsonify(data), 201.

abort(404) rzuca wyjątek HTTPException, który Flask przetwarza na odpowiedź z błędem. Domyślnie Flask zwraca stronę HTML z opisem błędu. Aby zamiast tego zwracać błędy w formacie JSON, należy zarejestrować własny handler: @app.errorhandler(404).

Instalacja

pip install Flask

Uruchomienie serwera

python app.py

Pliki projektu

app.py

Plik zawiera wszystko: instancję aplikacji, dane przechowywane w pamięci oraz definicje wszystkich tras.

app = Flask(__name__)

Argument __name__ informuje Flask o nazwie bieżącego modułu, co jest wykorzystywane do lokalizowania zasobów (szablony, pliki statyczne).

Lista todos i licznik next_id to zmienne globalne na poziomie modułu. W trybie deweloperskim Flask działa jednowątkowo, więc równoległy dostęp nie stanowi problemu. W środowisku produkcyjnym z wieloma procesami roboczymi wystąpiłyby wyścigi (race conditions) — w rzeczywistych aplikacjach należy korzystać z bazy danych.

Wzorzec definicji trasy:

@app.route('/todos/<int:todo_id>', methods=['PUT'])
def update_todo(todo_id):
    ...

<int:todo_id> to konwerter URL. Flask wyodrębnia todo_id z adresu URL, konwertuje go na liczbę całkowitą i przekazuje jako argument do funkcji. Inne dostępne konwertery to: <string:name>, <float:value>, <path:filepath>.

Testowanie krok po kroku

# Utwórz zadanie
curl -X POST http://127.0.0.1:5000/todos \
  -H "Content-Type: application/json" \
  -d '{"title": "Zakupy spożywcze"}'

# Pobierz wszystkie zadania
curl http://127.0.0.1:5000/todos

# Zaktualizuj zadanie o id 1
curl -X PUT http://127.0.0.1:5000/todos/1 \
  -H "Content-Type: application/json" \
  -d '{"title": "Zakupy spożywcze i mleko"}'

# Usuń zadanie o id 1
curl -X DELETE http://127.0.0.1:5000/todos/1

Typowe problemy

• 405 Method Not Allowed — trasa istnieje, ale nie akceptuje użytej metody HTTP. Sprawdź listę methods w dekoratorze @app.route.
• 400 Bad Request / None z request.json — w żądaniu brakuje nagłówka Content-Type: application/json.

---

Ćwiczenie 02: Blog osobisty

Lokalizacja: flask-assignments/02-personal-blog/

Co się nauczysz

• Integracja z bazą danych za pomocą Flask-SQLAlchemy
• Definiowanie modeli SQLAlchemy jako klas Pythona
• Tworzenie i wyszukiwanie rekordów w bazie danych
• Konwersja obiektów modelu do słowników serializowalnych do JSON

Kluczowe pojęcia

Flask-SQLAlchemy to rozszerzenie integrujące SQLAlchemy z Flaskiem. Dostarcza obiekt db, który zarządza silnikiem bazy danych, sesją i klasą bazową modeli. Obsługuje również cykl życia sesji automatycznie — sesja jest tworzona na początku żądania i zamykana po jego zakończeniu.

Klasa Model w Flask-SQLAlchemy udostępnia atrybut query do wykonywania zapytań do bazy danych. Najczęściej stosowane wzorce:

• Post.query.all() — pobierz wszystkie wiersze
• Post.query.filter_by(slug=slug).first() — filtruj po konkretnej kolumnie, zwróć pierwszy wynik lub None
• Post.query.get(id) — pobierz rekord po kluczu głównym

db.session.add(obj) kolejkuje obiekt do wstawienia do bazy. db.session.commit() zapisuje oczekujące zmiany do bazy danych. Jeśli commit() rzuci wyjątek (np. naruszenie ograniczenia unikalności), należy wywołać db.session.rollback(), aby zresetować sesję.

Instalacja

pip install Flask Flask-SQLAlchemy

Uruchomienie serwera

python app.py

SQLAlchemy tworzy plik blog.db przy pierwszym uruchomieniu.

Pliki projektu

app.py

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///blog.db'
db = SQLAlchemy(app)

db.create_all() jest wywoływane wewnątrz bloku with app.app_context(), aby zagwarantować aktywność kontekstu aplikacji podczas tworzenia tabel.

models.py

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    slug = db.Column(db.String(200), unique=True, nullable=False)
    content = db.Column(db.Text, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    def to_dict(self):
        return { "id": self.id, "title": self.title, ... }

Parametr unique=True na kolumnie slug tworzy ograniczenie unikalności na poziomie bazy danych. Nawet jeśli kod aplikacji pominie odpowiedni sprawdzenie, baza danych odrzuci duplikaty.

Metoda to_dict() konwertuje obiekt SQLAlchemy na zwykły słownik Pythona. Jest to konieczne, ponieważ jsonify() nie potrafi bezpośrednio serializować obiektów SQLAlchemy.

Testowanie krok po kroku

# Utwórz wpis
curl -X POST http://127.0.0.1:5000/posts \
  -H "Content-Type: application/json" \
  -d '{"title": "Hello World", "slug": "hello-world", "content": "Pierwszy wpis."}'

# Pobierz wpis po slugu
curl http://127.0.0.1:5000/posts/hello-world

# Zatrzymaj serwer, uruchom ponownie i pobierz ten sam wpis
# Wpis nadal istnieje — SQLite utrwalił dane

---

Ćwiczenie 03: Profil użytkownika i przesyłanie plików

Lokalizacja: flask-assignments/03-user-profile-uploads/

Co się nauczysz

• Przesyłanie plików we Flasku poprzez request.files
• Zabezpieczanie nazw przesyłanych plików
• Serwowanie plików statycznych przez Flask
• Odczytywanie pól formularza przez request.form

Kluczowe pojęcia

request.files to słownik przesłanych plików. Każda wartość jest obiektem FileStorage (z biblioteki Werkzeug, będącej podstawą Flaska). Obiekt FileStorage udostępnia:

• .filename — oryginalna nazwa pliku podana przez klienta
• .content_type — typ MIME
• .save(path) — zapisuje plik pod wskazaną ścieżką lokalną

secure_filename(filename) z modułu werkzeug.utils oczyszcza nazwy plików. Usuwa separatory ścieżek i inne znaki, które mogłyby umożliwić atak path traversal. Przykładowo secure_filename("../../etc/passwd") zwraca "etc_passwd".

Pliki statyczne we Flasku: gdy parametr static_folder jest ustawiony (domyślna wartość to "static"), Flask serwuje pliki z tego katalogu pod prefiksem URL /static/. Plik zapisany w static/uploads/avatar.png jest dostępny pod adresem http://localhost:5000/static/uploads/avatar.png.

request.form zawiera pola przesłane jako application/x-www-form-urlencoded lub multipart/form-data. Do odczytu wartości używaj request.form.get("nazwa_pola") — metoda ta zwraca None przy braku klucza zamiast rzucać wyjątek KeyError.

Instalacja

pip install Flask

Werkzeug (dostarczający secure_filename) jest instalowany automatycznie jako zależność Flaska.

Uruchomienie serwera

python app.py

Testowanie krok po kroku

# Pobierz profil użytkownika
curl http://127.0.0.1:5000/profiles/john_doe

# Prześlij zdjęcie profilowe
curl -X POST -F "file=@./photo.jpg" \
  http://127.0.0.1:5000/profiles/john_doe/upload-image

# Zaktualizuj biogram przez formularz
curl -X PATCH -F "bio=Programista Python" \
  http://127.0.0.1:5000/profiles/john_doe

Po przesłaniu pliku otwórz zwrócony adres profile_image_url w przeglądarce, aby upewnić się, że obraz jest poprawnie serwowany.

---

Ćwiczenie 04: Menedżer zadań z przetwarzaniem w tle

Lokalizacja: flask-assignments/04-task-manager-bg-tasks/

Co się nauczysz

• Wykonywanie zadań w tle z użyciem modułu threading z biblioteki standardowej Pythona
• Różnica między synchronicznym modelem Flaska a asynchronicznym modelem FastAPI
• Kiedy użycie wątków jest wystarczające, a kiedy konieczna jest dedykowana kolejka zadań

Kluczowe pojęcia

Flask jest domyślnie synchroniczny. Każda funkcja obsługująca żądanie wykonuje się od początku do końca w jednym wątku, zanim zostanie wysłana odpowiedź. Brak jest wbudowanego odpowiednika BackgroundTasks z FastAPI.

threading.Thread z biblioteki standardowej Pythona tworzy nowy wątek systemowy. Wywołując thread.start() bez thread.join(), wątek główny (obsługujący żądanie HTTP) kontynuuje działanie bez oczekiwania na zakończenie wątku pobocznego. Odpowiedź HTTP jest wysyłana natychmiast; wątek poboczny działa współbieżnie.

Ograniczenia wątków we Flasku:

• Global Interpreter Lock (GIL): GIL Pythona uniemożliwia prawdziwe równoległe wykonanie kodu CPU, jednak zadania I/O-bound (wywołania sieciowe, operacje plikowe) zwalniają GIL i są efektywnie wykonywane współbieżnie.
• Brak trwałości: jeśli proces zakończy się awaryjnie, wszystkie zadania w toku przepadają.
• Brak mechanizmu ponowień: w przypadku niepowodzenia symulowanego wysyłania e-maila nic nie ponowi próby.

W środowiskach produkcyjnych należy korzystać z dedykowanej kolejki zadań. Celery (z Redisem lub RabbitMQ jako brokerem) to standardowe rozwiązanie dla aplikacji Flask — utrwala zadania, umożliwia ponowne próby przy błędach i oferuje narzędzia do monitorowania.

Instalacja

pip install Flask

Moduł threading jest częścią biblioteki standardowej Pythona.

Uruchomienie serwera

python app.py

Pliki projektu

app.py

Endpoint complete_task:

thread = threading.Thread(target=send_task_notification,
                          kwargs={"email": task["user_email"],
                                  "task_title": task["title"]})
thread.start()
return jsonify({"id": task_id, "completed": True}), 200

Wątek jest uruchamiany przez thread.start(). Odpowiedź jest zwracana bezpośrednio po tym wywołaniu — wątek poboczny kontynuuje działanie niezależnie.

email_utils.py

def send_task_notification(email: str, task_title: str):
    logging.info("Rozpoczynanie zadania w tle...")
    time.sleep(5)
    logging.info(f"Powiadomienie WYSŁANE do {email}: zadanie '{task_title}' ukończone.")

Testowanie krok po kroku

1. Uruchom serwer z widocznym oknem terminala.
2. Utwórz zadanie: curl -X POST http://127.0.0.1:5000/tasks -H "Content-Type: application/json" -d '{"title": "Deploy app", "user_email": "dev@example.com"}'
3. Zanotuj identyfikator zadania (np. 1).
4. Oznacz zadanie jako ukończone: curl -X POST http://127.0.0.1:5000/tasks/1/complete
5. Odpowiedź pojawi się natychmiast. Pięć sekund później w terminalu pojawi się log Powiadomienie WYSŁANE do dev@example.com.

---

Ćwiczenie 05: Panel pogodowy

Lokalizacja: flask-assignments/05-weather-dashboard/

Co się nauczysz

• Wykonywanie synchronicznych żądań HTTP z poziomu Flaska przy użyciu biblioteki requests
• Parsowanie odpowiedzi JSON z zewnętrznych API
• Budowanie cache w pamięci z mechanizmem TTL

Kluczowe pojęcia

Biblioteka requests to standardowy synchroniczny klient HTTP dla Pythona. Oferuje prosty interfejs: requests.get(url, params=dict). W odróżnieniu od httpx (stosowanego w wersji FastAPI) biblioteka requests jest synchroniczna — wątek blokuje się do momentu otrzymania odpowiedzi.

Synchroniczne vs asynchroniczne I/O: we Flasku zablokowanie na requests.get(...) oznacza, że wątek nie może obsługiwać innych żądań podczas oczekiwania. Dla aplikacji edukacyjnych o niewielkim ruchu jest to akceptowalne. W systemach produkcyjnych o wysokiej współbieżności konieczne jest asynchroniczne I/O lub uruchomienie wielu procesów roboczych.

requests.raise_for_status() rzuca wyjątek, gdy kod statusu HTTP wskazuje na błąd (4xx lub 5xx). Bez tego wywołania nieudana odpowiedź API zostałaby po cichu zwrócona jako bezużyteczny obiekt odpowiedzi.

request.args.get() odczytuje parametry zapytania z adresu URL. Dla URL /weather?city_name=London&lat=51.5 wywołanie request.args.get("city_name") zwraca "London". Brakujący parametr zwraca None (lub wartość domyślną, jeśli została podana).

Instalacja

pip install Flask requests

Uruchomienie serwera

python app.py

Wymagane jest aktywne połączenie z Internetem.

Pliki projektu

weather_service.py

class WeatherService:
    def get_weather(self, lat, lon):
        url = "https://api.open-meteo.com/v1/forecast"
        params = {"latitude": lat, "longitude": lon, "current_weather": True}
        response = requests.get(url, params=params)
        response.raise_for_status()
        return response.json()

app.py

Struktura cache:

weather_cache = {}
# Klucz: (round(lat, 1), round(lon, 1))
# Wartość: {"data": {...}, "timestamp": float}

Sprawdzenie TTL: time.time() - entry["timestamp"] < 900 (15 minut).

Testowanie krok po kroku

# Pierwsze żądanie — rzeczywiste wywołanie API
curl "http://127.0.0.1:5000/weather?city_name=London&lat=51.5&lon=-0.12"
# Odpowiedź zawiera "cached": false

# Drugie żądanie — odpowiedź z cache
curl "http://127.0.0.1:5000/weather?city_name=London&lat=51.5&lon=-0.12"
# Odpowiedź zawiera "cached": true

---

Ćwiczenie 06: Czat w czasie rzeczywistym

Lokalizacja: flask-assignments/06-real-time-chat/

Co się nauczysz

• WebSockety we Flasku z użyciem Flask-SocketIO
• Protokół Socket.IO w porównaniu z surowymi WebSocketami
• Rozgłaszanie zdarzeń w pokojach (rooms)
• Różnica WSGI/ASGI i dlaczego Flask wymaga asynchronicznego workera do obsługi WebSocketów

Kluczowe pojęcia

Flask-SocketIO dodaje obsługę WebSocketów (i protokołu Socket.IO) do Flaska. Surowy Flask korzysta z WSGI (Web Server Gateway Interface), który jest synchroniczny i nie obsługuje natywnie długotrwałych połączeń. Flask-SocketIO wymaga asynchronicznego backendu — eventlet lub gevent — który stosuje technikę monkey-patching na operacjach I/O Pythona.

Socket.IO to protokół zbudowany na bazie WebSocketów (z fallbackiem do HTTP long-polling). Dodaje mechanizmy pokojów (rooms), przestrzeni nazw (namespaces) i niezawodnego dostarczania zdarzeń. Klient JavaScript musi korzystać z biblioteki socket.io.js — surowy klient WebSocket nie będzie kompatybilny.

Zdarzenia: Socket.IO operuje na nazwanych zdarzeniach zamiast surowego strumienia wiadomości. @socketio.on('chat_message') rejestruje handler dla zdarzenia o nazwie chat_message. Klient JavaScript emituje zdarzenia przez socket.emit('chat_message', text).

emit(event, data, broadcast=True, include_self=False) wysyła zdarzenie do klientów. Parametr broadcast=True rozsyła wiadomość do wszystkich połączonych klientów w domyślnej przestrzeni nazw. include_self=False zapobiega otrzymaniu własnej wiadomości przez nadawcę.

Instalacja

pip install Flask Flask-SocketIO eventlet

Eventlet jest asynchronicznym backendem. Stosuje monkey-patching na standardowej bibliotece sieciowej Pythona, dzięki czemu blokujące wywołania stają się nieblokujące.

Uruchomienie serwera

python app.py

Uwaga: w pliku app.py należy używać socketio.run(app) zamiast standardowego app.run().

Pliki projektu

app.py

socketio = SocketIO(app, async_mode='eventlet', cors_allowed_origins='*')

@socketio.on('join')
def handle_join(data):
    client_id = data['client_id']
    emit('system', f'Klient {client_id} dołączył', broadcast=True)

@socketio.on('chat_message')
def handle_message(data):
    emit('chat_message', data, broadcast=True, include_self=False)

@socketio.on('disconnect')
def handle_disconnect():
    emit('system', 'Klient rozłączył się', broadcast=True)

Handler @socketio.on('disconnect') jest wywoływany automatycznie w momencie rozłączenia klienta.

Testowanie krok po kroku

1. Uruchom serwer.
2. Otwórz http://127.0.0.1:5000/ w jednej karcie przeglądarki, następnie w drugiej.
3. Każda karta wyświetla unikalny identyfikator klienta.
4. Wpisz wiadomość w jednej karcie i kliknij „Wyślij". Wiadomość pojawi się w obu kartach.
5. Zamknij jedną kartę. W drugiej pojawi się powiadomienie o rozłączeniu.

Typowe problemy

• RuntimeError: The server is not configured to use long-polling — eventlet nie jest zainstalowany.
• ImportError: cannot import name 'SocketIO' — Flask-SocketIO nie jest zainstalowany.

---

Ćwiczenie 07: Obsługa formularzy i szablony Jinja2

Lokalizacja: flask-assignments/07-forms-jinja2/

Co się nauczysz

• System szablonów Flaska (Jinja2)
• Różnica między render_template a render_template_string
• Przetwarzanie formularzy przez request.form
• Walidacja po stronie serwera i renderowanie komunikatów o błędach
• Wstępne wypełnianie pól formularza po nieudanym przesłaniu

Kluczowe pojęcia

Flask natywnie korzysta z Jinja2. Szablony są przechowywane w katalogu templates/ względem pliku app.py. Wywołanie render_template("index.html", name=name, errors=errors) renderuje plik szablonu i przekazuje podane zmienne do kontekstu Jinja2.

Wzorzec GET/POST w obsłudze formularzy we Flasku:

• GET / — wyrenderuj pusty formularz
• POST /subscribe — przetwórz przesłane dane

Realizuje się to przez zarejestrowanie dwóch tras z różnymi metodami. Alternatywnie można użyć jednej trasy z methods=["GET", "POST"] i warunkowego sprawdzenia request.method.

Składnia szablonów Jinja2:

• {{ variable }} — wyświetl wartość zmiennej
• {% if condition %}...{% endif %} — blok warunkowy
• {% for item in list %}...{% endfor %} — pętla
• {{ value if value else '' }} — warunkowy zapis inline (odpowiednik operatora trójargumentowego)

Instalacja

pip install Flask

Jinja2 jest zależnością Flaska i instaluje się automatycznie.

Uruchomienie serwera

python app.py

Pliki projektu

app.py

@app.route('/subscribe', methods=['POST'])
def subscribe():
    name = request.form.get('name', '')
    email = request.form.get('email', '')
    errors = {}
    if len(name) < 2:
        errors['name'] = 'Imię musi zawierać co najmniej 2 znaki.'
    if not re.match(r'^[^@]+@[^@]+\.[^@]+$', email):
        errors['email'] = 'Podaj prawidłowy adres e-mail.'
    if errors:
        return render_template('index.html', name=name, email=email, errors=errors), 422
    # poprawne dane — zapisz i wyświetl stronę sukcesu
    subscribers.append({"name": name, "email": email})
    return render_template('success.html', name=name, email=email)

Kod statusu 422 informuje, że serwer rozumiał żądanie, lecz odrzucił je z powodu błędów walidacji. Użycie 400 jest również akceptowalne, choć mniej precyzyjne.

templates/index.html

<input name="name" value="{{ name if name else '' }}">
{% if errors.name %}
  <p class="error">{{ errors.name }}</p>
{% endif %}

Wyrażenie errors.name w Jinja2 odwołuje się do klucza name słownika errors. Jest to równoważne errors["name"] w Pythonie — Jinja2 używa notacji z kropką zarówno dla atrybutów obiektów, jak i dla kluczy słowników.

Testowanie krok po kroku

1. Otwórz http://127.0.0.1:5000/.
2. Prześlij prawidłowe imię (co najmniej 3 znaki) i poprawny adres e-mail. Obserwuj stronę potwierdzenia.
3. Prześlij formularz z imieniem „A". Obserwuj komunikat błędu i sprawdź, że pole e-mail zachowuje wpisaną wartość.
4. Prześlij formularz z adresem e-mail „niepoprawny". Obserwuj komunikat błędu.

---

Ćwiczenie 08: Odbiornik webhooków

Lokalizacja: flask-assignments/08-webhook-receiver/

Co się nauczysz

• Odczytywanie surowego body żądania we Flasku
• Weryfikacja podpisu HMAC
• Porównywanie w stałym czasie z użyciem hmac.compare_digest
• Znaczenie weryfikacji webhooków przed ich parsowaniem

Kluczowe pojęcia

Szczegółowe wyjaśnienie dotyczące webhooków, algorytmu HMAC i ataków czasowych (timing attacks) znajduje się w ćwiczeniu 08 przewodnika FastAPI — koncepcje kryptograficzne są identyczne we wszystkich frameworkach. Specyfika Flaska jest następująca:

request.data we Flasku zawiera surowe body żądania jako bajty. To właśnie ta wartość musi być używana do weryfikacji HMAC. Jeśli zamiast tego odczytasz request.json, Flask przeparsuje body i oryginalna sekwencja bajtów przestanie być dostępna. Zawsze najpierw odczytuj request.data, a następnie parsuj JSON ręcznie: json.loads(request.data).

abort(401) rzuca wyjątek HTTPException ze statusem 401 Unauthorized. Flask przetwarza go na odpowiedź z błędem. Format odpowiedzi można dostosować za pomocą handlera błędu:

@app.errorhandler(401)
def unauthorised(e):
    return jsonify({"error": "Nieprawidłowy podpis"}), 401

Instalacja

pip install Flask

Wszystkie funkcje kryptograficzne (hmac, hashlib) są modułami biblioteki standardowej Pythona.

Testowanie krok po kroku

Skorzystaj ze skryptu testowego w języku Python dołączonego do pliku README ćwiczenia. Kluczowe scenariusze:

1. Wyślij poprawnie podpisane żądanie → odpowiedź 200 OK, zdarzenie zostaje zalogowane.
2. Zmień secret w skrypcie testowym → odpowiedź 401 Unauthorized.
3. Usuń nagłówek x-webhook-signature z żądania → odpowiedź 401 Unauthorized.
4. Wywołaj GET /admin/events, aby wyświetlić zalogowane zdarzenia.

---

Ćwiczenie 09: System zarządzania inwentarzem

Lokalizacja: flask-assignments/09-inventory-management/

Co się nauczysz

• Blueprinty Flask do organizacji kodu i prefiksowania URL
• Implementacja wersjonowania API przy użyciu oddzielnych modułów
• Paginacja offset/limit
• Różnice między prostą a strukturalną odpowiedzią paginowaną

Kluczowe pojęcia

Blueprinty Flask to modularne komponenty służące do organizacji aplikacji. Każdy blueprint może definiować własne trasy, handlery błędów i pliki statyczne. Główna aplikacja rejestruje blueprinty, opcjonalnie podając prefix URL:

app.register_blueprint(v1_bp, url_prefix='/v1')
app.register_blueprint(v2_bp, url_prefix='/v2')

Trasy zdefiniowane wewnątrz blueprintu otrzymują prefix podany podczas rejestracji. Trasa @v1_bp.route('/items') w pliku v1/routes.py staje się /v1/items w całej aplikacji.

Blueprinty vs APIRouter z FastAPI: koncepcja jest identyczna — oba rozwiązania umożliwiają podział dużej aplikacji na mniejsze, wyspecjalizowane moduły z własnymi prefiksami URL.

Instalacja

pip install Flask

Uruchomienie serwera

python app.py

Pliki projektu

v1/routes.py

v1_bp = Blueprint('v1', __name__)

@v1_bp.route('/items', methods=['GET'])
def list_items_v1():
    offset = int(request.args.get('offset', 0))
    limit = int(request.args.get('limit', 20))
    return jsonify(inventory_data[offset : offset + limit])

Zwraca surową tablicę JSON — bez metadanych.

v2/routes.py

v2_bp = Blueprint('v2', __name__)

@v2_bp.route('/items', methods=['GET'])
def list_items_v2():
    offset = int(request.args.get('offset', 0))
    limit = int(request.args.get('limit', 20))
    category = request.args.get('category')
    filtered = [i for i in inventory_data if not category or i['category'] == category]
    total = len(filtered)
    items = filtered[offset : offset + limit]
    return jsonify({
        "items": items,
        "total": total,
        "offset": offset,
        "limit": limit,
        "has_more": offset + limit < total
    })

app.py

from v1.routes import v1_bp
from v2.routes import v2_bp
app.register_blueprint(v1_bp, url_prefix='/v1')
app.register_blueprint(v2_bp, url_prefix='/v2')

Testowanie krok po kroku

# V1: prosta tablica
curl "http://127.0.0.1:5000/v1/items?offset=0&limit=5"

# V2: strukturalna odpowiedź z metadanymi
curl "http://127.0.0.1:5000/v2/items?offset=0&limit=5"

# V2: filtrowanie po kategorii
curl "http://127.0.0.1:5000/v2/items?category=Electronics&offset=0&limit=10"

---

Ćwiczenie 10: Bezpieczny sejf na dokumenty

Lokalizacja: flask-assignments/10-secure-document-vault/

Co się nauczysz

• Uwierzytelnianie JWT we Flasku z użyciem biblioteki PyJWT
• Haszowanie haseł przy użyciu passlib i bcrypt
• Autoryzacja oparta na własnych dekoratorach
• Wyodrębnianie Bearer token z nagłówka Authorization
• Kontrola dostępu bazująca na zakresach uprawnień (scopes) i poziomach poświadczeń bezpieczeństwa

Kluczowe pojęcia

Tło koncepcyjne dotyczące JWT, haszowania haseł i RBAC (Role-Based Access Control) znajduje się w ćwiczeniu 10 przewodnika FastAPI. Implementacja we Flasku różni się mechanizmem egzekwowania uprawnień:

Własne dekoratory: Flask nie posiada wbudowanego systemu Security(...) wzorem FastAPI. Kontrola dostępu jest implementowana jako dekorator Pythona owijający funkcje obsługujące trasy. Dekorator:

1. Odczytuje nagłówek Authorization: request.headers.get("Authorization")
2. Wyodrębnia token: wartość nagłówka ma format Bearer <token>
3. Dekoduje JWT przy użyciu PyJWT
4. Weryfikuje wymagane zakresy uprawnień (scopes)
5. Dołącza użytkownika do kontekstu żądania: request.user = user
6. Wywołuje oryginalną funkcję, jeśli wszystko jest poprawne; w przeciwnym razie zwraca kod 401 lub 403

def token_required(required_scopes=None):
    def decorator(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            auth_header = request.headers.get("Authorization")
            if not auth_header or not auth_header.startswith("Bearer "):
                return jsonify({"error": "Brak tokenu"}), 401
            token = auth_header.split(" ")[1]
            try:
                payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
            except jwt.ExpiredSignatureError:
                return jsonify({"error": "Token wygasł"}), 401
            except jwt.InvalidTokenError:
                return jsonify({"error": "Nieprawidłowy token"}), 401
            # weryfikacja zakresów, dołączenie użytkownika...
            return f(*args, **kwargs)
        return decorated
    return decorator

@wraps(f) z modułu functools zachowuje oryginalną nazwę i docstring funkcji, co jest istotne dla systemu routingu Flaska.

Instalacja

pip install Flask PyJWT passlib[bcrypt]

• PyJWT — kodowanie i dekodowanie tokenów JWT
• passlib[bcrypt] — haszowanie haseł

Uruchomienie serwera

python app.py

Testowanie krok po kroku

# Uwierzytelnij się jako alice
curl -X POST http://127.0.0.1:5000/token \
  -H "Content-Type: application/json" \
  -d '{"username": "alice", "password": "alice123"}'
# Skopiuj wartość access_token z odpowiedzi

# Pobierz dokumenty (wymaga zakresu vault:read)
curl http://127.0.0.1:5000/documents \
  -H "Authorization: Bearer <twój_token>"

# Utwórz dokument z secret_level powyżej poziomu poświadczeń alice (zakończy się błędem)
curl -X POST http://127.0.0.1:5000/documents \
  -H "Authorization: Bearer <twój_token>" \
  -H "Content-Type: application/json" \
  -d '{"title": "Ściśle tajne", "content": "...", "secret_level": 4}'
# Oczekiwana odpowiedź: 403 Forbidden

# Spróbuj uzyskać dostęp do endpointu administratora jako alice (zakończy się błędem — nieprawidłowy zakres)
curl http://127.0.0.1:5000/admin/users \
  -H "Authorization: Bearer <twój_token>"
# Oczekiwana odpowiedź: 403 Forbidden

Powtórz testy z danymi admin / admin123, aby potwierdzić, że dostęp na poziomie administratora działa poprawnie.