[Python] python core module to create 1 log file for each call

I have below code for logging, in a core_replication.py file. This file is called by multiples python scripts (say table1.py, table2.py), each of which is to replicate a single table from source to target. core_replication.py takes care of creating a unique log file for each call. Each call can be a direct call to the core_replication.py or by importing core_replication.py into each of table1.py, table2.py etc. Many of these calls can happen in parallel.

How to make sure I create log files as core_replication_pid_timestamp.log or core_replication_table1_pid_timestamp.log etc.

I am concerned parallel calls can cause one log message being written to wrong log file.

import logging
import os
import sys
import time
from pathlib import Path

# ----------------------------------------------------------------------------------------------------
# Log Function
# ----------------------------------------------------------------------------------------------------
# define module-level logger (aka child logger)
logger = logging.getLogger(__name__)

project_name= 'tsmr' #Adjust as needed. Give a project name to uniquely identify tables.
script_start_time= time.strftime('%Y%m%d_%H%M%S',time.localtime())
script_pid= os.getpid() #pid of caller is probably pid for core also, unless called as subprocess.

def logger_setup_fn():
# check caller script basename
this_script_basename= Path(__file__).stem
this_script_dir= Path(__file__).parent
try:
caller_script_basename= Path(sys.argv[0]).stem
except:
caller_script_basename= Path(__file__).stem
# create logs dir if not exists, based on this script location.
if this_script_dir.name=='modules':
log_file_dir= this_script_dir.parent.joinpath('logs')
else:
log_file_dir= this_script_dir.joinpath('logs')
log_file_dir.mkdir(exist_ok=True)
# construct log file full name
if caller_script_basename != this_script_basename:
log_file_basename= f'{this_script_basename}_{caller_script_basename}_{project_name}_{script_pid}_{script_start_time}.log'
else:
log_file_basename= f'{this_script_basename}_{project_name}_{script_pid}_{script_start_time}.log'
log_file_fullname= log_file_dir.joinpath(log_file_basename)

# define root logger (once) (aka parent logger)
if not logging.getLogger().hasHandlers():
logging.basicConfig(
filename= log_file_fullname,
filemode= 'a',
level= logging.INFO, #CRITICAL, ERROR, WARNING, INFO, DEBUG
format='%(asctime)s.%(msecs)03d/%(name)s/%(levelname)s//%(message)s',
datefmt='%Y%m%d_%H%M%S'
)

# redirect print results to logger
class StreamToLoggerCl: #followed suffix Cl for classes
def write(self,message): #standrd interface. do not suffix _mt for this method
if message.strip():
logger.info(message.strip())
def flush(self): pass #standrd interface. do not suffix _mt for this method

sys.stdout= StreamToLoggerCl()
sys.stderr= StreamToLoggerCl()

# notify log file name to stdout once (exception for redirection class above)
sys.__stdout__.write(f'PFB log file: {log_file_fullname.resolve()}')
# check this script depth
this_script_depth= len(this_script_dir.resolve().parts) -1
min_depth= 3
if this_script_depth < min_depth:
print(f'ERROR! core script is located too close to root directory. Maintain a mimimun depth of {min_depth} sub directories from root.')
sys.exit(1)
else: pass

# Call logger_setup_fn() at import
logger_setup_fn()

#...other code including main() etc.

Continue reading...