bcgov/mines-digital-trust

Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
import os, threading, time, json, logging, requests
Identical blocks of code found in 2 locations. Consider refactoring.
from datetime import datetime
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
TRACE_EVENTS = os.getenv("TRACE_EVENTS", "True").lower() == "true"
Identical blocks of code found in 2 locations. Consider refactoring.
TRACE_LABEL = os.getenv("TRACE_LABEL", "bcreg.controller")
Identical blocks of code found in 2 locations. Consider refactoring.
TRACE_TAG = os.getenv("TRACE_TAG", "acapy.events")
Identical blocks of code found in 2 locations. Consider refactoring.
TRACE_LOG_TARGET = "log"
Identical blocks of code found in 2 locations. Consider refactoring.
TRACE_TARGET = os.getenv("TRACE_TARGET", TRACE_LOG_TARGET)
Identical blocks of code found in 2 locations. Consider refactoring.
LOG_LEVEL = os.environ.get('LOG_LEVEL', 'WARNING').upper()
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
# need to specify an env variable RECORD_TIMINGS=True to get method timings
Identical blocks of code found in 2 locations. Consider refactoring.
RECORD_TIMINGS = os.getenv('RECORD_TIMINGS', 'False').lower() == 'true'
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
timings = {}
Identical blocks of code found in 2 locations. Consider refactoring.
timing_lock = threading.Lock()
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
LOGGER = logging.getLogger(__name__)
Identical blocks of code found in 2 locations. Consider refactoring.
if TRACE_EVENTS and TRACE_TARGET == TRACE_LOG_TARGET:
Identical blocks of code found in 2 locations. Consider refactoring.
    LOGGER.setLevel(logging.INFO)
Identical blocks of code found in 2 locations. Consider refactoring.
elif LOG_LEVEL and 0 < len(LOG_LEVEL):
Identical blocks of code found in 2 locations. Consider refactoring.
    LOGGER.setLevel(LOG_LEVEL)
Identical blocks of code found in 2 locations. Consider refactoring.
DT_FMT = '%Y-%m-%d %H:%M:%S.%f%z'
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
def clear_stats():
Identical blocks of code found in 2 locations. Consider refactoring.
    global timings
Identical blocks of code found in 2 locations. Consider refactoring.
    timing_lock.acquire()
Identical blocks of code found in 2 locations. Consider refactoring.
    try:
Identical blocks of code found in 2 locations. Consider refactoring.
        timings = {}
Identical blocks of code found in 2 locations. Consider refactoring.
    finally:
Identical blocks of code found in 2 locations. Consider refactoring.
        timing_lock.release()
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
def get_stats():
Identical blocks of code found in 2 locations. Consider refactoring.
    timing_lock.acquire()
Identical blocks of code found in 2 locations. Consider refactoring.
    try:
Identical blocks of code found in 2 locations. Consider refactoring.
        return timings
Identical blocks of code found in 2 locations. Consider refactoring.
    finally:
Identical blocks of code found in 2 locations. Consider refactoring.
        timing_lock.release()
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
def log_timing_method(method, start_time, end_time, success, data=None):
Identical blocks of code found in 2 locations. Consider refactoring.
    if not RECORD_TIMINGS:
Identical blocks of code found in 2 locations. Consider refactoring.
        return
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
    timing_lock.acquire()
Identical blocks of code found in 2 locations. Consider refactoring.
    try:
Identical blocks of code found in 2 locations. Consider refactoring.
        elapsed_time = end_time - start_time
Identical blocks of code found in 2 locations. Consider refactoring.
        if not method in timings:
Identical blocks of code found in 2 locations. Consider refactoring.
            timings[method] = {
Identical blocks of code found in 2 locations. Consider refactoring.
                "total_count": 1,
Identical blocks of code found in 2 locations. Consider refactoring.
                "success_count": 1 if success else 0,
Identical blocks of code found in 2 locations. Consider refactoring.
                "fail_count": 0 if success else 1,
Identical blocks of code found in 2 locations. Consider refactoring.
                "min_time": elapsed_time,
Identical blocks of code found in 2 locations. Consider refactoring.
                "max_time": elapsed_time,
Identical blocks of code found in 2 locations. Consider refactoring.
                "total_time": elapsed_time,
Identical blocks of code found in 2 locations. Consider refactoring.
                "avg_time": elapsed_time,
Identical blocks of code found in 2 locations. Consider refactoring.
                "data": {},
Identical blocks of code found in 2 locations. Consider refactoring.
            }
Identical blocks of code found in 2 locations. Consider refactoring.
        else:
Identical blocks of code found in 2 locations. Consider refactoring.
            timings[method]["total_count"] = timings[method]["total_count"] + 1
Identical blocks of code found in 2 locations. Consider refactoring.
            if success:
Identical blocks of code found in 2 locations. Consider refactoring.
                timings[method]["success_count"] = timings[method]["success_count"] + 1
Identical blocks of code found in 2 locations. Consider refactoring.
            else:
Identical blocks of code found in 2 locations. Consider refactoring.
                timings[method]["fail_count"] = timings[method]["fail_count"] + 1
Identical blocks of code found in 2 locations. Consider refactoring.
            if elapsed_time > timings[method]["max_time"]:
Identical blocks of code found in 2 locations. Consider refactoring.
                timings[method]["max_time"] = elapsed_time
