"""General utility functions.""" import os import time from functools import wraps from signal import SIGINT, SIGTERM from sys import exit as sys_exit import psutil from connectors.logging import log from config import ( ENVIRONMENT, INPUT_FILE_NAME, VENDOR_ID, ) def getTimeStamp(): return time.strftime('%Y%m%d-%H%M%S') def error_prompt(text='', stars=15): text = '*' * stars + text + '*' * stars print(text) def convert_snake_to_camel(text): components = text.split('_') return components[0] + ''.join(x.title() for x in components[1:]) def cleanup(): # Kill any lingering child processes current_process = psutil.Process(os.getpid()) for child in current_process.children(recursive=True): log.info(f"Terminating child process: {child.pid}") log.info(f'Child PID: {child.pid}') log.info(f'Child Name: {child.name()}') log.info(f'Child Status: {child.status()}') log.info(f'Child CPU Times: {child.cpu_times()}') log.info(f'Child Memory Info: {child.memory_info()}') log.info(f'Child Open Files: {child.open_files()}') log.info(f'Child Connections: {child.net_connections()}') child.terminate() # child.wait() # Ensure the child process has terminated # Give processes a moment to terminate, then force kill any stubborn ones for child in current_process.children(recursive=True): if child.is_running(): log.warning(f"Force killing process: {child.pid}") child.kill() def signal_handler(sig, frame): if sig == SIGINT: log.info("Caught SIGINT, shutting down cleanly...") elif sig == SIGTERM: log.info("Caught SIGTERM, shutting down cleanly...") cleanup() sys_exit(0) def track_runtime(func): """Decorator to log the runtime of a function. This is an example of a decorator. The decorator is a function that takes another function as an argument and returns a new function. The new function is a wrapper that adds some functionality to the original function. In this case, the wrapper logs the runtime of the function it wraps. Args: func (function): The function to wrap. Returns: function: The wrapped function. """ @wraps(func) # Imported from functools above -- THIS IS THE MAGIC def wrapper(*args, **kwargs): # This is the new function # This is where we add our "Decorations" # Start the timer! start_time = time.time() # Call the original function, and grab the result # -- THIS IS WHAT WE ARE WRAPPING result = func(*args, **kwargs) # End the timer! end_time = time.time() elapsed_time = end_time - start_time # Log the runtime if elapsed_time > 60: elapsed_time /= 60 log.info( f'{func.__name__} completed in {elapsed_time:.2f} minutes.') else: log.info( f'{func.__name__} completed in {elapsed_time:.2f} seconds.') # Return the ORIGINAL FUNCTION'S result return result # Return the new function, to substitute for the original whereever the # decorator is used return wrapper def check_env(filename, cached_filename): """Check the environment for required variables. Args: filename (str): The filename to process. Returns: None """ # if not VENDOR_ID: # log.error( # 'Vendor ID is not set in. Set VENDOR_ID in env.' # ) # return if ENVIRONMENT.lower() not in ['qa', 'prod', 'dev']: log.error( 'Valid ENVIRONMENT is required. Set ENVIRONMENT in env.' ) return if not INPUT_FILE_NAME and not filename and not cached_filename: log.error( 'Input file is required. Set INPUT_FILE_NAME in env, or pass ' '"--filename" or "--cached_filename" arguments.' ) return log.info( f'Started at: {getTimeStamp()}' ) log.info( f'Environment: {ENVIRONMENT}' ) if INPUT_FILE_NAME: log.info( f'Using Config Input File: {INPUT_FILE_NAME}' ) if VENDOR_ID: log.info( f'Using Vendor ID: {VENDOR_ID}' ) def log_runtime(func): """Decorator to log the runtime of a function. This is an example of a decorator. The decorator is a function that takes another function as an argument and returns a new function. The new function is a wrapper that adds some functionality to the original function. In this case, the wrapper logs the runtime of the function it wraps. Args: func (function): The function to wrap. Returns: function: The wrapped function. """ @wraps(func) # Imported from functools above -- THIS IS THE MAGIC def wrapper(*args, **kwargs): # This is the new function # This is where we add our "Decorations" # Start the timer! start_time = time.time() # Call the original function, and grab the result # -- THIS IS WHAT WE ARE WRAPPING result = func(*args, **kwargs) # End the timer! end_time = time.time() elapsed_time = end_time - start_time # Log the runtime if elapsed_time > 60: elapsed_time /= 60 log.info( f'{func.__name__} completed in {elapsed_time:.2f} minutes.') else: log.info( f'{func.__name__} completed in {elapsed_time:.2f} seconds.') # Return the ORIGINAL FUNCTION'S result return result # Return the new function, to substitute for the original whereever the # decorator is used return wrapper