Asynchroniczność w Python web apps

Asynchroniczność w Pythonie rozwiązuje ten problem. Pozwala obsługiwać wiele żądań jednocześnie, bez potrzeby uruchamiania wielu wątków czy procesów. To podejście znacząco zwiększa wydajność i responsywność aplikacji. W tym przewodniku poznasz, jak działa asynchroniczność w Pythonie, jak używać async/await oraz jak implementować asynchroniczne aplikacje webowe w FastAPI i Django.

Jak działa asynchroniczność w Pythonie?

Python od wersji 3.5 wprowadził moduł asyncio oraz słowa kluczowe async i await, które umożliwiają pisanie kodu asynchronicznego w sposób czytelny i deklaratywny. Zamiast używać callbacków (jak w JavaScript), Python używa coroutines i event loop.

Podstawowy przykład asyncio

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
import asyncio

async def fetch_data():
print("Pobieram dane...")
await asyncio.sleep(2)# Symuluje operację I/O
print("Dane pobrane")
return {"data": 123}

async def main():
result = await fetch_data()
print(result)

asyncio.run(main())

Podczas await asyncio.sleep(2) Python nie blokuje programu, tylko przełącza się do innych zadań — to właśnie sedno asynchroniczności. Event loop zarządza wieloma coroutines i wykonuje je, gdy nie czekają na operacje I/O.

Współbieżne wykonywanie zadań

Asynchroniczność pozwala na równoległe wykonywanie wielu operacji:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import asyncio
import httpx

async def fetch_url(url):
async with httpx.AsyncClient() as client:
response = await client.get(url)
return response.json()

async def main():
urls = [
"https://api.example.com/data1",
"https://api.example.com/data2",
"https://api.example.com/data3",
]

# Wykonaj wszystkie równolegle
results = await asyncio.gather(*[fetch_url(url) for url in urls])
return results

# Zamiast 3 sekundy (3 × 1s synchronicznie), wykonuje się w ~1 sekundę

asyncio.gather() wykonuje wszystkie zadania równolegle i czeka, aż wszystkie się zakończą. To znacznie szybciej niż wykonywanie ich sekwencyjnie.

Asynchroniczność w aplikacjach webowych

Dzięki standardowi ASGI (Asynchronous Server Gateway Interface), Python może obsługiwać żądania HTTP w sposób asynchroniczny. ASGI to następca WSGI — pozwala na działanie serwerów takich jak Uvicorn, Hypercorn czy Daphne.

Frameworki takie jak FastAPI, Starlette, a w nowszych wersjach także Django, w pełni wspierają asynchroniczne przetwarzanie żądań. Różnica między WSGI a ASGI polega na tym, że ASGI pozwala na obsługę WebSocketów, Server-Sent Events i innych protokołów długożyjących połączeń.

Asynchroniczność w FastAPI

FastAPI to framework zbudowany od podstaw z myślą o asynchroniczności. Wszystkie endpointy mogą być asynchroniczne, co pozwala na obsługę tysięcy żądań jednocześnie na jednym procesie serwera.

Podstawowy przykład asynchronicznego endpointu w FastAPI

Python
Kopiuj
1
2
3
4
5
6
7
8
9
from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/data")
async def get_data():
await asyncio.sleep(1)# Symuluje kosztowną operację I/O
return {"status": "done", "message": "Dane pobrane asynchronicznie"}

Tutaj każde żądanie /data może czekać na dane asynchronicznie, nie blokując innych użytkowników. W praktyce oznacza to obsługę tysięcy żądań jednocześnie na jednym procesie serwera.

Równoległe wywołania zewnętrznych API

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from fastapi import FastAPI
import httpx

app = FastAPI()

@app.get("/aggregated-data")
async def get_aggregated_data():
async with httpx.AsyncClient() as client:
# Równoległe pobranie danych z wielu źródeł
user_task = client.get("http://user-service/users")
posts_task = client.get("http://post-service/posts")
stats_task = client.get("http://stats-service/stats")

