Skip to content

Hooks

fastapi_gcp_tasks.hooks

noop_hook 

noop_hook(request: RequestT) -> RequestT

Inspired by https://github.com/kelseyhightower/nocode.

Source code in fastapi_gcp_tasks/hooks.py 
15
16
17
def noop_hook(request: RequestT) -> RequestT:
    """Inspired by https://github.com/kelseyhightower/nocode."""
    return request

chained_hook 

chained_hook(
    *hooks: Callable[[RequestT], RequestT],
) -> Callable[[RequestT], RequestT]

Call all hooks sequentially with the result from the previous hook.

Source code in fastapi_gcp_tasks/hooks.py 
20
21
22
23
24
25
26
27
28
def chained_hook(*hooks: Callable[[RequestT], RequestT]) -> Callable[[RequestT], RequestT]:
    """Call all hooks sequentially with the result from the previous hook."""

    def chain(request: RequestT) -> RequestT:
        for hook in hooks:
            request = hook(request)
        return request

    return chain

oidc_scheduled_hook 

oidc_scheduled_hook(token: OidcToken) -> ScheduledHook

Returns a hook for ScheduledRouteBuilder to add OIDC token to all requests.

https://cloud.google.com/scheduler/docs/reference/rpc/google.cloud.scheduler.v1#google.cloud.scheduler.v1.HttpTarget

Source code in fastapi_gcp_tasks/hooks.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def oidc_scheduled_hook(token: scheduler_v1.OidcToken) -> ScheduledHook:
    """
    Returns a hook for ScheduledRouteBuilder to add OIDC token to all requests.

    https://cloud.google.com/scheduler/docs/reference/rpc/google.cloud.scheduler.v1#google.cloud.scheduler.v1.HttpTarget
    """

    def add_token(
        request: scheduler_v1.CreateJobRequest,
    ) -> scheduler_v1.CreateJobRequest:
        request.job.http_target.oidc_token = token
        return request

    return add_token

oidc_delayed_hook 

oidc_delayed_hook(token: OidcToken) -> DelayedTaskHook

Returns a hook for DelayedRouteBuilder to add OIDC token to all requests.

https://cloud.google.com/tasks/docs/reference/rpc/google.cloud.tasks.v2#google.cloud.tasks.v2.HttpRequest

Source code in fastapi_gcp_tasks/hooks.py 
47
48
49
50
51
52
53
54
55
56
57
58
def oidc_delayed_hook(token: tasks_v2.OidcToken) -> DelayedTaskHook:
    """
    Returns a hook for DelayedRouteBuilder to add OIDC token to all requests.

    https://cloud.google.com/tasks/docs/reference/rpc/google.cloud.tasks.v2#google.cloud.tasks.v2.HttpRequest
    """

    def add_token(request: tasks_v2.CreateTaskRequest) -> tasks_v2.CreateTaskRequest:
        request.task.http_request.oidc_token = token
        return request

    return add_token

oauth_scheduled_hook 

oauth_scheduled_hook(token: OAuthToken) -> ScheduledHook

Returns a hook for ScheduledRouteBuilder to add OAuth token to all requests.

https://cloud.google.com/scheduler/docs/reference/rpc/google.cloud.scheduler.v1#google.cloud.scheduler.v1.HttpTarget

Source code in fastapi_gcp_tasks/hooks.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def oauth_scheduled_hook(token: scheduler_v1.OAuthToken) -> ScheduledHook:
    """
    Returns a hook for ScheduledRouteBuilder to add OAuth token to all requests.

    https://cloud.google.com/scheduler/docs/reference/rpc/google.cloud.scheduler.v1#google.cloud.scheduler.v1.HttpTarget
    """

    def add_token(
        request: scheduler_v1.CreateJobRequest,
    ) -> scheduler_v1.CreateJobRequest:
        request.job.http_target.oauth_token = token
        return request

    return add_token

oauth_delayed_hook 

oauth_delayed_hook(token: OAuthToken) -> DelayedTaskHook

Returns a hook for DelayedRouteBuilder to add OAuth token to all requests.

https://cloud.google.com/tasks/docs/reference/rpc/google.cloud.tasks.v2#google.cloud.tasks.v2.HttpRequest

Source code in fastapi_gcp_tasks/hooks.py 
77
78
79
80
81
82
83
84
85
86
87
88
def oauth_delayed_hook(token: tasks_v2.OAuthToken) -> DelayedTaskHook:
    """
    Returns a hook for DelayedRouteBuilder to add OAuth token to all requests.

    https://cloud.google.com/tasks/docs/reference/rpc/google.cloud.tasks.v2#google.cloud.tasks.v2.HttpRequest
    """

    def add_token(request: tasks_v2.CreateTaskRequest) -> tasks_v2.CreateTaskRequest:
        request.task.http_request.oauth_token = token
        return request

    return add_token

deadline_scheduled_hook 

deadline_scheduled_hook(
    duration: Duration,
) -> ScheduledHook

Returns a hook for ScheduledRouteBuilder to set Deadline for job execution.

https://cloud.google.com/scheduler/docs/reference/rpc/google.cloud.scheduler.v1#google.cloud.scheduler.v1.Job

Source code in fastapi_gcp_tasks/hooks.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def deadline_scheduled_hook(duration: duration_pb2.Duration) -> ScheduledHook:
    """
    Returns a hook for ScheduledRouteBuilder to set Deadline for job execution.

    https://cloud.google.com/scheduler/docs/reference/rpc/google.cloud.scheduler.v1#google.cloud.scheduler.v1.Job
    """

    def deadline(
        request: scheduler_v1.CreateJobRequest,
    ) -> scheduler_v1.CreateJobRequest:
        request.job.attempt_deadline = duration
        return request

    return deadline

deadline_delayed_hook 

deadline_delayed_hook(
    duration: Duration,
) -> DelayedTaskHook

Returns a hook for DelayedRouteBuilder to set Deadline for task execution.

https://cloud.google.com/tasks/docs/reference/rpc/google.cloud.tasks.v2#google.cloud.tasks.v2.Task

Source code in fastapi_gcp_tasks/hooks.py 
107
108
109
110
111
112
113
114
115
116
117
118
def deadline_delayed_hook(duration: duration_pb2.Duration) -> DelayedTaskHook:
    """
    Returns a hook for DelayedRouteBuilder to set Deadline for task execution.

    https://cloud.google.com/tasks/docs/reference/rpc/google.cloud.tasks.v2#google.cloud.tasks.v2.Task
    """

    def deadline(request: tasks_v2.CreateTaskRequest) -> tasks_v2.CreateTaskRequest:
        request.task.dispatch_deadline = duration
        return request

    return deadline