def main()

in src/cai-export/main.py [0:0]

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

def main(request):
    org_id = os.environ.get("ORG_ID")
    project_id = os.environ.get("PROJECT_ID")
    resource_dataset = os.environ.get("RESOURCE_DATASET")
    table_prefix = os.environ.get("TABLE_PREFIX")

    client = asset_v1.AssetServiceClient()
    output_config = asset_v1.OutputConfig()

    # Parent Config
    parent = f"organizations/{org_id}"

    # BQ Destination Config
    output_config.bigquery_destination.table = table_prefix
    output_config.bigquery_destination.force = True
    output_config.bigquery_destination.separate_tables_per_asset_type = True

    # Partition Spec Config
    partition_spec = asset_v1.PartitionSpec()
    partition_key = asset_v1.PartitionSpec.PartitionKey.REQUEST_TIME
    partition_spec.partition_key = asset_v1.PartitionSpec.PartitionKey.REQUEST_TIME

    output_config.bigquery_destination.partition_spec.partition_key = partition_key

    configs = {
        "resource": {
            "destination": f"projects/{project_id}/datasets/{resource_dataset}",
            "content_type": asset_v1.ContentType.RESOURCE,
            "asset_types": [
                "iam.googleapis.com/ServiceAccount",
                "iam.googleapis.com/ServiceAccountKey",
            ],
            "response": None,
        }
    }

    for config in configs:
        output_config.bigquery_destination.dataset = configs[config]["destination"]
        configs[config]["response"] = client.export_assets(
            request={
                "parent": parent,
                "content_type": configs[config]["content_type"],
                "asset_types": configs[config]["asset_types"],
                "output_config": output_config,
            }
        )

    # Resource Export Time >>> IAM Export Time, so we output operationId for Resource
    output = {"operationId": configs["resource"]["response"].operation.name}
    return jsonify(output)