fastapi-gcp-tasks

KO | EN

gitlite — search

#hacktoberfest #python #python3 #typescript #graphql #java #golang #matlab #machine-learning #gui #react #docker

fastapi-gcp-tasks

★ 24 Open GitHub ↗

Distributed task orchestration with FastAPI and GCP Cloud Run, Tasks and Scheduler. Replacement for Celery for serverless

// repository documentation

Was this content helpful?

★ 0 (0 ratings)

# FastAPI Cloud Tasks

Strongly typed background tasks with FastAPI and Google Cloud Run, Tasks and Scheduler. This is a fork of [fastapi-gcp-tasks](https://github.com/adori/fastapi-cloud-tasks), updated with new features and bug fixes.

**📚 Documentation: [simplifyjobs.github.io/fastapi-gcp-tasks](https://simplifyjobs.github.io/fastapi-gcp-tasks/)**

```mermaid
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

```

## Installation

```
pip install fastapi-gcp-tasks
```

## Key features

- 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](#type-safety).
- Familiar and simple public API
  - `.delay` method that takes same arguments as the task.
  - `.scheduler` method to create recurring job.
  - Async variants (`AsyncDelayedRouteBuilder` / `AsyncScheduledRouteBuilder`) so `await .delay()` never blocks the event loop.
- Tasks are regular FastAPI endpoints on plain old HTTP.
  - `Depends` just 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](https://cloud.google.com/tasks/pricing).
    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.](https://cloud.google.com/scheduler#pricing)
    Either free or almost always cheaper than always running beat worker.
  - If somehow, this cost ever becomes a concern, the `client` can be overriden to call any gRPC server with a compatible API.
    [Here's a trivial emulator implementation that we will use locally](https://github.com/aertje/cloud-tasks-emulator)
- 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.

## Type safety

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:

```python
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 `TypedDict`s
(`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.

## How it works

### Delayed job

```python
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

```python
make_chili.delay(branch="Scranton", recipe=Recipe(ingredients=["Ground beef", "Undercooked onions"]))
```

If we want to trigger the task 30 minutes later

```python
make_chili.options(countdown=1800).delay(...)
```

### Scheduled Task

```python
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"]))
```

### Async usage

`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:

```python
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):

```python
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"])
    )
    yield
```

Things 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. `client`
  accepts a client instance, a zero-argument factory returning one, or `None` (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_queue` defaults to
  `False` so no unexpected RPC runs inside a request handler. Either ensure the queue from your lifespan with
  the `ensure_queue_async` util (recommended), or opt in with `auto_create_queue=True` to ensure it lazily on
  the first `.delay()`:

```python
  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_hook`s 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.

## Concept

[`Cloud Tasks`](https://cloud.google.com/tasks) allows us to schedule a HTTP request in the future.

[FastAPI](https://fastapi.tiangolo.com/tutorial/body/) makes us define complete schema and params for an HTTP endpoint.

[`Cloud Scheduler`](https://cloud.google.com/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.

## Running

### Local

Pre-requisites:
- `pip install fastapi-gcp-tasks`
- Install [cloud-tasks-emulator](https://github.com/aertje/cloud-tasks-emulator)
  - Alternatively install ngrok and forward the server's port

Start running the emulator in a terminal
```sh
cloud-tasks-emulator
```

Start running the task runner on port 8000 so that it is accessible from cloud tasks.

```sh
uvicorn examples.simple.main:app --reload --port 8000
```

In 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:

```python
# 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,
    callback_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`](examples/simple/main.py)

In the real world you'd have a separate process for task runner and actual task.

### Deployed environment / Cloud Run

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 Invoker` role to it.

```python
# URL of the Cloud Run service
service_url = "https://hello-randomchars-el.a.run.app"

DelayedRoute = DelayedRouteBuilder(
    callback_base_url=service_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=service_url,
        ),
    ),
)
```

Check the fleshed out example at [`examples/full/tasks.py`](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](fastapi_gcp_tasks/hooks.py) to get the hang od hooks and how you can use them.

## Configuration

### DelayedRouteBuilder

Usage:

```python
DelayedRoute = DelayedRouteBuilder(...)
delayed_router = APIRouter(route_class=DelayedRoute)

@delayed_router.get("/simple_task")
def simple_task():
    return {}
```

- `callback_base_url` - The externally reachable URL prefix to which the task route's own path is appended.
  Include any prefix added later by an outer `include_router()` call.

- `base_url` - Legacy alias for `callback_base_url`. Pass only one of the two.

- `queue_path` - Full path of the Cloud Tasks queue. (Hint: use the util function `queue_path`)

- `task_create_timeout` - How long should we wait before giving up on creating cloud task.

- `pre_create_hook` - If you need to edit the `CreateTaskRequest` before 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)

#### Task level default options

Usage:

```python
@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:

```python
# Trigger after 5 minutes
@delayed_router.get("/simple_task")
@task_default_options(countdown=300)
def simple_task():
    return {}
```

#### Delayer Options

Usage:

```python
simple_task.options(...).delay()
```

All options from above can be overwritten per call (including `callback_base_url`) with kwargs to the `options`
function before calling delay.

Example:

```python
# Trigger after 2 minutes
simple_task.options(countdown=120).delay()
```

#### Routers mounted under an outer prefix

FastAPI 0.137 and later preserve included routers instead of cloning their routes. A route class therefore knows the
path declared on its own router, but it cannot unambiguously infer a prefix added later—especially because the same
router can be included more than once. Put that externally visible prefix in `callback_base_url`:

```python
service_url = "https://worker.example.com"

DelayedRoute = DelayedRouteBuilder(
    callback_base_url=f"{service_url}/tasks",
    queue_path=queue_path(...),
    pre_create_hook=oidc_delayed_hook(
        token=tasks_v2.OidcToken(
            service_account_email="tasks@example.iam.gserviceaccount.com",
            audience=service_url,
        ),
    ),
)

task_router = APIRouter(route_class=DelayedRoute, prefix="/auth")

@task_router.post("/confirm/{user_id}")
def confirm_user(user_id: str): ...

app.include_router(task_router, prefix="/tasks")
```

`confirm_user.delay(user_id="42")` now targets
`https://worker.example.com/tasks/auth/confirm/42`. The OIDC audience remains the service URL; it does not need the
callback path prefix.

### AsyncDelayedRouteBuilder

Same options as `DelayedRouteBuilder`, with two differences:

- `client` - A `CloudTasksAsyncClient`, a zero-argument factory returning one, or `None`. Resolved lazily on
  the first awaited `.delay()` because grpc.aio clients bind to the running event loop.
- `auto_create_queue` - Defaults to `False` (the sync builder defaults to `True`). When `True`, the queue is
  ensured lazily on the first `.delay()`. Prefer calling `ensure_queue_async` from your lifespan instead.

The same `callback_base_url` rule applies when an async delayed router is mounted under an outer prefix.

### ScheduledRouteBuilder

Usage:

```python
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()
```

`callback_base_url` has the same meaning as it does for delayed routes: it is the externally reachable URL prefix to
which the scheduled route's own path is appended. Include prefixes added by an outer `include_router()` call:

```python
service_url = "https://worker.example.com"

ScheduledRoute = ScheduledRouteBuilder(
    callback_base_url=f"{service_url}/tasks",
    location_path="projects/my-project/locations/us-central1",
)
scheduled_router = APIRouter(route_class=ScheduledRoute, prefix="/maintenance")

@scheduled_router.post("/sweep")
def sweep(): ...

app.include_router(scheduled_router, prefix="/tasks")
```

The scheduled callback is `https://worker.example.com/tasks/maintenance/sweep`. As with delayed tasks, keep a Cloud
Run OIDC audience set to `service_url`, without the callback path prefix. `base_url` remains a legacy alias for
`callback_base_url`; pass only one.

### AsyncScheduledRouteBuilder

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. The same nested-router `callback_base_url` rule
applies to async scheduled routes.

## Hooks

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 with `chained_hook(hook1, hook2)`

## Helper dependencies

### max_retries

```python
@delayed_router.post("/fail_twice", dependencies=[Depends(max_retries(2))])
async def fail_twice():
    raise Exception("nooo")
```

### CloudTasksHeaders

```python
@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](fastapi_gcp_tasks/dependencies.py) for details.

## Development

### Prerequisites

- [uv](https://docs.astral.sh/uv/)
- Docker (for the Cloud Tasks emulator)

### Running tests

```sh
docker compose up -d       # start emulator
make test                  # run tests
docker compose down        # stop emulator
```

### Linting & formatting

```sh
make lint                  # check
make format                # auto-fix
```

## Contributing

- Run `make lint` and `make format` before 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.

## License

This project is licensed under the terms of the MIT license. This project was forked from [fastapi-gcp-tasks](https://github.com/Adori/fastapi-cloud-tasks) under the MIT license. All changes made to the original project are also licensed under the MIT license.

## Disclaimer

This project is neither affiliated with, nor sponsored by Google.

fastapi-gcp-tasks

Explore Similar Repositories

Was this content helpful?

Recent Feedback

Download README