feat: 支持代码执行 Block #1475
Merged
feat: 支持代码执行 Block #1475
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…ution capabilities - Added a new Python Language Server implementation to support LSP features such as code completion, hover information, and signature help. - Implemented various diagnostic checkers including import validation, syntax error checking using Jedi, and mandatory function checks. - Enhanced the API with a WebSocket endpoint for LSP communication, allowing real-time code analysis and feedback. - Updated dependencies in `pyproject.toml` to include necessary libraries for LSP functionality. - Introduced a new `CodeBlock` class to execute user-defined code dynamically within the workflow. This commit significantly enhances the development experience by providing real-time feedback and code analysis capabilities.
Contributor
文件级别更改
提示和命令与 Sourcery 互动
自定义您的体验访问您的 仪表板 以：
获取帮助Original review guide in EnglishReviewer's GuideThis pull request integrates a Python Language Server (LSP) using Class Diagram for LSP and CodeBlock ComponentsclassDiagram
direction LR
class Block {
<<Abstract>>
+name: str
+inputs: dict
+outputs: dict
+execute() Any
}
class CodeBlock {
+name: str = "code_block"
+inputs: dict
+outputs: dict
+code: str
+execute(**kwargs) Dict[str, Any]
}
Block <|-- CodeBlock
note for CodeBlock "Executes user-provided Python code using exec()."
class LanguageServer {
<<pygls>>
+feature()
+thread()
+publish_diagnostics()
+show_message()
+workspace
}
class PythonLanguageServer {
-diagnostic_checkers: List~BaseDiagnostic~
-mandatory_function_checker: Optional~MandatoryFunctionDiagnostic~
+configure_mandatory_function_checker(config)
+_setup_handlers()
+_get_script(params) Optional~jedi.Script~
+_get_completions(params) CompletionList
+_get_hover(params) Optional~Hover~
+_get_signature_help(params) Optional~SignatureHelp~
+_publish_diagnostics(ls, doc_uri)
+_get_code_actions(params) Optional~List~CodeAction~~
}
LanguageServer <|-- PythonLanguageServer
PythonLanguageServer o-- "many" BaseDiagnostic : uses >
note for PythonLanguageServer "Handles LSP requests using Jedi and diagnostic checkers."
class asyncio.Transport {
<<Interface>>
+write(data)
+close()
}
class QuartWsTransport {
-queue: asyncio.Queue
+write(message: str)
+close()
}
asyncio.Transport <|-- QuartWsTransport
PythonLanguageServer o-- QuartWsTransport : uses >
class BaseDiagnostic {
<<ABC>>
+SOURCE_NAME: str
+ls: LanguageServer
+check(doc: Document) List~Diagnostic~
+get_code_actions(params, diagnostics) List~CodeAction~
+_create_diagnostic() Diagnostic
}
note for BaseDiagnostic "Abstract base for all diagnostic checkers."
class ImportDiagnostic {
+SOURCE_NAME: str = "import-check"
+check(doc: Document) List~Diagnostic~
+get_code_actions(params, diagnostics) List~CodeAction~
}
BaseDiagnostic <|-- ImportDiagnostic
ImportDiagnostic ..> ast : uses
ImportDiagnostic ..> importlib.util : uses
class JediSyntaxErrorDiagnostic {
+SOURCE_NAME: str = "syntax-error"
+check(doc: Document) List~Diagnostic~
}
BaseDiagnostic <|-- JediSyntaxErrorDiagnostic
JediSyntaxErrorDiagnostic ..> jedi : uses
class MandatoryFunctionDiagnostic {
+SOURCE_NAME: str = "mandatory-function-check"
-config: dict
+update_config(config)
+check(doc: Document) List~Diagnostic~
+get_code_actions(params, diagnostics) List~CodeAction~
}
BaseDiagnostic <|-- MandatoryFunctionDiagnostic
MandatoryFunctionDiagnostic ..> ast : uses
class PyflakesDiagnostic {
+SOURCE_NAME: str = "pyflakes"
+check(doc: Document) List~Diagnostic~
+get_code_actions(params, diagnostics) List~CodeAction~
}
BaseDiagnostic <|-- PyflakesDiagnostic
PyflakesDiagnostic ..> pyflakes_api : uses
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
lss233
changed the title
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| """ | ||
| try: | ||
| self.config = config | ||
| self.config_name = self.config["name"] |
There was a problem hiding this comment.
MyPy 类型错误: Value of type "None" is not indexable (index)
详细信息请参考 mypy 文档。
| try: | ||
| self.config = config | ||
| self.config_name = self.config["name"] | ||
| self.config_params = self.config["params"] |
There was a problem hiding this comment.
MyPy 类型错误: Value of type "None" is not indexable (index)
详细信息请参考 mypy 文档。
| self.config = config | ||
| self.config_name = self.config["name"] | ||
| self.config_params = self.config["params"] | ||
| self.config_return = self.config["return_type"] |
There was a problem hiding this comment.
MyPy 类型错误: Value of type "None" is not indexable (index)
详细信息请参考 mypy 文档。
| self.config_params = self.config["params"] | ||
| self.config_return = self.config["return_type"] | ||
| self.param_signatures = [ | ||
| f"{p['name']}: {p['type_hint']}" for p in self.config_params] |
There was a problem hiding this comment.
MyPy 类型错误: "None" has no attribute "iter" (not iterable) (attr-defined)
详细信息请参考 mypy 文档。
| self.config_return = self.config["return_type"] | ||
| self.param_signatures = [ | ||
| f"{p['name']}: {p['type_hint']}" for p in self.config_params] | ||
| self.expected_signature_str = f"def {self.config_name}({', '.join(self.param_signatures)}) -> {self.config_return}" |
There was a problem hiding this comment.
MyPy 类型错误: Argument 1 to "join" of "str" has incompatible type "None"; expected "Iterable[str]" (arg-type)
详细信息请参考 mypy 文档。
|
|
||
| def get_code_actions(self, params: CodeActionParams, relevant_diagnostics: List[Diagnostic]) -> List[CodeAction]: | ||
| """为强制函数错误提供代码操作""" | ||
| actions = [] |
There was a problem hiding this comment.
MyPy 类型错误: Need type annotation for "actions" (hint: "actions: list[] = ...") (var-annotated)
详细信息请参考 mypy 文档。
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
嘿 @lss233 - 我已经查看了你的更改 - 这里有一些反馈：
CodeBlock在用户提供的代码上使用exec()；考虑添加沙盒或探索更安全的执行替代方案，以减轻安全风险。- 新的
python_lsp.py文件非常大；考虑是否可以将与特定功能（例如，完成、悬停、不同的诊断）相关的部分进一步模块化。 - 审查 WebSocket LSP 处理程序的错误恢复逻辑，以解决客户端-服务器通信中潜在的边缘情况。
这是我在审查期间查看的内容
- 🟢 一般问题：一切看起来都不错
- 🟢 安全性：一切看起来都不错
- 🟢 测试：一切看起来都不错
- 🟡 复杂性：发现 2 个问题
- 🟢 文档：一切看起来都不错
帮助我更有用！请点击每个评论上的 👍 或 👎，我将使用反馈来改进你的评论。
Original comment in English
Hey @lss233 - I've reviewed your changes - here's some feedback:
- The
CodeBlockusesexec()on user-provided code; consider adding sandboxing or exploring safer execution alternatives to mitigate security risks. - The new
python_lsp.pyfile is quite large; consider if parts related to specific features (e.g., completions, hover, different diagnostics) could be further modularized. - Review the WebSocket LSP handler's error recovery logic for potential edge cases in client-server communication.
Here's what I looked at during the review
- 🟢 General issues: all looks good
- 🟢 Security: all looks good
- 🟢 Testing: all looks good
- 🟡 Complexity: 2 issues found
- 🟢 Documentation: all looks good
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| registry: BlockRegistry = g.container.resolve(BlockRegistry) | ||
| return jsonify(registry.get_type_compatibility_map()) | ||
|
|
||
| @block_bp.websocket("/code/lsp") |
Contributor
There was a problem hiding this comment.
issue (complexity): 考虑将发送者和接收者函数提取到单独的辅助函数中，以减少嵌套。
嵌套的异步发送者/接收者函数增加了复杂性。考虑将它们提取到单独的辅助函数中。例如：
# 在单独的模块中或在模块级别
async def lsp_sender(websocket, queue, logger):
while True:
message = await queue.get()
if message is None:
break
await websocket.send(message)
async def lsp_receiver(websocket, lsp_server, logger):
while True:
message_str = await websocket.receive()
try:
parsed_message = json.loads(
message_str,
object_hook=lsp_server.lsp._deserialize_message,
)
lsp_server.lsp._procedure_handler(parsed_message)
except json.JSONDecodeError:
logger.error(
f"无法解析收到的 LSP 消息: {message_str}",
exc_info=True,
)
except Exception as e:
logger.error(f"处理 LSP 消息时出错: {e}", exc_info=True)然后相应地更新路由处理程序：
@block_bp.websocket("/code/lsp")
async def code_lsp():
lsp_server = PythonLanguageServer(loop=asyncio.get_event_loop())
logger = get_logger("Web.Block.LSP")
queue = asyncio.Queue()
transport = QuartWsTransport(queue)
lsp_server.lsp.connection_made(transport)
lsp_server.lsp._send_only_body = True
logger.info("LSP WebSocket 连接已建立")
receive_task = asyncio.create_task(lsp_receiver(websocket, lsp_server, logger))
send_task = asyncio.create_task(lsp_sender(websocket, queue, logger))
logger.debug("已创建 LSP WebSocket 发送者和接收者任务")
try:
await asyncio.gather(receive_task, send_task)
except asyncio.CancelledError:
logger.info("LSP WebSocket 任务已取消")
except Exception as e:
logger.error(f"LSP WebSocket 连接错误: {e}", exc_info=True)
finally:
send_task.cancel()
try:
await send_task
except asyncio.CancelledError:
logger.debug("LSP WebSocket 发送者任务已取消")
receive_task.cancel()
try:
await receive_task
except asyncio.CancelledError:
logger.debug("LSP WebSocket 接收者任务已取消")
loop = asyncio.get_event_loop()
loop.run_in_executor(None, lsp_server.shutdown)
logger.info("websocket 连接已关闭")此重构保持了功能，同时减少了嵌套并提高了可测试性。
Original comment in English
issue (complexity): Consider extracting the sender and receiver functions into separate helper functions to reduce nesting.
The nested async sender/receiver functions add complexity. Consider extracting them into separate helper functions. For example:
# In a separate module or at the module level
async def lsp_sender(websocket, queue, logger):
while True:
message = await queue.get()
if message is None:
break
await websocket.send(message)
async def lsp_receiver(websocket, lsp_server, logger):
while True:
message_str = await websocket.receive()
try:
parsed_message = json.loads(
message_str,
object_hook=lsp_server.lsp._deserialize_message,
)
lsp_server.lsp._procedure_handler(parsed_message)
except json.JSONDecodeError:
logger.error(
f"Unable to parse received LSP message: {message_str}",
exc_info=True,
)
except Exception as e:
logger.error(f"Error processing LSP message: {e}", exc_info=True)Then update the route handler accordingly:
@block_bp.websocket("/code/lsp")
async def code_lsp():
lsp_server = PythonLanguageServer(loop=asyncio.get_event_loop())
logger = get_logger("Web.Block.LSP")
queue = asyncio.Queue()
transport = QuartWsTransport(queue)
lsp_server.lsp.connection_made(transport)
lsp_server.lsp._send_only_body = True
logger.info("LSP WebSocket connection established")
receive_task = asyncio.create_task(lsp_receiver(websocket, lsp_server, logger))
send_task = asyncio.create_task(lsp_sender(websocket, queue, logger))
logger.debug("Created LSP WebSocket sender and receiver tasks")
try:
await asyncio.gather(receive_task, send_task)
except asyncio.CancelledError:
logger.info("LSP WebSocket task cancelled")
except Exception as e:
logger.error(f"LSP WebSocket connection error: {e}", exc_info=True)
finally:
send_task.cancel()
try:
await send_task
except asyncio.CancelledError:
logger.debug("LSP WebSocket sender task cancelled")
receive_task.cancel()
try:
await receive_task
except asyncio.CancelledError:
logger.debug("LSP WebSocket receiver task cancelled")
loop = asyncio.get_event_loop()
loop.run_in_executor(None, lsp_server.shutdown)
logger.info("websocket connection closed")This refactor maintains functionality while reducing nesting and improving testability.
| self.inputs[input_spec["name"]] = Input(input_spec["name"], input_spec["label"], Any, 'user-specified object') # type: ignore | ||
| for output_spec in outputs: | ||
| self.outputs[output_spec["name"]] = Output(output_spec["name"], output_spec["label"], Any, 'user-specified object') # type: ignore | ||
| self.code = code |
Contributor
There was a problem hiding this comment.
issue (complexity): 考虑将两个 exec 调用重构为辅助方法，以隔离错误处理并简化测试。
考虑将两个单独的 exec 调用重构为辅助方法。这样，您可以隔离错误处理并简化测试。例如，您可以将定义执行和函数调用提取到如下方法中：
def _execute_definition(self, code: str, exec_globals: Dict[str, Any], exec_locals: Dict[str, Any]) -> None:
try:
exec(code, exec_globals, exec_locals)
except Exception as e:
logger.error(f"代码定义执行期间出错: {e}", exc_info=True)
raise RuntimeError(f"提供的代码定义中出错: {e}") from e
def _call_user_function(self, exec_globals: Dict[str, Any], exec_locals: Dict[str, Any], kwargs: Dict[str, Any]) -> None:
exec_locals['__input_kwargs__'] = kwargs
exec_globals.update(exec_locals)
call_code = "__result__ = execute(**__input_kwargs__)"
try:
exec(call_code, exec_globals, exec_locals)
except Exception as e:
logger.error(f"用户函数 'execute' 执行期间出错: {e}", exc_info=True)
raise RuntimeError(f"执行用户函数 'execute' 期间出错: {e}") from e然后，在您的主 execute 方法中，您可以调用这些辅助方法：
def execute(self, **kwargs: Any) -> Dict[str, Any]:
logger = get_logger("Block.Code")
exec_globals = globals().copy()
exec_locals: Dict[str, Any] = {}
logger.debug(f"正在执行代码定义:\n{self.code}")
self._execute_definition(self.code, exec_globals, exec_locals)
if 'execute' not in exec_locals or not callable(exec_locals['execute']):
raise ValueError("提供的代码必须定义一个名为 'execute' 的可调用函数")
logger.debug(f"正在执行函数调用: execute(**{list(kwargs.keys())})")
self._call_user_function(exec_globals, exec_locals, kwargs)
if '__result__' not in exec_locals:
logger.error("内部错误：执行用户代码调用后未找到结果 '__result__'。")
raise RuntimeError("无法从用户代码执行中检索结果。")
return exec_locals['__result__']此重构减少了 execute 中的嵌套，同时保持了功能完整。
Original comment in English
issue (complexity): Consider refactoring the two exec calls into helper methods to isolate error handling and simplify testing.
Consider refactoring the two separate exec calls into helper methods. This way you can isolate error handling and simplify testing. For example, you could extract the definition execution and the function call into methods like:
def _execute_definition(self, code: str, exec_globals: Dict[str, Any], exec_locals: Dict[str, Any]) -> None:
try:
exec(code, exec_globals, exec_locals)
except Exception as e:
logger.error(f"Error during code definition execution: {e}", exc_info=True)
raise RuntimeError(f"Error in provided code definition: {e}") from e
def _call_user_function(self, exec_globals: Dict[str, Any], exec_locals: Dict[str, Any], kwargs: Dict[str, Any]) -> None:
exec_locals['__input_kwargs__'] = kwargs
exec_globals.update(exec_locals)
call_code = "__result__ = execute(**__input_kwargs__)"
try:
exec(call_code, exec_globals, exec_locals)
except Exception as e:
logger.error(f"Error during user function 'execute' execution: {e}", exc_info=True)
raise RuntimeError(f"Error during execution of user function 'execute': {e}") from eThen, in your main execute method, you can call these helpers:
def execute(self, **kwargs: Any) -> Dict[str, Any]:
logger = get_logger("Block.Code")
exec_globals = globals().copy()
exec_locals: Dict[str, Any] = {}
logger.debug(f"Executing code definition:\n{self.code}")
self._execute_definition(self.code, exec_globals, exec_locals)
if 'execute' not in exec_locals or not callable(exec_locals['execute']):
raise ValueError("Provided code must define a callable function named 'execute'")
logger.debug(f"Executing function call: execute(**{list(kwargs.keys())})")
self._call_user_function(exec_globals, exec_locals, kwargs)
if '__result__' not in exec_locals:
logger.error("Internal error: Result '__result__' not found after executing user code call.")
raise RuntimeError("Failed to retrieve result from user code execution.")
return exec_locals['__result__']This refactoring reduces nesting within execute while keeping functionality intact.
|
|
||
| try: | ||
| position = params.position | ||
| line = position.line + 1 |
Contributor
There was a problem hiding this comment.
issue (code-quality): 使用命名表达式简化赋值和条件 [×2] (use-named-expression)
Original comment in English
issue (code-quality): Use named expression to simplify assignment and conditional [×2] (use-named-expression)
|
|
||
| try: | ||
| position = params.position | ||
| line = position.line + 1 |
Contributor
There was a problem hiding this comment.
issue (code-quality): 我们发现了这些问题：
- 使用命名表达式简化赋值和条件 (
use-named-expression) - 将代码提取到方法中 (
extract-method) - 删除未使用的索引时，删除对
enumerate的不必要调用 (remove-unused-enumerate)
Original comment in English
issue (code-quality): We've found these issues:
- Use named expression to simplify assignment and conditional (
use-named-expression) - Extract code out into method (
extract-method) - Remove unnecessary calls to
enumeratewhen the index is not used (remove-unused-enumerate)
Comment on lines
+476
to
+477
| checker_diagnostics = checker.check(document) | ||
| if checker_diagnostics: |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): 使用命名表达式简化赋值和条件 (use-named-expression)
Suggested change
| checker_diagnostics = checker.check(document) | |
| if checker_diagnostics: | |
| if checker_diagnostics := checker.check(document): |
Original comment in English
suggestion (code-quality): Use named expression to simplify assignment and conditional (use-named-expression)
Suggested change
| checker_diagnostics = checker.check(document) | |
| if checker_diagnostics: | |
| if checker_diagnostics := checker.check(document): |
Comment on lines
+511
to
+520
| relevant_diagnostics = diagnostics_by_source.get(checker_name, []) | ||
| if relevant_diagnostics: | ||
| try: | ||
| checker_actions = checker.get_code_actions( | ||
| params, relevant_diagnostics) | ||
| if checker_actions: | ||
| actions.extend(checker_actions) | ||
| except Exception as e: | ||
| logger.error( | ||
| f"Code action checker '{checker_name}' error: {str(e)}", exc_info=True) |
Contributor
There was a problem hiding this comment.
issue (code-quality): 使用命名表达式简化赋值和条件 [×2] (use-named-expression)
Original comment in English
issue (code-quality): Use named expression to simplify assignment and conditional [×2] (use-named-expression)
| self.code = code | ||
|
|
||
| def execute(self, **kwargs: Any) -> Dict[str, Any]: # 使用 Any 兼容各种输入类型 | ||
| logger = get_logger("Block.Code") |
Contributor
There was a problem hiding this comment.
issue (code-quality): 我们发现了这些问题：
- 内联立即返回的变量 (
inline-immediately-returned-variable) - 通过联合运算符合并字典更新 (
dict-assign-update-to-union)
Original comment in English
issue (code-quality): We've found these issues:
- Inline variable that is immediately returned (
inline-immediately-returned-variable) - Merge dictionary updates via the union operator (
dict-assign-update-to-union)
Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com>
| """ | ||
| try: | ||
| self.config = config | ||
| self.config_name = self.config["name"] |
There was a problem hiding this comment.
MyPy 类型错误: Value of type "None" is not indexable (index)
详细信息请参考 mypy 文档。
| try: | ||
| self.config = config | ||
| self.config_name = self.config["name"] | ||
| self.config_params = self.config["params"] |
There was a problem hiding this comment.
MyPy 类型错误: Value of type "None" is not indexable (index)
详细信息请参考 mypy 文档。
| self.config = config | ||
| self.config_name = self.config["name"] | ||
| self.config_params = self.config["params"] | ||
| self.config_return = self.config["return_type"] |
There was a problem hiding this comment.
MyPy 类型错误: Value of type "None" is not indexable (index)
详细信息请参考 mypy 文档。
| self.config_params = self.config["params"] | ||
| self.config_return = self.config["return_type"] | ||
| self.param_signatures = [ | ||
| f"{p['name']}: {p['type_hint']}" for p in self.config_params] |
There was a problem hiding this comment.
MyPy 类型错误: "None" has no attribute "iter" (not iterable) (attr-defined)
详细信息请参考 mypy 文档。
| self.config_return = self.config["return_type"] | ||
| self.param_signatures = [ | ||
| f"{p['name']}: {p['type_hint']}" for p in self.config_params] | ||
| self.expected_signature_str = f"def {self.config_name}({', '.join(self.param_signatures)}) -> {self.config_return}" |
There was a problem hiding this comment.
MyPy 类型错误: Argument 1 to "join" of "str" has incompatible type "None"; expected "Iterable[str]" (arg-type)
详细信息请参考 mypy 文档。
|
|
||
| def get_code_actions(self, params: CodeActionParams, relevant_diagnostics: List[Diagnostic]) -> List[CodeAction]: | ||
| """为强制函数错误提供代码操作""" | ||
| actions = [] |
There was a problem hiding this comment.
MyPy 类型错误: Need type annotation for "actions" (hint: "actions: list[] = ...") (var-annotated)
详细信息请参考 mypy 文档。
… class - Replaced `python-lsp-server` with `pygls` in `pyproject.toml` to streamline language server integration. - Enhanced the `MandatoryFunctionDiagnostic` class by adding optional configuration attributes for improved flexibility and error handling during function declaration checks.
MyPy 类型检查通过 ✅PR 修改的代码行通过了类型检查。 |
Codecov ReportAttention: Patch coverage is
✅ All tests successful. No failed tests found. Additional details and impacted files@@ Coverage Diff @@
## master #1475 +/- ##
==========================================
- Coverage 65.84% 60.72% -5.12%
==========================================
Files 161 167 +6
Lines 8148 9044 +896
==========================================
+ Hits 5365 5492 +127
- Misses 2783 3552 +769 ☔ View full report in Codecov by Sentry. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
…ution capabilities
pyproject.tomlto include necessary libraries for LSP functionality.CodeBlockclass to execute user-defined code dynamically within the workflow.This commit significantly enhances the development experience by providing real-time feedback and code analysis capabilities.
好的，这是将拉取请求摘要翻译成中文的结果：
Sourcery 总结
将一个全面的 Python 语言服务器与高级代码分析和执行功能集成到工作流系统中
新功能：
CodeBlock增强功能：
Any类型文档：
pyproject.toml中的项目依赖项，以包含 LSP 和诊断库测试：
Original summary in English
Summary by Sourcery
Integrate a comprehensive Python Language Server with advanced code analysis and execution capabilities into the workflow system
New Features:
Enhancements:
Documentation:
Tests: