in tools/cloud_functions/gcs_event_based_ingest/backfill.py [0:0]
def parse_args(args: List[str]) -> argparse.Namespace:
"""argument parser for backfill CLI"""
parser = argparse.ArgumentParser(
description="utility to backfill success file notifications "
"or run the cloud function locally in concurrent threads.")
parser.add_argument(
"--gcs-path",
"-p",
help="GCS path (e.g. gs://bucket/prefix/to/search/)to search for "
"existing _SUCCESS files",
required=True,
)
parser.add_argument(
"--mode",
"-m",
help="How to perform the backfill: LOCAL run cloud function main"
" method locally (in concurrent threads) or NOTIFICATIONS just push"
" notifications to Pub/Sub for a deployed version of the cloud function"
" to pick up. Default is NOTIFICATIONS.",
required=False,
type=str.upper,
choices=["LOCAL", "NOTIFICATIONS"],
default="NOTIFICATIONS",
)
parser.add_argument(
"--pubsub-topic",
"--topic",
"-t",
help="Pub/Sub notifications topic to post notifications for. "
"i.e. projects/{PROJECT_ID}/topics/{TOPIC_ID} "
"Required if using NOTIFICATIONS mode.",
required=False,
default=None,
)
parser.add_argument(
"--success-filename",
"-f",
help="Override the default success filename '_SUCCESS'",
required=False,
default="_SUCCESS",
)
parser.add_argument(
"--destination-regex",
"-r",
help="Override the default destination regex for determining BigQuery"
"destination based on information encoded in the GCS path of the"
"success file",
required=False,
default=None,
)
return parser.parse_args(args)