import uuid import logging import argparse from pathlib import Path from gn_libs import jobs, mysqldb, sqlite3 logging.basicConfig( format="%(asctime)s — %(filename)s:%(lineno)s — %(levelname)s: %(message)s") logger = logging.getLogger(__name__) def load_data(conn, job): """Load the data attached in the given job.""" pass if __name__ == "__main__": def parse_args(): """Setup command-line arguments.""" parser = argparse.ArgumentParser( prog="load_phenotypes_to_db", description="Process the phenotypes' data and load it into the database.") parser.add_argument("db_uri", type=str, help="MariaDB/MySQL connection URL") parser.add_argument( "jobs_db_path", type=Path, help="Path to jobs' SQLite database.") parser.add_argument("job_id", type=uuid.UUID, help="ID of the running job") parser.add_argument( "--log-level", type=str, help="Determines what is logged out.", choices=("debug", "info", "warning", "error", "critical"), default="info") return parser.parse_args() def setup_logging(log_level: str): """Setup logging for the script.""" logger.setLevel(log_level) logging.getLogger("uploader.phenotypes.models").setLevel(log_level) def main(): """Entry-point for this script.""" args = parse_args() setup_logging(args.log_level.upper()) with (mysqldb.database_connection(args.db_uri) as conn, sqlite3.connection(args.jobs_db_path) as jobs_conn): try: return load_data(conn, jobs.job(jobs_conn, args.job_id)) except jobs.jobs.JobNotFound as _jne: logger.error("Could not find job with ID: %s", args.job_id) except Exception as _exc: logger.error("Loading failed with general exception!", exc_info=True, stack_info=True) return 1 main()