瀏覽代碼

refactor(code_executor): update input type annotations to use Mapping for better type safety (#10478)

tags/0.11.1
-LAN- 11 月之前
父節點
當前提交
b8b6cd409a
No account linked to committer's email address

+ 3
- 0
api/core/helper/code_executor/__init__.py 查看文件

from .code_executor import CodeExecutor, CodeLanguage

__all__ = ["CodeExecutor", "CodeLanguage"]

+ 3
- 2
api/core/helper/code_executor/code_executor.py 查看文件

import logging import logging
from collections.abc import Mapping
from enum import Enum from enum import Enum
from threading import Lock from threading import Lock
from typing import Optional
from typing import Any, Optional


from httpx import Timeout, post from httpx import Timeout, post
from pydantic import BaseModel from pydantic import BaseModel
return response.data.stdout or "" return response.data.stdout or ""


@classmethod @classmethod
def execute_workflow_code_template(cls, language: CodeLanguage, code: str, inputs: dict) -> dict:
def execute_workflow_code_template(cls, language: CodeLanguage, code: str, inputs: Mapping[str, Any]) -> dict:
""" """
Execute code Execute code
:param language: code language :param language: code language

+ 5
- 3
api/core/helper/code_executor/template_transformer.py 查看文件

import re import re
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from base64 import b64encode from base64 import b64encode
from collections.abc import Mapping
from typing import Any




class TemplateTransformer(ABC): class TemplateTransformer(ABC):
_result_tag: str = "<<RESULT>>" _result_tag: str = "<<RESULT>>"


@classmethod @classmethod
def transform_caller(cls, code: str, inputs: dict) -> tuple[str, str]:
def transform_caller(cls, code: str, inputs: Mapping[str, Any]) -> tuple[str, str]:
""" """
Transform code to python runner Transform code to python runner
:param code: code :param code: code
pass pass


@classmethod @classmethod
def serialize_inputs(cls, inputs: dict) -> str:
def serialize_inputs(cls, inputs: Mapping[str, Any]) -> str:
inputs_json_str = json.dumps(inputs, ensure_ascii=False).encode() inputs_json_str = json.dumps(inputs, ensure_ascii=False).encode()
input_base64_encoded = b64encode(inputs_json_str).decode("utf-8") input_base64_encoded = b64encode(inputs_json_str).decode("utf-8")
return input_base64_encoded return input_base64_encoded


@classmethod @classmethod
def assemble_runner_script(cls, code: str, inputs: dict) -> str:
def assemble_runner_script(cls, code: str, inputs: Mapping[str, Any]) -> str:
# assemble runner script # assemble runner script
script = cls.get_runner_script() script = cls.get_runner_script()
script = script.replace(cls._code_placeholder, code) script = script.replace(cls._code_placeholder, code)

Loading…
取消
儲存