Strongly typed background tasks with FastAPI and Google Cloud Run, Tasks and Scheduler. This is a fork of fastapi-gcp-tasks, updated with new features and bug fixes.
📚 Documentation: simplifyjobs.github.io/fastapi-gcp-tasks
sequenceDiagram
autonumber
actor User
participant Service
participant CloudTasks
participant Worker
User ->>+ Service: /trigger
rect rgb(100,130,180)
note right of Service: hello.delay()
Service -->>+ CloudTasks: Create task
CloudTasks -->>- Service: Accepted
end
Service ->>- User: Hello task triggered
note right of CloudTasks: Async
CloudTasks -->>+ Worker: /hello
Worker -->>- CloudTasks: 200
pip install fastapi-gcp-tasks
- Strongly typed tasks.
- Fail at invocation site to make it easier to develop and debug.
- Breaking schema changes between versions will fail at task runner with Pydantic.
- Fully type-checked public API (PEP 561
py.typed), with opt-in static typing for.delay/.scheduler— see Type safety.
- Familiar and simple public API
.delaymethod that takes same arguments as the task..schedulermethod to create recurring job.- Async variants (
AsyncDelayedRouteBuilder/AsyncScheduledRouteBuilder) soawait .delay()never blocks the event loop.
- Tasks are regular FastAPI endpoints on plain old HTTP.
Dependsjust works!- All middlewares, telemetry, auth, debugging etc solutions for FastAPI work as is.
- Host task runners independent of GCP. If CloudTasks can reach the URL, it can invoke the task.
- Save money.
- Task invocation with GCP is free for first million, then costs $0.4/million. That's almost always cheaper than running a RabbitMQ/Redis/SQL backend for celery.
- Jobs cost $0.1 per job per month irrespective of invocations. 3 jobs are free. Either free or almost always cheaper than always running beat worker.
- If somehow, this cost ever becomes a concern, the
clientcan be overriden to call any gRPC server with a compatible API. Here's a trivial emulator implementation that we will use locally
- Autoscale.
- With a FaaS setup, your task workers can autoscale based on load.
- Most FaaS services have free tiers making it much cheaper than running a celery worker.
The package ships a py.typed marker (PEP 561), so mypy and pyright check everything you import from it.
.delay, .options, and .scheduler are attached to your endpoint at route registration time, which plain
function annotations can't express. Add one of the as_*_task decorators (identity functions at runtime) as the
innermost decorator and type checkers will see those methods with your endpoint's own signature:
from fastapi_gcp_tasks import as_delayed_task
@delayed_router.post("/{branch}/make_chili")
@as_delayed_task
async def make_chili(branch: str, recipe: Recipe) -> None: ...
make_chili.delay(branch="Scranton", recipe=Recipe(ingredients=["Ground beef", "Undercooked onions"])) # statically checked
make_chili.delay(branch="Scranton", recipe="oops") # type error
make_chili.options(countdown=1800).delay(branch="Scranton", recipe=Recipe(ingredients=["Ground beef", "Undercooked onions"]))Use as_async_delayed_task, as_scheduled_task, and as_async_scheduled_task for the other route builders.
Options accepted by .options(), .scheduler(), and task_default_options are typed via TypedDicts
(DelayOptions, SchedulerOptions, ...), so misspelled or wrongly-typed options are also caught statically.
Note: call .delay() and .schedule() with keyword arguments — the runtime only accepts keywords.
from fastapi_gcp_tasks import DelayedRouteBuilder
delayed_router = APIRouter(route_class=DelayedRouteBuilder(...))
class Recipe(BaseModel):
ingredients: List[str]
@delayed_router.post("/{branch}/make_chili")
async def make_chili(branch: str, recipe: Recipe):
# Do a ton of work here. The secret is to undercook the onions.
...
app.include_router(delayed_router)Now we can trigger the task with
make_chili.delay(branch="Scranton", recipe=Recipe(ingredients=["Ground beef", "Undercooked onions"]))If we want to trigger the task 30 minutes later
make_chili.options(countdown=1800).delay(...)from fastapi_gcp_tasks import ScheduledRouteBuilder
scheduled_router = APIRouter(route_class=ScheduledRouteBuilder(...))
class Recipe(BaseModel):
ingredients: List[str]
@scheduled_router.post("/pretzel_day")
async def pretzel_day(recipe: Recipe):
# Everyone gets one free soft pretzel
...
app.include_router(scheduled_router)
# Every Friday at 9AM in Scranton, it's Pretzel Day.
pretzel_day.scheduler(name="pretzel-day-9AM-scranton", schedule="0 9 * * 5", time_zone="America/New_York").schedule(
recipe=Recipe(ingredients=["Sweet glaze", "Cinnamon sugar"]))DelayedRouteBuilder's .delay() makes a blocking gRPC call. Called from an async endpoint, it stalls the
event loop until Cloud Tasks responds. AsyncDelayedRouteBuilder uses the native
CloudTasksAsyncClient instead, so triggering a task is a proper coroutine:
from fastapi_gcp_tasks import AsyncDelayedRouteBuilder
async_delayed_router = APIRouter(route_class=AsyncDelayedRouteBuilder(...))
@async_delayed_router.post("/{branch}/make_chili")
async def make_chili(branch: str, recipe: Recipe):
...
app.include_router(async_delayed_router)
# In an async context (endpoint, lifespan, etc):
await make_chili.delay(branch="Scranton", recipe=Recipe(ingredients=["Ground beef", "Undercooked onions"]))
await make_chili.options(countdown=1800).delay(...)Similarly, AsyncScheduledRouteBuilder provides awaitable .schedule() and .delete() — useful when
creating Cloud Scheduler jobs dynamically from request handlers. Since it can't run at module import
time like the sync version, await it from a lifespan (or a handler):
from contextlib import asynccontextmanager
from fastapi_gcp_tasks import AsyncScheduledRouteBuilder
async_scheduled_router = APIRouter(route_class=AsyncScheduledRouteBuilder(...))
@async_scheduled_router.post("/pretzel_day")
async def pretzel_day(recipe: Recipe):
...
@asynccontextmanager
async def lifespan(app: FastAPI):
await pretzel_day.scheduler(name="pretzel-day-9AM-scranton", schedule="0 9 * * 5", time_zone="America/New_York").schedule(
recipe=Recipe(ingredients=["Sweet glaze", "Cinnamon sugar"])
)
yieldThings to know about the async builders:
-
The client is created lazily. grpc.aio clients bind to the event loop that is running when they are constructed, so the builder resolves its client on the first awaited call, inside your app's loop.
clientaccepts a client instance, a zero-argument factory returning one, orNone(default credentials). If your client needs custom construction — like the local emulator — pass a factory:client=lambda: async_emulator_client(). -
The queue is not auto-created by default. Unlike
DelayedRouteBuilder,auto_create_queuedefaults toFalseso no unexpected RPC runs inside a request handler. Either ensure the queue from your lifespan with theensure_queue_asyncutil (recommended), or opt in withauto_create_queue=Trueto ensure it lazily on the first.delay():from contextlib import asynccontextmanager from fastapi_gcp_tasks.utils import ensure_queue_async @asynccontextmanager async def lifespan(app: FastAPI): await ensure_queue_async(client=my_async_client, path=MY_QUEUE_PATH) yield
-
Hooks are unchanged. The same (synchronous)
pre_create_hooks work with both builders — they are pure in-memory mutations of the request proto and run inline on the event loop, so they must not block.
Cloud Tasks allows us to schedule a HTTP request in the future.
FastAPI makes us define complete schema and params for an HTTP endpoint.
Cloud Scheduler allows us to schedule recurring HTTP requests in the future.
FastAPI Cloud Tasks works by putting the three together:
- GCP's Cloud Tasks + FastAPI = Partial replacement for celery's async delayed tasks.
- GCP's Cloud Scheduler + FastAPI = Replacement for celery beat.
- FastAPI Cloud Tasks + Cloud Run = Autoscaled delayed tasks.
Pre-requisites:
pip install fastapi-gcp-tasks- Install cloud-tasks-emulator
- Alternatively install ngrok and forward the server's port
Start running the emulator in a terminal
cloud-tasks-emulatorStart running the task runner on port 8000 so that it is accessible from cloud tasks.
uvicorn examples.simple.main:app --reload --port 8000In another terminal, trigger the task with curl
curl http://localhost:8000/trigger
Check the logs on the server, you should see
WARNING: Hello task ran with payload: Triggered task
Important bits of code:
# complete file: examples/simple/main.py
# For local, we connect to the emulator client
client = None
if IS_LOCAL:
client = emulator_client()
# Construct our DelayedRoute class with all relevant settings
# This can be done once across the entire project
DelayedRoute = DelayedRouteBuilder(
client=client,
base_url="http://localhost:8000"
queue_path=queue_path(
project="gcp-project-id",
location="us-central1",
queue="test-queue",
),
)
# Override the route_class so that we can add .delay method to the endpoints and know their complete URL
delayed_router = APIRouter(route_class=DelayedRoute, prefix="/delayed")
class Payload(BaseModel):
message: str
@delayed_router.post("/hello")
async def hello(p: Payload = Payload(message="Default")):
logger.warning(f"Hello task ran with payload: {p.message}")
# Define our app and add trigger to it.
app = FastAPI()
@app.get("/trigger")
async def trigger():
# Trigger the task
hello.delay(p=Payload(message="Triggered task"))
return {"message": "Hello task triggered"}
app.include_router(delayed_router)Note: You can read complete working source code of the above example in examples/simple/main.py
In the real world you'd have a separate process for task runner and actual task.
Running on Cloud Run with authentication needs us to supply an OIDC token. To do that we can use a hook.
Pre-requisites:
- Create a task queue. Copy the project id, location and queue name.
- Deploy the worker as a service on Cloud Run and copy it's URL.
- Create a service account in cloud IAM and add
Cloud Run Invokerrole to it.
# URL of the Cloud Run service
base_url = "https://hello-randomchars-el.a.run.app"
DelayedRoute = DelayedRouteBuilder(
base_url=base_url,
# Task queue, same as above.
queue_path=queue_path(...),
pre_create_hook=oidc_task_hook(
token=tasks_v2.OidcToken(
# Service account that you created
service_account_email="fastapi-gcp-tasks@gcp-project-id.iam.gserviceaccount.com",
audience=base_url,
),
),
)Check the fleshed out example at examples/full/tasks.py
If you're not running on CloudRun and want to an OAuth Token instead, you can use the oauth_task_hook instead.
Check fastapi_cloud_tasks/hooks.py to get the hang od hooks and how you can use them.
Usage:
DelayedRoute = DelayedRouteBuilder(...)
delayed_router = APIRouter(route_class=DelayedRoute)
@delayed_router.get("/simple_task")
def simple_task():
return {}-
base_url- The URL of your worker FastAPI service. -
queue_path- Full path of the Cloud Tasks queue. (Hint: use the util functionqueue_path) -
task_create_timeout- How long should we wait before giving up on creating cloud task. -
pre_create_hook- If you need to edit theCreateTaskRequestbefore sending it to Cloud Tasks (eg: Auth for Cloud Run), you can do that with this hook. See hooks section below for more. -
client- If you need to override the Cloud Tasks client, pass the client here. (eg: changing credentials, transport etc)
Usage:
@delayed_router.get("/simple_task")
@task_default_options(...)
def simple_task():
return {}All options from above can be passed as kwargs to the decorator, including client — the decorator is overloaded so a CloudTasksClient or a CloudTasksAsyncClient both typecheck with the matching route builder.
Additional options:
countdown- Seconds in the future to schedule the task.task_id- named task id for deduplication. (One task id will only be queued once.)
Example:
# Trigger after 5 minutes
@delayed_router.get("/simple_task")
@task_default_options(countdown=300)
def simple_task():
return {}Usage:
simple_task.options(...).delay()All options from above can be overwritten per call (including DelayedRouteBuilder options like base_url) with kwargs to the options function before calling delay.
Example:
# Trigger after 2 minutes
simple_task.options(countdown=120).delay()Same options as DelayedRouteBuilder, with two differences:
client- ACloudTasksAsyncClient, a zero-argument factory returning one, orNone. Resolved lazily on the first awaited.delay()because grpc.aio clients bind to the running event loop.auto_create_queue- Defaults toFalse(the sync builder defaults toTrue). WhenTrue, the queue is ensured lazily on the first.delay(). Prefer callingensure_queue_asyncfrom your lifespan instead.
Usage:
ScheduledRoute = ScheduledRouteBuilder(...)
scheduled_router = APIRouter(route_class=ScheduledRoute)
@scheduled_router.get("/simple_scheduled_task")
def simple_scheduled_task():
return {}
simple_scheduled_task.scheduler(name="simple_scheduled_task", schedule="* * * * *").schedule()Same options as ScheduledRouteBuilder, except client accepts a CloudSchedulerAsyncClient, a
zero-argument factory returning one, or None (resolved lazily, as above). .schedule() and .delete()
are coroutines — await them from a lifespan or a request handler.
We might need to override things in the task being sent to Cloud Tasks. The pre_create_hook allows us to do that.
Some hooks are included in the library.
oidc_delayed_hook/oidc_scheduled_hook- Used to pass OIDC token (for Cloud Run etc).deadline_delayed_hook/deadline_scheduled_hook- Used to change the timeout for the worker of a task. (PS: this deadline is decided by the sender to the queue and not the worker)chained_hook- If you need to chain multiple hooks together, you can do that withchained_hook(hook1, hook2)
@delayed_router.post("/fail_twice", dependencies=[Depends(max_retries(2))])
async def fail_twice():
raise Exception("nooo")@delayed_router.get("/my_task")
async def my_task(ct_headers: CloudTasksHeaders = Depends()):
print(ct_headers.queue_name)Check the file fastapi_cloud_tasks/dependencies.py for details.
- uv
- Docker (for the Cloud Tasks emulator)
docker compose up -d # start emulator
make test # run tests
docker compose down # stop emulatormake lint # check
make format # auto-fix- Run
make lintandmake formatbefore raising a PR. - Add examples and/or tests for new features.
- If the change is massive, open an issue to discuss it before writing code.
This project is licensed under the terms of the MIT license. This project was forked from fastapi-gcp-tasks under the MIT license. All changes made to the original project are also licensed under the MIT license.
This project is neither affiliated with, nor sponsored by Google.