FastAPI Backend Development Bootcamp · Lekcja

Lekkie odciążanie za pomocą BackgroundTasks

Używaj wbudowanego w FastAPI BackgroundTasks do efektów ubocznych typu fire-and-forget bez blokowania odpowiedzi.

Lekcja 1 z 413 kroki

Lekkie odciążanie za pomocą BackgroundTasks to bezpłatna lekcja FastAPI Backend Development Bootcamp na CoddyKit. To lekcja 1 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej FastAPI Backend Development Bootcamp, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs FastAPI Backend Development Bootcamp zawiera 4 lekcji w sumie.

Dlaczego odciążać główny przepływ?

Gdy klient wysyła żądanie, czeka na odpowiedź. Jeśli endpoint dodatkowo wysyła wiadomość powitalną, zapisuje dziennik audytowy lub rozgrzewa pamięć podręczną, użytkownik musi czekać na operacje, które nie są dla niego istotne.

Efekty uboczne typu fire-and-forget to zadania, które powinny uruchomić się po wysłaniu odpowiedzi, bez jej blokowania:

Wysyłanie wiadomości e-mail z powiadomieniami
Zapisywanie dzienników analitycznych lub audytowych
Unieważnianie lub rozgrzewanie pamięci podręcznych
Czyszczenie plików tymczasowych

FastAPI udostępnia wbudowane narzędzie przeznaczone dokładnie do tego celu: BackgroundTasks.

Deklarowanie BackgroundTasks

Aby z niego skorzystać, dodaj do funkcji obsługującej operację ścieżki parametr typowany jako BackgroundTasks. FastAPI rozpoznaje adnotację typu i wstrzykuje instancję, tak jak w przypadku każdej innej zależności.

Następnie rejestrujesz pracę za pomocą .add_task(func, *args, **kwargs). Funkcja nie jest wywoływana natychmiast — zostaje dodana do kolejki i uruchomi się po zwróceniu odpowiedzi.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_log(message: str) -> None:
    with open("log.txt", mode="a") as f:
        f.write(message + "\n")


@app.post("/signup")
async def signup(email: str, tasks: BackgroundTasks):
    tasks.add_task(write_log, f"signup: {email}")
    return {"status": "accepted"}

Kolejność wykonywania

Najważniejszy szczegół: zadania w tle są uruchamiane po wysłaniu odpowiedzi do klienta, ale nadal w ramach tego samego procesu serwera.

Endpoint zwraca swój dict lub Response.
FastAPI przesyła odpowiedź przez sieć.
Dopiero wtedy wykonuje każde zadanie z kolejki, w kolejności, w której je dodano.

Dzięki temu użytkownik otrzymuje natychmiastową odpowiedź w stylu 202, podczas gdy wiadomość e-mail lub wpis w dzienniku są obsługiwane w tle.

Przekazywanie argumentów do zadania

Argumenty przekazane do add_task są przechowywane i przekazywane dalej, gdy zadanie zostanie ostatecznie uruchomione. Działają zarówno argumenty pozycyjne, jak i nazwane.

Ten wzorzec utrzymuje logikę efektów ubocznych w zwykłej funkcji, którą łatwo testować jednostkowo w izolacji, całkowicie niezależnie od FastAPI.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def send_email(to: str, subject: str, body: str) -> None:
    # imagine an SMTP client here
    print(f"Sending to {to}: {subject}")


@app.post("/orders")
async def create_order(email: str, tasks: BackgroundTasks):
    order_id = 1234
    tasks.add_task(
        send_email,
        to=email,
        subject="Order confirmed",
        body=f"Your order {order_id} is on the way!",
    )
    return {"order_id": order_id}

Synchroniczne a asynchroniczne funkcje zadań

Funkcja zadania może być zwykłą funkcją def albo funkcją async def.

Zadanie async jest bezpośrednio oczekiwane w pętli zdarzeń.
Zadanie będące zwykłą funkcją def jest uruchamiane w puli wątków, aby nie blokować pętli.

