Skip to content

feat(taskqueue): migrate push queue client to Cloud Tasks v2 while retaining v2beta3 for QueueStats #160

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
riddhi-shivhare wants to merge 1 commit into
base: taskqueue-migration-preview
Choose a base branch
from
Open

feat(taskqueue): migrate push queue client to Cloud Tasks v2 while retaining v2beta3 for QueueStats #160

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 14 additions & 12 deletions src/google/appengine/api/taskqueue/cloudtask.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
from google.appengine.api import app_identity
from google.appengine.api.taskqueue import taskqueue
from google.appengine.api.taskqueue import taskqueue_service_bytes_pb2 as taskqueue_service_pb2
from google.cloud import tasks_v2
from google.cloud import tasks_v2beta3
from google.protobuf import duration_pb2
from google.protobuf import field_mask_pb2
Expand Down Expand Up @@ -71,7 +72,7 @@ def create_tasks_in_cloud_tasks(queue_name, tasks, multiple):

def delete_tasks_in_cloud_tasks(queue_name, tasks, multiple):
"""Deletes tasks from a queue using Cloud Tasks Client SDK (supporting BatchDeleteTasks)."""
client = tasks_v2beta3.CloudTasksClient()
client = tasks_v2.CloudTasksClient()
project = _get_project_id()
region = _get_region()

Expand Down Expand Up @@ -131,7 +132,7 @@ def delete_tasks_in_cloud_tasks(queue_name, tasks, multiple):

def purge_queue_in_cloud_tasks(queue_name):
"""Purges all tasks in a queue using Cloud Tasks API."""
client = tasks_v2beta3.CloudTasksClient()
client = tasks_v2.CloudTasksClient()
project = _get_project_id()
region = _get_region()

Expand All @@ -147,7 +148,8 @@ def purge_queue_in_cloud_tasks(queue_name):


def fetch_queue_stats_in_cloud_tasks(queues, multiple):
"""Fetches queue statistics for given queues using Cloud Tasks API."""
"""Fetches queue statistics for given queues using Cloud Tasks API (v2beta3)."""
# QueueStats is retained on v2beta3 as it is out of scope for v2 GA
client = tasks_v2beta3.CloudTasksClient()
project = _get_project_id()
region = _get_region()
Expand Down Expand Up @@ -306,16 +308,16 @@ def _build_ct_task_payload(queue_name, task, client, project, region):
else:
body = task.payload

http_method = tasks_v2beta3.HttpMethod.POST
http_method = tasks_v2.HttpMethod.POST
if task.method:
method_map = {
'POST': tasks_v2beta3.HttpMethod.POST,
'GET': tasks_v2beta3.HttpMethod.GET,
'PUT': tasks_v2beta3.HttpMethod.PUT,
'DELETE': tasks_v2beta3.HttpMethod.DELETE,
'HEAD': tasks_v2beta3.HttpMethod.HEAD,
'POST': tasks_v2.HttpMethod.POST,
'GET': tasks_v2.HttpMethod.GET,
'PUT': tasks_v2.HttpMethod.PUT,
'DELETE': tasks_v2.HttpMethod.DELETE,
'HEAD': tasks_v2.HttpMethod.HEAD,
}
http_method = method_map.get(task.method, tasks_v2beta3.HttpMethod.POST)
http_method = method_map.get(task.method, tasks_v2.HttpMethod.POST)

app_engine_http_request = {
'http_method': http_method,
Expand Down Expand Up @@ -379,7 +381,7 @@ def _build_ct_task_payload(queue_name, task, client, project, region):

def _create_single_task_in_cloud_tasks(queue_name, task, multiple):
"""Helper to create a single task using CloudTasksClient CreateTask API."""
client = tasks_v2beta3.CloudTasksClient()
client = tasks_v2.CloudTasksClient()
project = _get_project_id()
region = _get_region()

Expand Down Expand Up @@ -410,7 +412,7 @@ def _create_single_task_in_cloud_tasks(queue_name, task, multiple):

def _create_batch_tasks_in_cloud_tasks(queue_name, tasks, multiple):
"""Helper to create tasks in batches using CloudTasksClient BatchCreateTasks API."""
client = tasks_v2beta3.CloudTasksClient()
client = tasks_v2.CloudTasksClient()
project = _get_project_id()
region = _get_region()

Expand Down
6 changes: 3 additions & 3 deletions src/google/appengine/api/taskqueue/cloudtask_transactional.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@
from google.appengine.api import datastore
from google.appengine.api.taskqueue import cloudtask
from google.appengine.api.taskqueue import taskqueue
from google.cloud import tasks_v2beta3
from google.cloud import tasks_v2
from google.protobuf.timestamp_pb2 import Timestamp

try:
Expand Down Expand Up @@ -117,7 +117,7 @@ def add_transactional_tasks(queue_name, tasks, multiple):

def build_task_payload_for_transactional_task(queue_name, task):
"""Builds the Cloud Tasks task payload for a transactional task."""
client = tasks_v2beta3.CloudTasksClient()
client = tasks_v2.CloudTasksClient()
project = cloudtask._get_project_id()
region = cloudtask._get_region()

Expand All @@ -126,7 +126,7 @@ def build_task_payload_for_transactional_task(queue_name, task):

def dispatch_task_payload(queue_name, task_payload):
"""Dispatches a pre-built task payload immediately using CloudTasksClient."""
client = tasks_v2beta3.CloudTasksClient()
client = tasks_v2.CloudTasksClient()
project = cloudtask._get_project_id()
region = cloudtask._get_region()

Expand Down