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
|
"""Build common script-entry structure."""
import logging
from typing import Callable
from argparse import Namespace
from redis import Redis
from MySQLdb import Connection
from gn_libs.mysqldb import database_connection
from uploader import jobs
from uploader.check_connections import check_db, check_redis
from scripts.redis_logger import setup_redis_logger
def build_main(
args: Namespace,
run_fn: Callable[[Connection, Namespace, logging.Logger], int],
loggername: str
) -> Callable[[],int]:
"""Build a function to be used as an entry-point for scripts."""
def main():
logging.basicConfig(
format=(
"%(asctime)s - %(levelname)s %(name)s: "
"(%(pathname)s: %(lineno)d) %(message)s"),
level=args.loglevel)
logger = logging.getLogger(loggername)
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):
fqjobid = jobs.job_key(args.redisprefix, args.jobid)
logger.addHandler(setup_redis_logger(
rconn,
fqjobid,
f"{fqjobid}:log-messages",
args.redisexpiry))
return run_fn(dbconn, args, logger)
return main
|
