"""Build common script-entry structure.""" import logging from typing import Callable from argparse import Namespace from redis import Redis from MySQLdb import Connection from uploader import jobs from uploader.db_utils import database_connection 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