1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
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()
|
