mirror of
https://github.com/Significant-Gravitas/AutoGPT.git
synced 2026-04-08 03:00:28 -04:00
113e87a23c
I'm getting circular import issues because there is a lot of cross-importing between `backend.data`, `backend.blocks`, and other modules. This change reduces block-related cross-imports and thus risk of breaking circular imports. ### Changes 🏗️ - Strip down `backend.data.block` - Move `Block` base class and related class/enum defs to `backend.blocks._base` - Move `is_block_auth_configured` to `backend.blocks._utils` - Move `get_blocks()`, `get_io_block_ids()` etc. to `backend.blocks` (`__init__.py`) - Update imports everywhere - Remove unused and poorly typed `Block.create()` - Change usages from `block_cls.create()` to `block_cls()` - Improve typing of `load_all_blocks` and `get_blocks` - Move cross-import of `backend.api.features.library.model` from `backend/data/__init__.py` to `backend/data/integrations.py` - Remove deprecated attribute `NodeModel.webhook` - Re-generate OpenAPI spec and fix frontend usage - Eliminate module-level `backend.blocks` import from `blocks/agent.py` - Eliminate module-level `backend.data.execution` and `backend.executor.manager` imports from `blocks/helpers/review.py` - Replace `BlockInput` with `GraphInput` for graph inputs ### Checklist 📋 #### For code changes: - [x] I have clearly listed my changes in the PR description - [x] I have made a test plan - [x] I have tested my changes according to the test plan: - CI static type-checking + tests should be sufficient for this
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
import os
|
|
import re
|
|
from typing import Type
|
|
|
|
from backend.blocks._base import (
|
|
Block,
|
|
BlockCategory,
|
|
BlockOutput,
|
|
BlockSchemaInput,
|
|
BlockSchemaOutput,
|
|
)
|
|
from backend.data.model import SchemaField
|
|
|
|
|
|
class BlockInstallationBlock(Block):
|
|
"""
|
|
This block allows the verification and installation of other blocks in the system.
|
|
|
|
NOTE:
|
|
This block allows remote code execution on the server, and it should be used
|
|
for development purposes only.
|
|
"""
|
|
|
|
class Input(BlockSchemaInput):
|
|
code: str = SchemaField(
|
|
description="Python code of the block to be installed",
|
|
)
|
|
|
|
class Output(BlockSchemaOutput):
|
|
success: str = SchemaField(
|
|
description="Success message if the block is installed successfully",
|
|
)
|
|
error: str = SchemaField(
|
|
description="Error message if the block installation fails",
|
|
)
|
|
|
|
def __init__(self):
|
|
super().__init__(
|
|
id="45e78db5-03e9-447f-9395-308d712f5f08",
|
|
description="Given a code string, this block allows the verification and installation of a block code into the system.",
|
|
categories={BlockCategory.BASIC},
|
|
input_schema=BlockInstallationBlock.Input,
|
|
output_schema=BlockInstallationBlock.Output,
|
|
disabled=True,
|
|
)
|
|
|
|
async def run(self, input_data: Input, **kwargs) -> BlockOutput:
|
|
code = input_data.code
|
|
|
|
if search := re.search(r"class (\w+)\(Block\):", code):
|
|
class_name = search.group(1)
|
|
else:
|
|
raise RuntimeError("No class found in the code.")
|
|
|
|
if search := re.search(r"id=\"(\w+-\w+-\w+-\w+-\w+)\"", code):
|
|
file_name = search.group(1)
|
|
else:
|
|
raise RuntimeError("No UUID found in the code.")
|
|
|
|
block_dir = os.path.dirname(__file__)
|
|
file_path = f"{block_dir}/{file_name}.py"
|
|
module_name = f"backend.blocks.{file_name}"
|
|
with open(file_path, "w") as f:
|
|
f.write(code)
|
|
|
|
try:
|
|
module = __import__(module_name, fromlist=[class_name])
|
|
block_class: Type[Block] = getattr(module, class_name)
|
|
block = block_class()
|
|
|
|
from backend.util.test import execute_block_test
|
|
|
|
await execute_block_test(block)
|
|
yield "success", "Block installed successfully."
|
|
except Exception as e:
|
|
os.remove(file_path)
|
|
raise RuntimeError(f"[Code]\n{code}\n\n[Error]\n{str(e)}")
|