Identical blocks of code found in 2 locations. Consider refactoring.
            if elapsed_time < timings[method]["min_time"]:
Identical blocks of code found in 2 locations. Consider refactoring.
                timings[method]["min_time"] = elapsed_time
Identical blocks of code found in 2 locations. Consider refactoring.
            timings[method]["total_time"] = timings[method]["total_time"] + elapsed_time
Identical blocks of code found in 2 locations. Consider refactoring.
            timings[method]["avg_time"] = (
Identical blocks of code found in 2 locations. Consider refactoring.
                timings[method]["total_time"] / timings[method]["total_count"]
Identical blocks of code found in 2 locations. Consider refactoring.
            )
Identical blocks of code found in 2 locations. Consider refactoring.
        if data:
Identical blocks of code found in 2 locations. Consider refactoring.
            timings[method]["data"][str(timings[method]["total_count"])] = data
Identical blocks of code found in 2 locations. Consider refactoring.
    finally:
Identical blocks of code found in 2 locations. Consider refactoring.
        timing_lock.release()
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
def log_timing_event(method, message, start_time, end_time, success, outcome=None):
Identical blocks of code found in 2 locations. Consider refactoring.
    """Record a timing event in the system log or http endpoint."""
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
    if (not TRACE_EVENTS) and (not message.get("trace")):
Identical blocks of code found in 2 locations. Consider refactoring.
        return
Identical blocks of code found in 2 locations. Consider refactoring.
    if not TRACE_TARGET:
Identical blocks of code found in 2 locations. Consider refactoring.
        return
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
    msg_id = "N/A"
Identical blocks of code found in 2 locations. Consider refactoring.
    thread_id = message["thread_id"] if message.get("thread_id") else "N/A"
Identical blocks of code found in 2 locations. Consider refactoring.
    handler = TRACE_LABEL
Identical blocks of code found in 2 locations. Consider refactoring.
    ep_time = time.time()
Identical blocks of code found in 2 locations. Consider refactoring.
    str_time = datetime.utcfromtimestamp(ep_time).strftime(DT_FMT)
Identical blocks of code found in 2 locations. Consider refactoring.
    if end_time:
Identical blocks of code found in 2 locations. Consider refactoring.
        str_outcome = method + ".SUCCESS" if success else ".FAIL"
Identical blocks of code found in 2 locations. Consider refactoring.
    else:
Identical blocks of code found in 2 locations. Consider refactoring.
        str_outcome = method + ".START"
Identical blocks of code found in 2 locations. Consider refactoring.
    if outcome:
Identical blocks of code found in 2 locations. Consider refactoring.
        str_outcome = str_outcome + "." + outcome
Identical blocks of code found in 2 locations. Consider refactoring.
    event = {
Identical blocks of code found in 2 locations. Consider refactoring.
        "msg_id": msg_id,
Identical blocks of code found in 2 locations. Consider refactoring.
        "thread_id": thread_id if thread_id else msg_id,
Identical blocks of code found in 2 locations. Consider refactoring.
        "traced_type": method,
Identical blocks of code found in 2 locations. Consider refactoring.
        "timestamp": ep_time,
Identical blocks of code found in 2 locations. Consider refactoring.
        "str_time": str_time,
Identical blocks of code found in 2 locations. Consider refactoring.
        "handler": str(handler),
Identical blocks of code found in 2 locations. Consider refactoring.
        "ellapsed_milli": int(1000 * (end_time - start_time)) if end_time else 0,
Identical blocks of code found in 2 locations. Consider refactoring.
        "outcome": str_outcome,
Identical blocks of code found in 2 locations. Consider refactoring.
    }
Identical blocks of code found in 2 locations. Consider refactoring.
    event_str = json.dumps(event)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
    try:
Identical blocks of code found in 2 locations. Consider refactoring.
        if TRACE_TARGET == TRACE_LOG_TARGET:
Identical blocks of code found in 2 locations. Consider refactoring.
            # write to standard log file
Identical blocks of code found in 2 locations. Consider refactoring.
            LOGGER.error(" %s %s", TRACE_TAG, event_str)
Identical blocks of code found in 2 locations. Consider refactoring.
        else:
Identical blocks of code found in 2 locations. Consider refactoring.
            # should be an http endpoint
Identical blocks of code found in 2 locations. Consider refactoring.
            _ = requests.post(
Identical blocks of code found in 2 locations. Consider refactoring.
                TRACE_TARGET + TRACE_TAG,
Identical blocks of code found in 2 locations. Consider refactoring.
                data=event_str,
Identical blocks of code found in 2 locations. Consider refactoring.
                headers={"Content-Type": "application/json"}
Identical blocks of code found in 2 locations. Consider refactoring.
            )
Identical blocks of code found in 2 locations. Consider refactoring.
    except Exception as e:
Identical blocks of code found in 2 locations. Consider refactoring.
        LOGGER.error(
Identical blocks of code found in 2 locations. Consider refactoring.
            "Error logging trace target: %s tag: %s event: %s",
Identical blocks of code found in 2 locations. Consider refactoring.
            TRACE_TARGET,
Identical blocks of code found in 2 locations. Consider refactoring.
            TRACE_TAG,
Identical blocks of code found in 2 locations. Consider refactoring.
            event_str
Identical blocks of code found in 2 locations. Consider refactoring.
        )
Identical blocks of code found in 2 locations. Consider refactoring.
        LOGGER.exception(e)