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
62
|
"""Build common script-entry structure."""
import sys
import logging
from typing import Callable
from argparse import Namespace
from logging import StreamHandler
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[
[Redis, Connection, str, Namespace, logging.Logger],
int
],
loggername: str
) -> Callable[[],int]:
"""Build a function to be used as an entry-point for scripts."""
def main():
try:
logging.basicConfig(
format=(
"%(asctime)s - %(levelname)s %(name)s: "
"(%(pathname)s: %(lineno)d) %(message)s"),
level=args.loglevel)
logger = logging.getLogger(loggername)
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)
rconn.hset(fqjobid, "status", "started")
logger.addHandler(setup_redis_logger(
rconn,
fqjobid,
f"{fqjobid}:log-messages",
args.redisexpiry))
logger.addHandler(StreamHandler(stream=sys.stdout))
check_db(args.databaseuri)
check_redis(args.redisuri)
if not args.rqtl2bundle.exists():
logger.error("File not found: '%s'.", args.rqtl2bundle)
return 2
returncode = run_fn(rconn, dbconn, fqjobid, args, logger)
if returncode == 0:
rconn.hset(fqjobid, "status", "completed:success")
return returncode
rconn.hset(fqjobid, "status", "completed:error")
return returncode
except Exception as _exc:# pylint: disable=[broad-except]
logger.error("The process failed!", exc_info=True)
rconn.hset(fqjobid, "status", "completed:error")
return 4
return main
|
