def get_or_create_dcr_listing()

in analytics-hub/snippets/create_listing_python/main.py [0:0]

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

def get_or_create_dcr_listing(client: bigquery_analyticshub_v1.AnalyticsHubServiceClient, project_id: str, location: str, exchange_id: str, listing_id: str, shared_ds: str, source_view: str):
    # Initialize request argument(s)
    request = bigquery_analyticshub_v1.GetListingRequest(
        name=f"projects/{project_id}/locations/{location}/dataExchanges/{exchange_id}/listings/{listing_id}",
    )

    # Make the request
    try:
        response = client.get_listing(request=request)
        # Handle the response
        print(response)
        return response
    except Exception as ex:
        if ex.code == HTTPStatus.NOT_FOUND:
            print("Not found, creating")
            # Initialize request argument(s)
            listing = bigquery_analyticshub_v1.Listing()
            listing.display_name = source_view
            listing.primary_contact = "primary@contact.co"

            listing.bigquery_dataset = bigquery_analyticshub_v1.Listing.BigQueryDatasetSource()
            listing.bigquery_dataset.dataset = f"projects/{project_id}/datasets/{shared_ds}"
            
            listing_ds_selected_resource = bigquery_analyticshub_v1.Listing.BigQueryDatasetSource.SelectedResource()
            listing_ds_selected_resource.table = f"projects/{project_id}/datasets/{shared_ds}/tables/{source_view}"

            listing.bigquery_dataset.selected_resources = [ listing_ds_selected_resource ]

            listing.restricted_export_config = bigquery_analyticshub_v1.Listing.RestrictedExportConfig()
            listing.restricted_export_config.enabled = True
            listing.restricted_export_config.restrict_direct_table_access = True
            listing.restricted_export_config.restrict_query_result = False

            request = bigquery_analyticshub_v1.CreateListingRequest(
                parent=f"projects/{project_id}/locations/{location}/dataExchanges/{exchange_id}",
                listing_id=listing_id,
                listing=listing,
            )

            try:
                # Make the request
                response = client.create_listing(request=request)
                # Handle the response
                print(response)
                return response
            except Exception as ex:
                print(ex)
        else:
            print(ex)
    return False