Zasada praktyczna: jeśli efekt uboczny wykonuje blokujące operacje wejścia-wyjścia (zapis pliku, synchroniczny sterownik bazy danych), zwykła funkcja def jest odpowiednia — FastAPI przeniesie ją do wątku. Używaj async def tylko wtedy, gdy rzeczywiście oczekujesz na asynchroniczne operacje wejścia-wyjścia.

async def notify_async(user_id: int) -> None:
    # awaits an async HTTP client, for example
    await some_async_push(user_id)


def notify_sync(user_id: int) -> None:
    # blocking call, run in a threadpool by FastAPI
    requests_post(user_id)

Dodawanie wielu zadań

Możesz wywołać add_task dowolną liczbę razy. Zadania są wykonywane sekwencyjnie, dokładnie w kolejności dodania, a każde z nich musi się zakończyć, zanim rozpocznie się następne.

Ponieważ uruchamiają się jedno po drugim, wolne zadanie opóźnia zadania znajdujące się za nim w kolejce, ale nigdy nie opóźnia samej odpowiedzi HTTP.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


@app.post("/publish")
async def publish(post_id: int, tasks: BackgroundTasks):
    tasks.add_task(reindex_search, post_id)
    tasks.add_task(invalidate_cache, post_id)
    tasks.add_task(notify_followers, post_id)
    return {"published": post_id}

Używanie BackgroundTasks w zależnościach

Przydatna możliwość: zależność również może deklarować parametr BackgroundTasks i dodawać zadania do kolejki. FastAPI łączy wszystko w jeden wspólny zestaw zadań dla danego żądania.

Dzięki temu elementy przekrojowe, takie jak rejestrowanie audytu, mogą znajdować się w wielokrotnego użytku zależności zamiast być kopiowane do każdego endpointu.

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def audit(action: str, tasks: BackgroundTasks):
    tasks.add_task(write_audit_row, action)
    return action


@app.delete("/items/{item_id}")
async def delete_item(item_id: int, action=Depends(audit)):
    return {"deleted": item_id}

Model mentalny kolejki zadań w czystym Pythonie

Wewnętrznie BackgroundTasks to niewiele więcej niż lista obiektów wywoływalnych uruchamianych po zwróceniu odpowiedzi. Możesz odwzorować tę ideę w czystym Pythonie, aby lepiej ją zrozumieć.

Poniższy fragment jest samodzielny i nie wymaga FastAPI; pokazuje wzorzec dodawania zadań, a następnie uruchamiania ich z opóźnieniem.

class TaskList:
    def __init__(self):
        self.tasks = []

    def add_task(self, func, *args, **kwargs):
        self.tasks.append((func, args, kwargs))

    def run_all(self):
        for func, args, kwargs in self.tasks:
            func(*args, **kwargs)


def log(msg):
    print("LOG:", msg)


q = TaskList()
q.add_task(log, "user signed up")
q.add_task(log, "email queued")
print("response sent")
q.run_all()

Obsługa błędów w zadaniach

Ponieważ zadanie jest uruchamiane po wysłaniu odpowiedzi, jego niepowodzenia nie można już zamienić na błąd HTTP — klient otrzymał już odpowiedź 200.

Nieobsłużony wyjątek w zadaniu w tle zostanie zapisany przez serwer w dzienniku, ale będzie niewidoczny dla klienta. Zawsze opakowuj ryzykowne operacje w try/except i samodzielnie zdecyduj o strategii ponawiania lub obsłudze kolejki niedostarczonych wiadomości.

def send_receipt(order_id: int) -> None:
    try:
        deliver_email(order_id)
    except Exception as exc:
        # the client already has its 200, so log and recover here
        logger.exception("receipt failed for %s: %s", order_id, exc)
        schedule_retry(order_id)

Najważniejsze ograniczenie: ten sam proces

BackgroundTasks działa w tym samym procesie roboczym co aplikacja. Wiąże się to z istotnymi ograniczeniami:

Intensywne obliczeniowo zadania nadal zużywają zasoby tego procesu roboczego.
Jeśli proces ulegnie awarii lub zostanie ponownie wdrożony, zadania znajdujące się w kolejce zostaną utracone, ponieważ nie ma trwałego przechowywania.
Zadania nie zachowują się między wieloma maszynami i nie można ich skalować poziomo.

