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
|
"""Build common script-entry structure."""
from logging import Logger
from typing import Callable
from argparse import Namespace
from redis import Redis
from MySQLdb import Connection
from qc_app.db_utils import database_connection
from qc_app.check_connections import check_db, check_redis
from scripts.redis_logger import setup_redis_logger
def build_main(cli_args: Callable[[], Namespace],
run_fn: Callable[[Connection, Namespace], int],
logger: Logger,
loglevel: str = "INFO") -> Callable[[],int]:
"""Build a function to be used as an entry-point for scripts."""
def main():
args = cli_args()
check_db(args.databaseuri)
check_redis(args.redisuri)
if not args.rqtl2bundle.exists():
logger.error("File not found: '%s'.", args.rqtl2bundle)
return 2
with (Redis.from_url(args.redisuri, decode_responses=True) as rconn,
database_connection(args.databaseuri) as dbconn):
logger.addHandler(setup_redis_logger(
rconn,
args.jobid,
f"{str(args.jobid)}:log-messages",
args.redisexpiry))
logger.setLevel(loglevel)
return run_fn(dbconn, args)
return main
|
