def get_or_create_exchange()

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

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

def get_or_create_exchange(client: bigquery_analyticshub_v1.AnalyticsHubServiceClient, project_id: str, location: str, exchange_id: str, is_dcr: bool):
    # Initialize request argument(s)
    request = bigquery_analyticshub_v1.GetDataExchangeRequest(
        name=f"projects/{project_id}/locations/{location}/dataExchanges/{exchange_id}",
    )

    # Make the request
    try:
        response = client.get_data_exchange(request=request)
        # Handle the response
        print(response)
        return response
    except Exception as ex:
        if ex.code == HTTPStatus.NOT_FOUND:
            print("Not found, creating")
            # DataCleanRoom
            shared_environment_config = bigquery_analyticshub_v1.SharingEnvironmentConfig()
            if is_dcr:
                shared_environment_config.dcr_exchange_config = bigquery_analyticshub_v1.SharingEnvironmentConfig.DcrExchangeConfig()
                exTitleTag = "Data Clean Room"
            else:
                shared_environment_config.default_exchange_config = bigquery_analyticshub_v1.SharingEnvironmentConfig.DefaultExchangeConfig()
                exTitleTag = "Data Exchange"
            # Initialize request argument(s)
            data_exchange = bigquery_analyticshub_v1.DataExchange()
            data_exchange.display_name = f"Example {exTitleTag} - created using python API"
            data_exchange.description = f"Example {exTitleTag} - created using python API"
            data_exchange.primary_contact = ""
            data_exchange.documentation = "https://link.to.optional.documentation/"
            data_exchange.sharing_environment_config = shared_environment_config

            request = bigquery_analyticshub_v1.CreateDataExchangeRequest(
                parent=f"projects/{project_id}/locations/{location}",
                data_exchange_id=exchange_id,
                data_exchange=data_exchange,
            )

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