Typed protocols

Typed views of task endpoints and the option TypedDicts. See the Type safety guide for usage.

fastapi_gcp_tasks.protocols

TaskDefaultOptions

Bases: TypedDict

Options accepted by the task_default_options decorator (shared by sync and async delayed routes).

client is the only delay option not accepted here, because its type differs between the sync and async route builders.

DelayOptions

Bases: TaskDefaultOptions

Per-call overrides accepted by .options() on a delayed task endpoint.

AsyncDelayOptions

Bases: TaskDefaultOptions

Per-call overrides accepted by .options() on an async delayed task endpoint.

SchedulerOptions

Bases: TypedDict

Per-call overrides accepted by .scheduler() on a scheduled task endpoint.

AsyncSchedulerOptions

Bases: TypedDict

Per-call overrides accepted by .scheduler() on an async scheduled task endpoint.

DelayerHandle

Bases: Protocol[P]

A configured delayer bound to an endpoint's signature.

delay

delay(*args: args, **kwargs: kwargs) -> tasks_v2.Task

Create the task on Cloud Tasks. Call with keyword arguments matching the endpoint.

Source code in fastapi_gcp_tasks/protocols.py

def delay(self, *args: P.args, **kwargs: P.kwargs) -> tasks_v2.Task:
    """Create the task on Cloud Tasks. Call with keyword arguments matching the endpoint."""
    ...

AsyncDelayerHandle

Bases: Protocol[P]

A configured async delayer bound to an endpoint's signature.

delay async

delay(*args: args, **kwargs: kwargs) -> tasks_v2.Task

Create the task on Cloud Tasks. Call with keyword arguments matching the endpoint.

Source code in fastapi_gcp_tasks/protocols.py

async def delay(self, *args: P.args, **kwargs: P.kwargs) -> tasks_v2.Task:
    """Create the task on Cloud Tasks. Call with keyword arguments matching the endpoint."""
    ...

SchedulerHandle

Bases: Protocol[P]

A configured scheduler bound to an endpoint's signature.

schedule

schedule(*args: args, **kwargs: kwargs) -> None

Create (or update) the job on Cloud Scheduler. Call with keyword arguments matching the endpoint.

Source code in fastapi_gcp_tasks/protocols.py

def schedule(self, *args: P.args, **kwargs: P.kwargs) -> None:
    """Create (or update) the job on Cloud Scheduler. Call with keyword arguments matching the endpoint."""
    ...

delete

delete() -> bool | Exception

Delete the job from Cloud Scheduler if it exists.

Source code in fastapi_gcp_tasks/protocols.py

def delete(self) -> bool | Exception:
    """Delete the job from Cloud Scheduler if it exists."""
    ...

AsyncSchedulerHandle

Bases: Protocol[P]

A configured async scheduler bound to an endpoint's signature.

schedule async

schedule(*args: args, **kwargs: kwargs) -> None

Create (or update) the job on Cloud Scheduler. Call with keyword arguments matching the endpoint.

Source code in fastapi_gcp_tasks/protocols.py

async def schedule(self, *args: P.args, **kwargs: P.kwargs) -> None:
    """Create (or update) the job on Cloud Scheduler. Call with keyword arguments matching the endpoint."""
    ...

delete async

delete() -> bool | Exception

Delete the job from Cloud Scheduler if it exists.

Source code in fastapi_gcp_tasks/protocols.py

async def delete(self) -> bool | Exception:
    """Delete the job from Cloud Scheduler if it exists."""
    ...

DelayedTask

Bases: Protocol[P, R]

Typed view of an endpoint registered on a DelayedRouteBuilder route.

The .delay and .options attributes are attached at route registration time, so plain function annotations can't see them. Apply as_delayed_task as the innermost decorator to give the endpoint this type.

__call__

__call__(*args: args, **kwargs: kwargs) -> R

Call the endpoint directly (as FastAPI does).

Source code in fastapi_gcp_tasks/protocols.py