user, posts, stats = await asyncio.gather(
user_task,
posts_task,
stats_task
)

return {
"user": user.json(),
"posts": posts.json(),
"stats": stats.json()
}

Dzięki asynchroniczności wszystkie trzy wywołania wykonują się równolegle, co skraca czas odpowiedzi z 3 sekund (jeśli byłyby synchroniczne) do ~1 sekundy.

Background tasks w FastAPI

FastAPI pozwala na wykonywanie zadań w tle po zwróceniu odpowiedzi:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_notification(email: str, message: str):
# Wyślij e-mail w tle
print(f"Sending email to {email}: {message}")

@app.post("/users")
async def create_user(user: UserCreate, background_tasks: BackgroundTasks):
new_user = create_user_in_db(user)

# Dodaj zadanie do wykonania w tle
background_tasks.add_task(send_notification, user.email, "Welcome!")

return new_user# Odpowiedź zwrócona od razu, e-mail wysłany w tle

Django i ASGI

Django przez wiele lat działał tylko w trybie synchronicznym (WSGI), ale od wersji 3.0 dodano wsparcie dla ASGI. Dzięki temu można łączyć klasyczne widoki synchroniczne z nowymi, asynchronicznymi endpointami.

Konfiguracja ASGI w Django

Najpierw stwórz plik asgi.py:

Python
Kopiuj
1
2
3
4
5
6
7
# asgi.py
import os
from django.core.asgi import get_asgi_application

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

application = get_asgi_application()

Asynchroniczny widok w Django

Python
Kopiuj
1
2
3
4
5
6
7
# views.py
from django.http import JsonResponse
import asyncio

async def async_view(request):
await asyncio.sleep(1)# Asynchroniczna operacja
return JsonResponse({"message": "Hello async Django!"})

Routing asynchronicznych widoków

Python
Kopiuj
1
2
3
4
5
6
7
# urls.py
from django.urls import path
from . import views

urlpatterns = [
path('async/', views.async_view),
]

W pliku asgi.py aplikacja jest uruchamiana przy pomocy ASGI serwera (np. Uvicorn):

Bash
Kopiuj
1
2
pip install uvicorn
uvicorn myproject.asgi:application

Mieszanie widoków synchronicznych i asynchronicznych

Django pozwala na mieszanie widoków synchronicznych i asynchronicznych:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
import asyncio

# Widok synchroniczny
def sync_view(request):
return JsonResponse({"type": "sync"})

# Widok asynchroniczny
async def async_view(request):
await asyncio.sleep(1)
return JsonResponse({"type": "async"})

Asynchroniczny middleware w Django

Możesz też tworzyć asynchroniczny middleware:

Python
Kopiuj
1
2
3
4
5
6
class AsyncMiddleware:
async def __call__(self, scope, receive, send):
# Asynchroniczna logika middleware
await asyncio.sleep(0.1)
# Przekaż request dalej
...

Różnica między async a threading

Zrozumienie różnicy między asynchronicznością a wielowątkowością jest kluczowe dla wyboru odpowiedniego podejścia:

Mechanizm	Czym jest	Zalety	Wady
Asynchroniczność (asyncio)	Jednowątkowy loop obsługujący wiele zadań	Małe zużycie pamięci, duża skalowalność dla I/O	Nie nadaje się do operacji CPU-bound
Wątki (threading)	Równoległe wątki systemowe	Dobre dla operacji I/O, łatwiejsze do zrozumienia	Większy narzut, blokady GIL w Pythonie
Procesy (multiprocessing)	Oddzielne procesy systemowe	Równoległość CPU, brak GIL	Większe zużycie zasobów, trudniejsza komunikacja

Kiedy używać asynchroniczności?

Asynchroniczność jest idealna dla operacji I/O-bound:

• zapytania do API
• operacje na bazach danych
• odczyt/zapis plików
• komunikacja przez sieć
• WebSockety

Kiedy NIE używać asynchroniczności?

Asynchroniczność nie przyspiesza operacji CPU-bound:

