def get_or_create_listing()

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

• 42 lines of code
• 5 McCabe index (conditional complexity)

def get_or_create_listing(client: bigquery_analyticshub_v1.AnalyticsHubServiceClient, project_id: str, location: str, exchange_id: str, listing_id: str, restrict_egress: bool, shared_ds: 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 = "Example Exchange Listing - created using python API"
            listing.description = "Example Exchange Listing - created using python API"

            listing.data_provider = bigquery_analyticshub_v1.DataProvider()
            listing.data_provider.name = "Example Exchange Listing - created using python API"
            listing.data_provider.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.restricted_export_config = bigquery_analyticshub_v1.Listing.RestrictedExportConfig()

            if restrict_egress:
                listing.restricted_export_config.enabled = True
                listing.restricted_export_config.restrict_direct_table_access = True
                listing.restricted_export_config.restrict_query_result = True
            else:
                listing.restricted_export_config.enabled = False
                listing.restricted_export_config.restrict_direct_table_access = False
                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