feat!: migrate to use microgen (#38)

* feat!: migrate to use microgen

* Update UPGRADING.md

Co-authored-by: Bu Sun Kim <[email protected]>

Co-authored-by: Bu Sun Kim <[email protected]>

Author: 2 people

cloud-tasks/snippets/create_http_task.py

@@ -17,13 +17,9 @@
 import argparse
 
 
-def create_http_task(project,
-                     queue,
-                     location,
-                     url,
-                     payload=None,
-                     in_seconds=None,
-                     task_name=None):
+def create_http_task(
+    project, queue, location, url, payload=None, in_seconds=None, task_name=None
+):
     # [START cloud_tasks_create_http_task]
     """Create a task for a given queue with an arbitrary payload."""
 
@@ -47,23 +43,23 @@ def create_http_task(project,
 
     # Construct the request body.
     task = {
-            'http_request': {  # Specify the type of request.
-                'http_method': 'POST',
-                'url': url  # The full url path that the task will be sent to.
-            }
+        "http_request": {  # Specify the type of request.
+            "http_method": tasks_v2.HttpMethod.POST,
+            "url": url,  # The full url path that the task will be sent to.
+        }
     }
     if payload is not None:
         if isinstance(payload, dict):
             # Convert dict to JSON string
             payload = json.dumps(payload)
             # specify http content-type to application/json
-            task['http_request']['headers'] = {'Content-type': 'application/json'}
+            task["http_request"]["headers"] = {"Content-type": "application/json"}
 
         # The API expects a payload of type bytes.
         converted_payload = payload.encode()
 
         # Add the payload to the request.
-        task['http_request']['body'] = converted_payload
+        task["http_request"]["body"] = converted_payload
 
     if in_seconds is not None:
         # Convert "seconds from now" into an rfc3339 datetime string.
@@ -74,65 +70,65 @@ def create_http_task(project,
         timestamp.FromDatetime(d)
 
         # Add the timestamp to the tasks.
-        task['schedule_time'] = timestamp
+        task["schedule_time"] = timestamp
 
     if task_name is not None:
         # Add the name to tasks.
-        task['name'] = task_name
+        task["name"] = task_name
 
     # Use the client to build and send the task.
-    response = client.create_task(parent, task)
+    response = client.create_task(request={"parent": parent, "task": task})
 
-    print('Created task {}'.format(response.name))
+    print("Created task {}".format(response.name))
     # [END cloud_tasks_create_http_task]
     return response
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     parser = argparse.ArgumentParser(
         description=create_http_task.__doc__,
-        formatter_class=argparse.RawDescriptionHelpFormatter)
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
 
     parser.add_argument(
-        '--project',
-        help='Project of the queue to add the task to.',
-        required=True,
+        "--project", help="Project of the queue to add the task to.", required=True,
     )
 
     parser.add_argument(
-        '--queue',
-        help='ID (short name) of the queue to add the task to.',
+        "--queue",
+        help="ID (short name) of the queue to add the task to.",
         required=True,
     )
 
     parser.add_argument(
-        '--location',
-        help='Location of the queue to add the task to.',
-        required=True,
+        "--location", help="Location of the queue to add the task to.", required=True,
     )
 
     parser.add_argument(
-        '--url',
-        help='The full url path that the request will be sent to.',
+        "--url",
+        help="The full url path that the request will be sent to.",
         required=True,
     )
 
     parser.add_argument(
-        '--payload',
-        help='Optional payload to attach to the push queue.'
+        "--payload", help="Optional payload to attach to the push queue."
     )
 
     parser.add_argument(
-        '--in_seconds', type=int,
-        help='The number of seconds from now to schedule task attempt.'
+        "--in_seconds",
+        type=int,
+        help="The number of seconds from now to schedule task attempt.",
     )
 
-    parser.add_argument(
-        '--task_name',
-        help='Task name of the task to create'
-    )
+    parser.add_argument("--task_name", help="Task name of the task to create")
     args = parser.parse_args()
 
     create_http_task(
-        args.project, args.queue, args.location, args.url,
-        args.payload, args.in_seconds, args.task_name)
+        args.project,
+        args.queue,
+        args.location,
+        args.url,
+        args.payload,
+        args.in_seconds,
+        args.task_name,
+    )

cloud-tasks/snippets/create_http_task_test.py

@@ -20,29 +20,29 @@
 
 import create_http_task
 
-TEST_PROJECT_ID = os.getenv('GOOGLE_CLOUD_PROJECT')
-TEST_LOCATION = os.getenv('TEST_QUEUE_LOCATION', 'us-central1')
-TEST_QUEUE_NAME = f'my-queue-{uuid.uuid4().hex}'
+TEST_PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT")
+TEST_LOCATION = os.getenv("TEST_QUEUE_LOCATION", "us-central1")
+TEST_QUEUE_NAME = f"my-queue-{uuid.uuid4().hex}"
 
 
 @pytest.fixture()
 def test_queue():
     client = tasks_v2.CloudTasksClient()
-    parent = client.location_path(TEST_PROJECT_ID, TEST_LOCATION)
+    parent = f"projects/{TEST_PROJECT_ID}/locations/{TEST_LOCATION}"
     queue = {
         # The fully qualified path to the queue
-        'name': client.queue_path(
-            TEST_PROJECT_ID, TEST_LOCATION, TEST_QUEUE_NAME),
+        "name": client.queue_path(TEST_PROJECT_ID, TEST_LOCATION, TEST_QUEUE_NAME),
     }
-    q = client.create_queue(parent, queue)
+    q = client.create_queue(request={"parent": parent, "queue": queue})
 
     yield q
 
-    client.delete_queue(q.name)
+    client.delete_queue(request={"name": q.name})
 
 
 def test_create_http_task(test_queue):
-    url = 'https://example.com/task_handler'
+    url = "https://example.com/task_handler"
     result = create_http_task.create_http_task(
-        TEST_PROJECT_ID, TEST_QUEUE_NAME, TEST_LOCATION, url)
+        TEST_PROJECT_ID, TEST_QUEUE_NAME, TEST_LOCATION, url
+    )
     assert TEST_QUEUE_NAME in result.name

cloud-tasks/snippets/create_http_task_with_token.py

@@ -17,14 +17,16 @@
 import datetime
 
 
-def create_http_task(project,
-                     queue,
-                     location,
-                     url,
-                     service_account_email,
-                     payload=None,
-                     in_seconds=None,
-                     task_name=None):
+def create_http_task(
+    project,
+    queue,
+    location,
+    url,
+    service_account_email,
+    payload=None,
+    in_seconds=None,
+    task_name=None,
+):
     # [START cloud_tasks_create_http_task_with_token]
     """Create a task for a given queue with an arbitrary payload."""
 
@@ -47,21 +49,19 @@ def create_http_task(project,
 
     # Construct the request body.
     task = {
-            'http_request': {  # Specify the type of request.
-                'http_method': 'POST',
-                'url': url,  # The full url path that the task will be sent to.
-                'oidc_token': {
-                    'service_account_email': service_account_email
-                }
-            }
+        "http_request": {  # Specify the type of request.
+            "http_method": tasks_v2.HttpMethod.POST,
+            "url": url,  # The full url path that the task will be sent to.
+            "oidc_token": {"service_account_email": service_account_email},
+        }
     }
 
     if payload is not None:
         # The API expects a payload of type bytes.
         converted_payload = payload.encode()
 
         # Add the payload to the request.
-        task['http_request']['body'] = converted_payload
+        task["http_request"]["body"] = converted_payload
 
     if in_seconds is not None:
         # Convert "seconds from now" into an rfc3339 datetime string.
@@ -72,15 +72,17 @@ def create_http_task(project,
         timestamp.FromDatetime(d)
 
         # Add the timestamp to the tasks.
-        task['schedule_time'] = timestamp
+        task["schedule_time"] = timestamp
 
     if task_name is not None:
         # Add the name to tasks.
-        task['name'] = task_name
+        task["name"] = task_name
 
     # Use the client to build and send the task.
-    response = client.create_task(parent, task)
+    response = client.create_task(request={"parent": parent, "task": task})
 
-    print('Created task {}'.format(response.name))
+    print("Created task {}".format(response.name))
     return response
+
+
 # [END cloud_tasks_create_http_task_with_token]

cloud-tasks/snippets/create_http_task_with_token_test.py

@@ -20,34 +20,32 @@
 
 import create_http_task_with_token
 
-TEST_PROJECT_ID = os.getenv('GOOGLE_CLOUD_PROJECT')
-TEST_LOCATION = os.getenv('TEST_QUEUE_LOCATION', 'us-central1')
-TEST_QUEUE_NAME = f'my-queue-{uuid.uuid4().hex}'
+TEST_PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT")
+TEST_LOCATION = os.getenv("TEST_QUEUE_LOCATION", "us-central1")
+TEST_QUEUE_NAME = f"my-queue-{uuid.uuid4().hex}"
 TEST_SERVICE_ACCOUNT = (
-    'test-run-invoker@python-docs-samples-tests.iam.gserviceaccount.com')
+    "test-run-invoker@python-docs-samples-tests.iam.gserviceaccount.com"
+)
 
 
 @pytest.fixture()
 def test_queue():
     client = tasks_v2.CloudTasksClient()
-    parent = client.location_path(TEST_PROJECT_ID, TEST_LOCATION)
+    parent = f"projects/{TEST_PROJECT_ID}/locations/{TEST_LOCATION}"
     queue = {
         # The fully qualified path to the queue
-        'name': client.queue_path(
-            TEST_PROJECT_ID, TEST_LOCATION, TEST_QUEUE_NAME),
+        "name": client.queue_path(TEST_PROJECT_ID, TEST_LOCATION, TEST_QUEUE_NAME),
     }
-    q = client.create_queue(parent, queue)
+    q = client.create_queue(request={"parent": parent, "queue": queue})
 
     yield q
 
-    client.delete_queue(q.name)
+    client.delete_queue(request={"name": q.name})
 
 
 def test_create_http_task_with_token(test_queue):
-    url = 'https://example.com/task_handler'
-    result = create_http_task_with_token.create_http_task(TEST_PROJECT_ID,
-                                                          TEST_QUEUE_NAME,
-                                                          TEST_LOCATION,
-                                                          url,
-                                                          TEST_SERVICE_ACCOUNT)
+    url = "https://example.com/task_handler"
+    result = create_http_task_with_token.create_http_task(
+        TEST_PROJECT_ID, TEST_QUEUE_NAME, TEST_LOCATION, url, TEST_SERVICE_ACCOUNT
+    )
     assert TEST_QUEUE_NAME in result.name