Underground-Digital/Workflow-Engine
0
1from core.ops.ops_trace_manager import OpsTraceManager, provider_config_map2from extensions.ext_database import db3from models.model import App, TraceAppConfig4 5 6class OpsService:7 @classmethod8 def get_tracing_app_config(cls, app_id: str, tracing_provider: str):9 """10 Get tracing app config11 :param app_id: app id12 :param tracing_provider: tracing provider13 :return:14 """15 trace_config_data: TraceAppConfig = (16 db.session.query(TraceAppConfig)17 .filter(TraceAppConfig.app_id == app_id, TraceAppConfig.tracing_provider == tracing_provider)18 .first()19 )20 21 if not trace_config_data:22 return None23 24 # decrypt_token and obfuscated_token25 tenant_id = db.session.query(App).filter(App.id == app_id).first().tenant_id26 decrypt_tracing_config = OpsTraceManager.decrypt_tracing_config(27 tenant_id, tracing_provider, trace_config_data.tracing_config28 )29 new_decrypt_tracing_config = OpsTraceManager.obfuscated_decrypt_token(tracing_provider, decrypt_tracing_config)30 31 if tracing_provider == "langfuse" and (32 "project_key" not in decrypt_tracing_config or not decrypt_tracing_config.get("project_key")33 ):34 try:35 project_key = OpsTraceManager.get_trace_config_project_key(decrypt_tracing_config, tracing_provider)36 new_decrypt_tracing_config.update(37 {38 "project_url": "{host}/project/{key}".format(39 host=decrypt_tracing_config.get("host"), key=project_key40 )41 }42 )43 except Exception:44 new_decrypt_tracing_config.update(45 {"project_url": "{host}/".format(host=decrypt_tracing_config.get("host"))}46 )47 48 if tracing_provider == "langsmith" and (49 "project_url" not in decrypt_tracing_config or not decrypt_tracing_config.get("project_url")50 ):51 try:52 project_url = OpsTraceManager.get_trace_config_project_url(decrypt_tracing_config, tracing_provider)53 new_decrypt_tracing_config.update({"project_url": project_url})54 except Exception:55 new_decrypt_tracing_config.update({"project_url": "https://smith.langchain.com/"})56 57 trace_config_data.tracing_config = new_decrypt_tracing_config58 return trace_config_data.to_dict()59 60 @classmethod61 def create_tracing_app_config(cls, app_id: str, tracing_provider: str, tracing_config: dict):62 """63 Create tracing app config64 :param app_id: app id65 :param tracing_provider: tracing provider66 :param tracing_config: tracing config67 :return:68 """69 if tracing_provider not in provider_config_map and tracing_provider:70 return {"error": f"Invalid tracing provider: {tracing_provider}"}71 72 config_class, other_keys = (73 provider_config_map[tracing_provider]["config_class"],74 provider_config_map[tracing_provider]["other_keys"],75 )76 default_config_instance = config_class(**tracing_config)77 for key in other_keys:78 if key in tracing_config and tracing_config[key] == "":79 tracing_config[key] = getattr(default_config_instance, key, None)80 81 # api check82 if not OpsTraceManager.check_trace_config_is_effective(tracing_config, tracing_provider):83 return {"error": "Invalid Credentials"}84 85 # get project url86 if tracing_provider == "langfuse":87 project_key = OpsTraceManager.get_trace_config_project_key(tracing_config, tracing_provider)88 project_url = "{host}/project/{key}".format(host=tracing_config.get("host"), key=project_key)89 elif tracing_provider == "langsmith":90 project_url = OpsTraceManager.get_trace_config_project_url(tracing_config, tracing_provider)91 else:92 project_url = None93 94 # check if trace config already exists95 trace_config_data: TraceAppConfig = (96 db.session.query(TraceAppConfig)97 .filter(TraceAppConfig.app_id == app_id, TraceAppConfig.tracing_provider == tracing_provider)98 .first()99 )100 101 if trace_config_data:102 return None103 104 # get tenant id105 tenant_id = db.session.query(App).filter(App.id == app_id).first().tenant_id106 tracing_config = OpsTraceManager.encrypt_tracing_config(tenant_id, tracing_provider, tracing_config)107 if project_url:108 tracing_config["project_url"] = project_url109 trace_config_data = TraceAppConfig(110 app_id=app_id,111 tracing_provider=tracing_provider,112 tracing_config=tracing_config,113 )114 db.session.add(trace_config_data)115 db.session.commit()116 117 return {"result": "success"}118 119 @classmethod120 def update_tracing_app_config(cls, app_id: str, tracing_provider: str, tracing_config: dict):121 """122 Update tracing app config123 :param app_id: app id124 :param tracing_provider: tracing provider125 :param tracing_config: tracing config126 :return:127 """128 if tracing_provider not in provider_config_map:129 raise ValueError(f"Invalid tracing provider: {tracing_provider}")130 131 # check if trace config already exists132 current_trace_config = (133 db.session.query(TraceAppConfig)134 .filter(TraceAppConfig.app_id == app_id, TraceAppConfig.tracing_provider == tracing_provider)135 .first()136 )137 138 if not current_trace_config:139 return None140 141 # get tenant id142 tenant_id = db.session.query(App).filter(App.id == app_id).first().tenant_id143 tracing_config = OpsTraceManager.encrypt_tracing_config(144 tenant_id, tracing_provider, tracing_config, current_trace_config.tracing_config145 )146 147 # api check148 # decrypt_token149 decrypt_tracing_config = OpsTraceManager.decrypt_tracing_config(tenant_id, tracing_provider, tracing_config)150 if not OpsTraceManager.check_trace_config_is_effective(decrypt_tracing_config, tracing_provider):151 raise ValueError("Invalid Credentials")152 153 current_trace_config.tracing_config = tracing_config154 db.session.commit()155 156 return current_trace_config.to_dict()157 158 @classmethod159 def delete_tracing_app_config(cls, app_id: str, tracing_provider: str):160 """161 Delete tracing app config162 :param app_id: app id163 :param tracing_provider: tracing provider164 :return:165 """166 trace_config = (167 db.session.query(TraceAppConfig)168 .filter(TraceAppConfig.app_id == app_id, TraceAppConfig.tracing_provider == tracing_provider)169 .first()170 )171 172 if not trace_config:173 return None174 175 db.session.delete(trace_config)176 db.session.commit()177 178 return True179