def main()

in cicd-deployers/bigquery_ddl_runner.py [0:0]

• 41 lines of code
• 1 McCabe index (conditional complexity)

def main(args: collections.abc.Sequence[str]) -> int:
    """The main function parses command-line arguments and calls the run_workflow function to execute the complete Dataform workflow.
    To run the script, provide the required command-line arguments:
        python intro.py --project_id your_project_id --location your_location --repository your_repo_name --dataset your_bq_dataset --branch your_branch
    """
    parser = argparse.ArgumentParser(description="BigQuery DDLs defined in files in a GCS bucket runner")
    parser.add_argument("--project_id",
                        type=str,
                        required=True,
                        help="The GCP project ID where the BigQuery client and storage client will be created.")
    parser.add_argument("--location",
                        type=str,
                        required=True,
                        help="The location of the BigQuery client and storage client")
    parser.add_argument("--bucket",
                        type=str,
                        required=True,
                        help="The bucket where there are DLL files to run")
    parser.add_argument("--ddl_project_id",
                        type=str,
                        required=True,
                        help="The project ID that will be replaced. It should be defined in the .sql file like: {$PROJECT_ID}")
    parser.add_argument("--ddl_dataset_id",
                        type=str,
                        required=True,
                        help="The dataset ID that will be replaced. It should be defined in the .sql file like: {$DATASET_ID}")
    parser.add_argument("--ddl_data_bucket_name",
                        type=str,
                        required=True,
                        help="The bucket name that will be replaced. It should be defined in the .sql file like: {DATA_BUCKET_NAME}")
    parser.add_argument("--ddl_connection_name",
                        type=str,
                        required=True,
                        help="The BigLake connection name that will be replaced. It should be defined in the .sql file like: {CONNECTION_NAME}")

    params = parser.parse_args(args)
    project_id = str(params.project_id)
    location = str(params.location)
    bucket = str(params.bucket)
    ddl_project_id = str(params.ddl_project_id)
    ddl_dataset_id = str(params.ddl_dataset_id)
    ddl_data_bucket_name = str(params.ddl_data_bucket_name)
    ddl_connection_name = str(params.ddl_connection_name)

    run_sql_queries_from_gcs(project_id=project_id, location=location, bucket=bucket, ddl_project_id=ddl_project_id,
                             ddl_dataset_id=ddl_dataset_id, ddl_data_bucket_name=ddl_data_bucket_name,
                             ddl_connection_name=ddl_connection_name)