import threading from utils.logging import logging local = threading.local() def set_tl_app_id_pubkey(app_id: int, pubkey: str): logging.info("set app_id %s and pubkey %s in thread local, thread id %s", app_id, pubkey, threading.get_ident()) local.app_id = app_id local.pubkey = pubkey def get_tl_app_id_pubkey(): return int(getattr(local, "app_id", 0)), getattr(local, "pubkey", "") def set_tl_user_token(user_token): logging.info("set user_token %s in thread local, thread id %s", user_token, threading.get_ident()) local.user_token = user_token def get_tl_user_token(): return getattr(local, "user_token", "") def get_tl_trace_id(): metrics = get_local_metrics() return metrics.get("trace_id", "") def get_data_context_id(): logging.info("get data_context_id from thread local, thread id %s, data_context_id is %s", threading.get_ident(), local.data_context_id) # we may got an attribute error here, but it's better than silence return local.data_context_id def set_data_context_id(data_context_id: int): logging.info("set data_context_id %s in thread local, thread id %s", data_context_id, threading.get_ident()) local.data_context_id = data_context_id def get_session(): return local.session def set_session(session): local.session = session def set_local_metrics(metrics): logging.info("set metrics to local: %s", metrics) local.metrics = metrics def get_local_metrics() -> dict: logging.info("load metrics from local") return getattr(local, "metrics", {}) def clear_data_context_id(): logging.info("clear data_context_id in thread local, thread id %s", threading.get_ident()) if hasattr(local, "data_context_id"): delattr(local, "data_context_id") def set_tl_override_ai_model(override_ai_model): logging.info("set override_ai_model %s in thread local, thread id %s", override_ai_model, threading.get_ident()) local.override_ai_model = override_ai_model def get_tl_override_ai_model(): return getattr(local, "override_ai_model", "")