def main()

in common/annotations_loader.py [0:0]

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

def main(args: typing.Sequence[str]) -> int:
    """BigQuery Annotations loader main"""

    parser = argparse.ArgumentParser(description="BigQuery Annotations Loader")
    parser.add_argument("--annotations-directory",
                        help="Annotation files directory",
                        type=str,
                        required=True)
    parser.add_argument("--debug",
                        help="Debugging mode.",
                        action="store_true",
                        default=False,
                        required=False)
    parser.add_argument("--config",
                        help="Data Foundation config.json.",
                        type=str,
                        required=False,
                        default="./config/config.json")
    options = parser.parse_args(args)
    logging.basicConfig(
        format="%(asctime)s | %(levelname)s | %(message)s",
        level=logging.INFO if not options.debug else logging.DEBUG,
    )

    logging.info("Cortex Annotations Loader for BigQuery.")

    config = load_config_file(options.config)

    logging.info("Loading BigQuery Annotations.")

    annotations_path = pathlib.Path(options.annotations_directory)
    if not annotations_path.exists():
        logging.critical("Directory `%s` doesn't exist.", str(annotations_path))
        return 1

    client = bigquery.Client(project=config["projectIdSource"],
                             location=config["location"])
    jinja_dict = initialize_jinja_from_config(config)

    for annotation_file in annotations_path.iterdir():
        load_annotations(jinja_dict, client, annotation_file.absolute())
    logging.info("BigQuery Annotations has been loaded!")

    return 0