• obliczenia matematyczne
• przetwarzanie obrazów
• kompresja danych
• analiza dużych zbiorów danych

Dla operacji CPU-bound lepsze są procesy (multiprocessing):

Python
Kopiuj
1
2
3
4
5
6
7
8
9
import multiprocessing

def cpu_intensive_task(data):
# Ciężkie obliczenia CPU
return sum(i**2 for i in range(data))

# Użyj procesów dla operacji CPU
with multiprocessing.Pool() as pool:
results = pool.map(cpu_intensive_task, [1000000, 2000000, 3000000])

Typowe zastosowania asynchroniczności w Python web apps

Asynchroniczność znajduje wiele praktycznych zastosowań w aplikacjach webowych:

1. Integracje z zewnętrznymi API

Równoległe pobieranie danych z wielu źródeł:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
import httpx
import asyncio

async def fetch_all_data():
async with httpx.AsyncClient() as client:
tasks = [
client.get("https://api1.example.com/data"),
client.get("https://api2.example.com/data"),
client.get("https://api3.example.com/data"),
]
responses = await asyncio.gather(*tasks)
return [r.json() for r in responses]

2. Zapytania do baz danych

Async ORM-y (np. Tortoise ORM, SQLAlchemy async) pozwalają na wykonywanie zapytań bez blokowania wątku.

3. WebSockety

Komunikacja w czasie rzeczywistym (czat, live dashboardy):

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Echo: {data}")

4. Background tasks

Przetwarzanie danych w tle:

Python
Kopiuj
1
2
3
4
5
6
from fastapi import BackgroundTasks

@app.post("/process")
async def process_data(data: Data, background_tasks: BackgroundTasks):
background_tasks.add_task(heavy_processing, data)
return {"status": "processing"}

5. Mikroserwisy i event-driven systems

Asynchroniczna komunikacja między serwisami przez message queues lub gRPC.

Asynchroniczne ORM-y i bazy danych

Dzięki nowym narzędziom jak SQLAlchemy 2.0, Tortoise ORM czy encode/databases, możliwe jest wykonywanie zapytań do baz danych bez blokowania wątku.

SQLAlchemy async

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.future import select
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(100))
email = Column(String(100))

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession)

async def get_users():
async with AsyncSessionLocal() as session:
result = await session.execute(select(User))
users = result.scalars().all()
return users

Uwaga: Musisz użyć async drivera — np. asyncpg dla PostgreSQL (zamiast psycopg2).

Tortoise ORM

Tortoise ORM to asynchroniczny ORM inspirowany Django ORM:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
from tortoise.models import Model
from tortoise import fields

class User(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=100)
email = fields.CharField(max_length=100)

async def get_users():
return await User.all()

encode/databases

Prostszy interfejs do asynchronicznych zapytań:

Python
Kopiuj
1
2
3
4
5
6
7
8
import databases
import sqlalchemy

database = databases.Database("postgresql+asyncpg://user:pass@localhost/db")

async def get_users():
query = "SELECT * FROM users"
return await database.fetch_all(query=query)

Asynchroniczność a testy

Testowanie aplikacji asynchronicznych wymaga narzędzi wspierających async — np. pytest-asyncio.

Instalacja

Bash
Kopiuj
1
pip install pytest-asyncio

Przykład testu

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
import pytest
import httpx
from fastapi.testclient import TestClient

@pytest.mark.asyncio
async def test_async_endpoint():
async with httpx.AsyncClient(app=app, base_url="http://test") as client:
response = await client.get("/data")
assert response.status_code == 200
assert response.json()["status"] == "done"

Testowanie z bazą danych

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
@pytest.mark.asyncio
async def test_async_db_operation():
async with AsyncSessionLocal() as session:
user = User(name="Test", email="test@example.com")
session.add(user)
await session.commit()

result = await session.execute(select(User).where(User.email == "test@example.com"))
found_user = result.scalar_one()
assert found_user.name == "Test"

Kiedy NIE używać asynchroniczności

Nie wszystko powinno być asynchroniczne. Asynchroniczność nie przyspiesza obliczeń CPU i czasami zwiększa złożoność projektu.

