CloudTasksQueue

A GCP Cloud Tasks Queue.

Like all Resources, this class configures itself across multiple Environments.

For more information see the official documentation.

Example Usage

1
import launchflow as lf
2
3
# Automatically creates / connects to a Cloud Tasks Queue in your GCP project
4
queue = lf.gcp.CloudTasksQueue("my-queue")
5
6
queue.enqueue("https://example.com/endpoint", json={"key": "value"})

Optional Arguments

  • location (Optional[str]): The location of the queue. If None, the GCP project's default region is used. Defaults to None.
1
import launchflow as lf
2
3
queue = lf.gcp.CloudTasksQueue("my-queue", location="us-central1")

Utility Methods

enqueue

1
CloudTasksQueue.enqueue(url: str, method: str = "POST", headers: Optional[Dict[str, str]] = None, body: Optional[bytes] = None, json: Optional[Dict] = None, oauth_token: Optional[str] = None, oidc_token: Optional[str] = None, schedule_time: Optional[datetime.datetime] = None) -> "tasks.Task"

Enqueue a task in the Cloud Tasks queue.

Args:

  • url: The url the task will call.
  • method: The HTTP method to use. Defaults to "POST".
  • headers: A dictionary of headers to include in the request.
  • body: The body of the request. Only one of body or json can be provided.
  • json: A dictionary to be serialized as JSON and sent as the body of the request. Only one of body or json can be provided.
  • oauth_token: An OAuth token to include in the request.
  • oidc_token: An OIDC token to include in the request.
  • schedule_time: Optional[datetime.datetime] = None,

Example usage:

1
import launchflow as lf
2
3
queue = lf.gcp.CloudTasksQueue("my-queue")
4
5
queue.enqueue("https://example.com/endpoint", json={"key": "value"})

enqueue_async

1
async CloudTasksQueue.enqueue_async(url: str, method: str = "POST", headers: Optional[Dict[str, str]] = None, body: Optional[bytes] = None, json: Optional[Dict] = None, oauth_token: Optional[str] = None, oidc_token: Optional[str] = None, schedule_time: Optional[datetime.datetime] = None) -> "tasks.Task"

Enqueue a task in the Cloud Tasks queue.

Args:

  • url: The url the task will call.
  • method: The HTTP method to use. Defaults to "POST".
  • headers: A dictionary of headers to include in the request.
  • body: The body of the request. Only one of body or json can be provided.
  • json: A dictionary to be serialized as JSON and sent as the body of the request. Only one of body or json can be provided.
  • oauth_token: An OAuth token to include in the request.
  • oidc_token: An OIDC token to include in the request.
  • schedule_time: The time to schedule the task for.

Example usage:

1
import launchflow as lf
2
3
queue = lf.gcp.CloudTasksQueue("my-queue")
4
5
await queue.enqueue_async("https://example.com/endpoint", json={"key": "value"})