To doskonałe rozwiązanie dla lekkich efektów ubocznych wykonywanych bez gwarancji, ale nie dla niezawodnych, długotrwałych ani rozproszonych zadań.

Kiedy zamiast tego wybrać Celery

Proszę wybrać BackgroundTasks, gdy praca jest krótka, niekrytyczna i jej utrata w razie awarii jest akceptowalna, na przykład przy wysyłaniu wiadomości e-mail, zwiększaniu licznika lub usuwaniu pliku tymczasowego.

Po Celery lub inną rozproszoną kolejkę (RQ, Dramatiq, Arq) należy sięgnąć, gdy potrzebne są:

Trwałość — zadania przetrwają ponowne uruchomienia dzięki brokerowi, takiemu jak Redis/RabbitMQ.
Ponowienia prób, planowanie i ograniczanie częstotliwości.
Skalowanie poziome między dedykowanymi maszynami roboczymi.
Intensywne obliczeniowo zadania, które w przeciwnym razie wyczerpywałyby zasoby procesów obsługujących żądania.

Szybkie sprawdzenie

Sprawdź swoje rozumienie tego, kiedy BackgroundTasks jest właściwym narzędziem.

Odpowiedz w interaktywnej lekcji

Podsumowanie

Najważniejsze informacje:

Dodaj parametr BackgroundTasks i wywołaj add_task(func, *args, **kwargs), aby odroczyć wykonanie efektów ubocznych.
Zadania są wykonywane po wysłaniu odpowiedzi, sekwencyjnie, w tym samym procesie roboczym.
Zadania synchroniczne def są wykonywane w puli wątków, a zadania async def — w pętli zdarzeń.
Zależności również mogą umieszczać zadania w kolejce, co świetnie sprawdza się w przypadku zagadnień przekrojowych, takich jak audyt.
Brak trwałego przechowywania: klient nie otrzymuje informacji o błędach, a zadania kończą działanie wraz z procesem.
Należy używać tego rozwiązania do lekkiej pracy wykonywanej bez gwarancji; w przypadku trwałych, ponawialnych, rozproszonych lub intensywnych obliczeniowo zadań należy wybrać Celery.

Bezpłatny start

Ucz się FastAPI Backend Development Bootcamp dzięki korepetycjom AI — za darmo

Pisz i uruchamiaj kod w przeglądarce, otrzymuj natychmiastową pomoc od korepetytora AI dostępnego 24/7 i kontynuuj naukę w sieci lub w aplikacji.

Kursy: 21
Lekcje: 84

Zacznij naukę bezpłatnie Wyświetl pełny kurs

Często zadawane pytania

Czy lekcja „Lekkie odciążanie za pomocą BackgroundTasks” jest bezpłatna?

Tak — pełny tekst „Lekkie odciążanie za pomocą BackgroundTasks” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu FastAPI Backend Development Bootcamp, przejdź na CoddyKit PRO. Kurs FastAPI Backend Development Bootcamp zawiera 4 lekcji w sumie.

Co nauczysz się w „Lekkie odciążanie za pomocą BackgroundTasks”?

Używaj wbudowanego w FastAPI BackgroundTasks do efektów ubocznych typu fire-and-forget bez blokowania odpowiedzi. Ćwiczysz FastAPI Backend Development Bootcamp z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć FastAPI Backend Development Bootcamp?

Nie wymagamy żadnego doświadczenia. FastAPI Backend Development Bootcamp w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 1 z 4.

Ile czasu zajmuje lekcja „Lekkie odciążanie za pomocą BackgroundTasks”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji FastAPI Backend Development Bootcamp?

Tak. Każda lekcja FastAPI Backend Development Bootcamp zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

Lekkie odciążanie za pomocą BackgroundTasks
Podłączanie workerów Celery do aplikacji FastAPI
Ponowienia, idempotencja i obsługa niedostarczonych komunikatów
Zadania zaplanowane i cykliczne za pomocą Celery Beat

← Powrót do FastAPI Backend Development Bootcamp