Nie stosuj jej, gdy:

1. Aplikacja jest mała i nie ma wielu zewnętrznych zależności — narzut asynchroniczności może być większy niż korzyści.

2. Przetwarzasz głównie dane lokalne (CPU-bound) — asynchroniczność nie pomoże w obliczeniach CPU, a nawet może spowolnić.

3. Prostszy model synchroniczny wystarcza — jeśli aplikacja działa dobrze synchronicznie i nie ma problemów z wydajnością, nie komplikuj.

4. Nie masz doświadczenia z async — źle napisany kod asynchroniczny może być wolniejszy niż synchroniczny i trudniejszy w debugowaniu.

Przykład, gdy async nie pomoże

Python
Kopiuj
1
2
3
4
5
# To NIE przyspieszy operacji CPU
async def calculate_fibonacci(n):
if n <= 1:
return n
return await calculate_fibonacci(n-1) + await calculate_fibonacci(n-2)# Źle!

To tylko zwiększy narzut — lepiej użyć multiprocessing lub po prostu synchronicznej wersji.

Najlepsze praktyki

Przy pisaniu asynchronicznego kodu warto pamiętać o kilku zasadach:

1. Nie blokuj event loopa

Unikaj synchronicznych operacji blokujących w funkcjach async:

Python
Kopiuj
1
2
3
4
5
6
7
# Źle - blokuje event loop
async def bad_example():
time.sleep(5)# Używa synchronicznego sleep

# Dobrze - używa asynchronicznego sleep
async def good_example():
await asyncio.sleep(5)

2. Używaj asynchronicznych bibliotek

Dla operacji I/O zawsze używaj asynchronicznych bibliotek:

Python
Kopiuj
1
2
3
4
5
6
7
8
# Źle
import requests
response = requests.get("https://api.example.com")

# Dobrze
import httpx
async with httpx.AsyncClient() as client:
response = await client.get("https://api.example.com")

3. Obsługuj błędy w async

Pamiętaj o obsłudze błędów w asynchronicznym kodzie:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
async def fetch_with_retry(url, max_retries=3):
for attempt in range(max_retries):
try:
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=5.0)
return response.json()
except httpx.TimeoutException:
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt)# Exponential backoff

4. Używaj timeoutów

Zawsze ustawiaj timeouty dla operacji sieciowych:

Python
Kopiuj
1
2
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(url)

5. Mierz wydajność

Nie wszystko, co async, jest szybsze — mierz i porównuj:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
import time

# Porównaj wydajność
start = time.time()
# Synchroniczna wersja
result_sync = [sync_fetch(url) for url in urls]
print(f"Sync: {time.time() - start}s")

start = time.time()
# Asynchroniczna wersja
result_async = await asyncio.gather(*[async_fetch(url) for url in urls])
print(f"Async: {time.time() - start}s")

Podsumowanie

Asynchroniczność w Pythonie to potężne narzędzie, które pozwala tworzyć szybkie, skalowalne i nowoczesne aplikacje webowe. Dzięki asyncio, FastAPI i ASGI Python stał się pełnoprawnym graczem w świecie wysokowydajnych serwerów.

Kluczowe zasady:

• Używaj async/await w miejscach, gdzie występuje I/O — zapytania do baz, API, pliki.
• Nie blokuj event loopa — zawsze używaj asynchronicznych wersji bibliotek.
• Korzystaj z asynchronicznych bibliotek — bazy danych, HTTP, ORM.
• Mierz wydajność — nie wszystko, co async, jest szybsze. Testuj i porównuj.
• Rozumiej różnicę między I/O-bound a CPU-bound — async pomaga przy I/O, nie przy CPU.

Asynchroniczność nie jest panaceum — ma sens tam, gdzie masz dużo operacji I/O, które mogą być wykonywane równolegle. Dla prostych aplikacji lub operacji CPU-bound może być niepotrzebną komplikacją. Zacznij od prostych przypadków użycia i rozbudowuj w miarę potrzeb.