def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Call the endpoint directly (as FastAPI does)."""
    ...

delay

delay(*args: args, **kwargs: kwargs) -> tasks_v2.Task

Trigger the task on Cloud Tasks. Call with keyword arguments matching the endpoint.

Source code in fastapi_gcp_tasks/protocols.py

def delay(self, *args: P.args, **kwargs: P.kwargs) -> tasks_v2.Task:
    """Trigger the task on Cloud Tasks. Call with keyword arguments matching the endpoint."""
    ...

options

options(
    **options: Unpack[DelayOptions],
) -> DelayerHandle[P]

Override task options for a single trigger, e.g. fn.options(countdown=5).delay(...).

Source code in fastapi_gcp_tasks/protocols.py

def options(self, **options: Unpack[DelayOptions]) -> DelayerHandle[P]:
    """Override task options for a single trigger, e.g. ``fn.options(countdown=5).delay(...)``."""
    ...

AsyncDelayedTask

Bases: Protocol[P, R]

Typed view of an endpoint registered on an AsyncDelayedRouteBuilder route.

Apply as_async_delayed_task as the innermost decorator to give the endpoint this type.

__call__

__call__(*args: args, **kwargs: kwargs) -> R

Call the endpoint directly (as FastAPI does).

Source code in fastapi_gcp_tasks/protocols.py

def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Call the endpoint directly (as FastAPI does)."""
    ...

delay async

delay(*args: args, **kwargs: kwargs) -> tasks_v2.Task

Trigger the task on Cloud Tasks. Call with keyword arguments matching the endpoint.

Source code in fastapi_gcp_tasks/protocols.py

async def delay(self, *args: P.args, **kwargs: P.kwargs) -> tasks_v2.Task:
    """Trigger the task on Cloud Tasks. Call with keyword arguments matching the endpoint."""
    ...

options

options(
    **options: Unpack[AsyncDelayOptions],
) -> AsyncDelayerHandle[P]

Override task options for a single trigger, e.g. await fn.options(countdown=5).delay(...).

Source code in fastapi_gcp_tasks/protocols.py

def options(self, **options: Unpack[AsyncDelayOptions]) -> AsyncDelayerHandle[P]:
    """Override task options for a single trigger, e.g. ``await fn.options(countdown=5).delay(...)``."""
    ...

ScheduledTask

Bases: Protocol[P, R]

Typed view of an endpoint registered on a ScheduledRouteBuilder route.

Apply as_scheduled_task as the innermost decorator to give the endpoint this type.

__call__

__call__(*args: args, **kwargs: kwargs) -> R

Call the endpoint directly (as FastAPI does).

Source code in fastapi_gcp_tasks/protocols.py

def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Call the endpoint directly (as FastAPI does)."""
    ...

scheduler

scheduler(
    *,
    name: str,
    schedule: str,
    **options: Unpack[SchedulerOptions],
) -> SchedulerHandle[P]

Configure a Cloud Scheduler job for this endpoint; call .schedule(...) on the result.

Source code in fastapi_gcp_tasks/protocols.py

def scheduler(self, *, name: str, schedule: str, **options: Unpack[SchedulerOptions]) -> SchedulerHandle[P]:
    """Configure a Cloud Scheduler job for this endpoint; call ``.schedule(...)`` on the result."""
    ...

AsyncScheduledTask

Bases: Protocol[P, R]

Typed view of an endpoint registered on an AsyncScheduledRouteBuilder route.

Apply as_async_scheduled_task as the innermost decorator to give the endpoint this type.

__call__

__call__(*args: args, **kwargs: kwargs) -> R

Call the endpoint directly (as FastAPI does).

Source code in fastapi_gcp_tasks/protocols.py

def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Call the endpoint directly (as FastAPI does)."""
    ...

scheduler

scheduler(
    *,
    name: str,
    schedule: str,
    **options: Unpack[AsyncSchedulerOptions],
) -> AsyncSchedulerHandle[P]

