def list_tasks()

in clouddq-migration/dataplex.py [0:0]

• 24 lines of code
• 4 McCabe index (conditional complexity)

def list_tasks(gcp_project_id, region_id, lake_id)-> dict:
    '''
        Method to get the CloudDQ tasks
    '''
    try:
        # Create a client
        client = dataplex_v1.DataplexServiceClient()

        # Initialize request argument(s)
        request = dataplex_v1.ListTasksRequest(
            parent=f"projects/{gcp_project_id}/locations/{region_id}/lakes/{lake_id}",
        )

        # Make the request
        page_result = client.list_tasks(request=request)

        # Handle the response
        tasks = {}
        for response in page_result:
            task_id = response.name.split('/')[-1]
            if response.spark.file_uris:
                file_uri = response.spark.file_uris[-1]
                trigger_spec = response.trigger_spec
                tasks[task_id] = {
                    'file_uri': file_uri,
                    'trigger_spec': trigger_spec
                }
        return tasks
    except Exception as error:
        print(error)
        return None