Configure a Cloud Scheduler job for this endpoint; await .schedule(...) on the result.

Source code in fastapi_gcp_tasks/protocols.py

def scheduler(
    self, *, name: str, schedule: str, **options: Unpack[AsyncSchedulerOptions]
) -> AsyncSchedulerHandle[P]:
    """Configure a Cloud Scheduler job for this endpoint; await ``.schedule(...)`` on the result."""
    ...

ensure_known_options

ensure_known_options(
    options: Mapping[str, object], allowed: type
) -> None

Raise TypeError if options contains keys not declared on the allowed TypedDict.

Type checkers already reject unknown keys statically; this keeps unchecked callers failing fast at runtime instead of silently dropping options.

Source code in fastapi_gcp_tasks/protocols.py

def ensure_known_options(options: Mapping[str, object], allowed: type) -> None:
    """
    Raise TypeError if ``options`` contains keys not declared on the ``allowed`` TypedDict.

    Type checkers already reject unknown keys statically; this keeps unchecked
    callers failing fast at runtime instead of silently dropping options.
    """
    unexpected = set(options) - (allowed.__required_keys__ | allowed.__optional_keys__)  # type: ignore[attr-defined]
    if unexpected:
        raise TypeError(f"Unknown option(s) for {allowed.__name__}: {', '.join(sorted(unexpected))}")

as_delayed_task

as_delayed_task(fn: Callable[P, R]) -> DelayedTask[P, R]

Identity cast that types an endpoint as a DelayedTask.

Apply as the innermost decorator (below the router decorator) so type checkers see .delay and .options with the endpoint's own signature:

@delayed_router.post("/hello")
@as_delayed_task
def hello(p: Payload) -> None: ...


hello.delay(p=Payload(...))  # statically checked

Source code in fastapi_gcp_tasks/protocols.py

def as_delayed_task(fn: Callable[P, R]) -> DelayedTask[P, R]:
    """
    Identity cast that types an endpoint as a DelayedTask.

    Apply as the innermost decorator (below the router decorator) so type
    checkers see ``.delay`` and ``.options`` with the endpoint's own signature:

    ```
    @delayed_router.post("/hello")
    @as_delayed_task
    def hello(p: Payload) -> None: ...


    hello.delay(p=Payload(...))  # statically checked
    ```
    """
    return cast(DelayedTask[P, R], fn)

as_async_delayed_task

as_async_delayed_task(
    fn: Callable[P, R],
) -> AsyncDelayedTask[P, R]

Identity cast that types an endpoint as an AsyncDelayedTask. See as_delayed_task.

Source code in fastapi_gcp_tasks/protocols.py

def as_async_delayed_task(fn: Callable[P, R]) -> AsyncDelayedTask[P, R]:
    """Identity cast that types an endpoint as an AsyncDelayedTask. See ``as_delayed_task``."""
    return cast(AsyncDelayedTask[P, R], fn)

as_scheduled_task

as_scheduled_task(
    fn: Callable[P, R],
) -> ScheduledTask[P, R]

Identity cast that types an endpoint as a ScheduledTask. See as_delayed_task.

Source code in fastapi_gcp_tasks/protocols.py

def as_scheduled_task(fn: Callable[P, R]) -> ScheduledTask[P, R]:
    """Identity cast that types an endpoint as a ScheduledTask. See ``as_delayed_task``."""
    return cast(ScheduledTask[P, R], fn)

as_async_scheduled_task

as_async_scheduled_task(
    fn: Callable[P, R],
) -> AsyncScheduledTask[P, R]

Identity cast that types an endpoint as an AsyncScheduledTask. See as_delayed_task.

Source code in fastapi_gcp_tasks/protocols.py

def as_async_scheduled_task(fn: Callable[P, R]) -> AsyncScheduledTask[P, R]:
    """Identity cast that types an endpoint as an AsyncScheduledTask. See ``as_delayed_task``."""
    return cast(AsyncScheduledTask[P, R], fn)