mirror of
https://github.com/Significant-Gravitas/AutoGPT.git
synced 2026-04-08 03:00:28 -04:00
2f8cdf62ba
<!-- Clearly explain the need for these changes: --> This PR addresses the need for consistent error handling across all blocks in the AutoGPT platform. Previously, each block had to manually define an `error` field in their output schema, leading to code duplication and potential inconsistencies. Some blocks might forget to include the error field, making error handling unpredictable. ### Changes 🏗️ <!-- Concisely describe all of the changes made in this pull request: --> - **Created `BlockSchemaOutput` base class**: New base class that extends `BlockSchema` with a standardized `error` field - **Created `BlockSchemaInput` base class**: Added for consistency and future extensibility - **Updated 140+ block implementations**: Changed all block `Output` classes from `class Output(BlockSchema):` to `class Output(BlockSchemaOutput):` - **Removed manual error field definitions**: Eliminated hundreds of duplicate `error: str = SchemaField(...)` definitions - **Updated type annotations**: Changed `Block[BlockSchema, BlockSchema]` to `Block[BlockSchemaInput, BlockSchemaOutput]` throughout the codebase - **Fixed imports**: Added `BlockSchemaInput` and `BlockSchemaOutput` imports to all relevant files - **Maintained backward compatibility**: Updated `EmptySchema` to inherit from `BlockSchemaOutput` **Key Benefits:** - Consistent error handling across all blocks - Reduced code duplication (removed ~200 lines of repetitive error field definitions) - Type safety improvements with distinct input/output schema types - Blocks can still override error field with more specific descriptions when needed ### 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: <!-- Put your test plan here: --> - [x] Verified `poetry run format` passes (all linting, formatting, and type checking) - [x] Tested block instantiation works correctly (MediaDurationBlock, UnrealTextToSpeechBlock) - [x] Confirmed error fields are automatically present in all updated blocks - [x] Verified block loading system works (successfully loads 353+ blocks) - [x] Tested backward compatibility with EmptySchema - [x] Confirmed blocks can still override error field with custom descriptions - [x] Validated core schema inheritance chain works correctly #### For configuration changes: - [x] `.env.default` is updated or already compatible with my changes - [x] `docker-compose.yml` is updated or already compatible with my changes - [x] I have included a list of my configuration changes in the PR description (under **Changes**) *Note: No configuration changes were needed for this refactoring.* 🤖 Generated with [Claude Code](https://claude.ai/code) --------- Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Lluis Agusti <hi@llu.lu> Co-authored-by: Ubbe <hi@ubbe.dev>
366 lines
12 KiB
Python
366 lines
12 KiB
Python
from enum import Enum
|
|
from typing import Optional
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from backend.data.block import (
|
|
Block,
|
|
BlockCategory,
|
|
BlockOutput,
|
|
BlockSchemaInput,
|
|
BlockSchemaOutput,
|
|
)
|
|
from backend.data.model import SchemaField
|
|
|
|
from ._api import get_api
|
|
from ._auth import (
|
|
TEST_CREDENTIALS,
|
|
TEST_CREDENTIALS_INPUT,
|
|
GithubCredentials,
|
|
GithubCredentialsField,
|
|
GithubCredentialsInput,
|
|
)
|
|
|
|
|
|
# queued, in_progress, completed, waiting, requested, pending
|
|
class ChecksStatus(Enum):
|
|
QUEUED = "queued"
|
|
IN_PROGRESS = "in_progress"
|
|
COMPLETED = "completed"
|
|
WAITING = "waiting"
|
|
REQUESTED = "requested"
|
|
PENDING = "pending"
|
|
|
|
|
|
class ChecksConclusion(Enum):
|
|
SUCCESS = "success"
|
|
FAILURE = "failure"
|
|
NEUTRAL = "neutral"
|
|
CANCELLED = "cancelled"
|
|
TIMED_OUT = "timed_out"
|
|
ACTION_REQUIRED = "action_required"
|
|
SKIPPED = "skipped"
|
|
|
|
|
|
class GithubCreateCheckRunBlock(Block):
|
|
"""Block for creating a new check run on a GitHub repository."""
|
|
|
|
class Input(BlockSchemaInput):
|
|
credentials: GithubCredentialsInput = GithubCredentialsField("repo:status")
|
|
repo_url: str = SchemaField(
|
|
description="URL of the GitHub repository",
|
|
placeholder="https://github.com/owner/repo",
|
|
)
|
|
name: str = SchemaField(
|
|
description="The name of the check run (e.g., 'code-coverage')",
|
|
)
|
|
head_sha: str = SchemaField(
|
|
description="The SHA of the commit to check",
|
|
)
|
|
status: ChecksStatus = SchemaField(
|
|
description="Current status of the check run",
|
|
default=ChecksStatus.QUEUED,
|
|
)
|
|
conclusion: Optional[ChecksConclusion] = SchemaField(
|
|
description="The final conclusion of the check (required if status is completed)",
|
|
default=None,
|
|
)
|
|
details_url: str = SchemaField(
|
|
description="The URL for the full details of the check",
|
|
default="",
|
|
)
|
|
output_title: str = SchemaField(
|
|
description="Title of the check run output",
|
|
default="",
|
|
)
|
|
output_summary: str = SchemaField(
|
|
description="Summary of the check run output",
|
|
default="",
|
|
)
|
|
output_text: str = SchemaField(
|
|
description="Detailed text of the check run output",
|
|
default="",
|
|
)
|
|
|
|
class Output(BlockSchemaOutput):
|
|
class CheckRunResult(BaseModel):
|
|
id: int
|
|
html_url: str
|
|
status: str
|
|
|
|
check_run: CheckRunResult = SchemaField(
|
|
description="Details of the created check run"
|
|
)
|
|
error: str = SchemaField(
|
|
description="Error message if check run creation failed"
|
|
)
|
|
|
|
def __init__(self):
|
|
super().__init__(
|
|
id="2f45e89a-3b7d-4f22-b89e-6c4f5c7e1234",
|
|
description="Creates a new check run for a specific commit in a GitHub repository",
|
|
categories={BlockCategory.DEVELOPER_TOOLS},
|
|
input_schema=GithubCreateCheckRunBlock.Input,
|
|
output_schema=GithubCreateCheckRunBlock.Output,
|
|
test_input={
|
|
"repo_url": "https://github.com/owner/repo",
|
|
"name": "test-check",
|
|
"head_sha": "ce587453ced02b1526dfb4cb910479d431683101",
|
|
"status": ChecksStatus.COMPLETED.value,
|
|
"conclusion": ChecksConclusion.SUCCESS.value,
|
|
"output_title": "Test Results",
|
|
"output_summary": "All tests passed",
|
|
"credentials": TEST_CREDENTIALS_INPUT,
|
|
},
|
|
# requires a github app not available to oauth in our current system
|
|
disabled=True,
|
|
test_credentials=TEST_CREDENTIALS,
|
|
test_output=[
|
|
(
|
|
"check_run",
|
|
{
|
|
"id": 4,
|
|
"html_url": "https://github.com/owner/repo/runs/4",
|
|
"status": "completed",
|
|
},
|
|
),
|
|
],
|
|
test_mock={
|
|
"create_check_run": lambda *args, **kwargs: {
|
|
"id": 4,
|
|
"html_url": "https://github.com/owner/repo/runs/4",
|
|
"status": "completed",
|
|
}
|
|
},
|
|
)
|
|
|
|
@staticmethod
|
|
async def create_check_run(
|
|
credentials: GithubCredentials,
|
|
repo_url: str,
|
|
name: str,
|
|
head_sha: str,
|
|
status: ChecksStatus,
|
|
conclusion: Optional[ChecksConclusion] = None,
|
|
details_url: Optional[str] = None,
|
|
output_title: Optional[str] = None,
|
|
output_summary: Optional[str] = None,
|
|
output_text: Optional[str] = None,
|
|
) -> dict:
|
|
api = get_api(credentials)
|
|
|
|
class CheckRunData(BaseModel):
|
|
name: str
|
|
head_sha: str
|
|
status: str
|
|
conclusion: Optional[str] = None
|
|
details_url: Optional[str] = None
|
|
output: Optional[dict[str, str]] = None
|
|
|
|
data = CheckRunData(
|
|
name=name,
|
|
head_sha=head_sha,
|
|
status=status.value,
|
|
)
|
|
|
|
if conclusion:
|
|
data.conclusion = conclusion.value
|
|
|
|
if details_url:
|
|
data.details_url = details_url
|
|
|
|
if output_title or output_summary or output_text:
|
|
output_data = {
|
|
"title": output_title or "",
|
|
"summary": output_summary or "",
|
|
"text": output_text or "",
|
|
}
|
|
data.output = output_data
|
|
|
|
check_runs_url = f"{repo_url}/check-runs"
|
|
response = await api.post(
|
|
check_runs_url, data=data.model_dump_json(exclude_none=True)
|
|
)
|
|
result = response.json()
|
|
|
|
return {
|
|
"id": result["id"],
|
|
"html_url": result["html_url"],
|
|
"status": result["status"],
|
|
}
|
|
|
|
async def run(
|
|
self,
|
|
input_data: Input,
|
|
*,
|
|
credentials: GithubCredentials,
|
|
**kwargs,
|
|
) -> BlockOutput:
|
|
try:
|
|
result = await self.create_check_run(
|
|
credentials=credentials,
|
|
repo_url=input_data.repo_url,
|
|
name=input_data.name,
|
|
head_sha=input_data.head_sha,
|
|
status=input_data.status,
|
|
conclusion=input_data.conclusion,
|
|
details_url=input_data.details_url,
|
|
output_title=input_data.output_title,
|
|
output_summary=input_data.output_summary,
|
|
output_text=input_data.output_text,
|
|
)
|
|
yield "check_run", result
|
|
except Exception as e:
|
|
yield "error", str(e)
|
|
|
|
|
|
class GithubUpdateCheckRunBlock(Block):
|
|
"""Block for updating an existing check run on a GitHub repository."""
|
|
|
|
class Input(BlockSchemaInput):
|
|
credentials: GithubCredentialsInput = GithubCredentialsField("repo:status")
|
|
repo_url: str = SchemaField(
|
|
description="URL of the GitHub repository",
|
|
placeholder="https://github.com/owner/repo",
|
|
)
|
|
check_run_id: int = SchemaField(
|
|
description="The ID of the check run to update",
|
|
)
|
|
status: ChecksStatus = SchemaField(
|
|
description="New status of the check run",
|
|
)
|
|
conclusion: ChecksConclusion = SchemaField(
|
|
description="The final conclusion of the check (required if status is completed)",
|
|
)
|
|
output_title: Optional[str] = SchemaField(
|
|
description="New title of the check run output",
|
|
default=None,
|
|
)
|
|
output_summary: Optional[str] = SchemaField(
|
|
description="New summary of the check run output",
|
|
default=None,
|
|
)
|
|
output_text: Optional[str] = SchemaField(
|
|
description="New detailed text of the check run output",
|
|
default=None,
|
|
)
|
|
|
|
class Output(BlockSchemaOutput):
|
|
class CheckRunResult(BaseModel):
|
|
id: int
|
|
html_url: str
|
|
status: str
|
|
conclusion: Optional[str]
|
|
|
|
check_run: CheckRunResult = SchemaField(
|
|
description="Details of the updated check run"
|
|
)
|
|
|
|
def __init__(self):
|
|
super().__init__(
|
|
id="8a23c567-9d01-4e56-b789-0c12d3e45678", # Generated UUID
|
|
description="Updates an existing check run in a GitHub repository",
|
|
categories={BlockCategory.DEVELOPER_TOOLS},
|
|
input_schema=GithubUpdateCheckRunBlock.Input,
|
|
output_schema=GithubUpdateCheckRunBlock.Output,
|
|
# requires a github app not available to oauth in our current system
|
|
disabled=True,
|
|
test_input={
|
|
"repo_url": "https://github.com/owner/repo",
|
|
"check_run_id": 4,
|
|
"status": ChecksStatus.COMPLETED.value,
|
|
"conclusion": ChecksConclusion.SUCCESS.value,
|
|
"output_title": "Updated Results",
|
|
"output_summary": "All tests passed after retry",
|
|
"credentials": TEST_CREDENTIALS_INPUT,
|
|
},
|
|
test_credentials=TEST_CREDENTIALS,
|
|
test_output=[
|
|
(
|
|
"check_run",
|
|
{
|
|
"id": 4,
|
|
"html_url": "https://github.com/owner/repo/runs/4",
|
|
"status": "completed",
|
|
"conclusion": "success",
|
|
},
|
|
),
|
|
],
|
|
test_mock={
|
|
"update_check_run": lambda *args, **kwargs: {
|
|
"id": 4,
|
|
"html_url": "https://github.com/owner/repo/runs/4",
|
|
"status": "completed",
|
|
"conclusion": "success",
|
|
}
|
|
},
|
|
)
|
|
|
|
@staticmethod
|
|
async def update_check_run(
|
|
credentials: GithubCredentials,
|
|
repo_url: str,
|
|
check_run_id: int,
|
|
status: ChecksStatus,
|
|
conclusion: Optional[ChecksConclusion] = None,
|
|
output_title: Optional[str] = None,
|
|
output_summary: Optional[str] = None,
|
|
output_text: Optional[str] = None,
|
|
) -> dict:
|
|
api = get_api(credentials)
|
|
|
|
class UpdateCheckRunData(BaseModel):
|
|
status: str
|
|
conclusion: Optional[str] = None
|
|
output: Optional[dict[str, str]] = None
|
|
|
|
data = UpdateCheckRunData(
|
|
status=status.value,
|
|
)
|
|
|
|
if conclusion:
|
|
data.conclusion = conclusion.value
|
|
|
|
if output_title or output_summary or output_text:
|
|
output_data = {
|
|
"title": output_title or "",
|
|
"summary": output_summary or "",
|
|
"text": output_text or "",
|
|
}
|
|
data.output = output_data
|
|
|
|
check_run_url = f"{repo_url}/check-runs/{check_run_id}"
|
|
response = await api.patch(
|
|
check_run_url, data=data.model_dump_json(exclude_none=True)
|
|
)
|
|
result = response.json()
|
|
|
|
return {
|
|
"id": result["id"],
|
|
"html_url": result["html_url"],
|
|
"status": result["status"],
|
|
"conclusion": result.get("conclusion"),
|
|
}
|
|
|
|
async def run(
|
|
self,
|
|
input_data: Input,
|
|
*,
|
|
credentials: GithubCredentials,
|
|
**kwargs,
|
|
) -> BlockOutput:
|
|
try:
|
|
result = await self.update_check_run(
|
|
credentials=credentials,
|
|
repo_url=input_data.repo_url,
|
|
check_run_id=input_data.check_run_id,
|
|
status=input_data.status,
|
|
conclusion=input_data.conclusion,
|
|
output_title=input_data.output_title,
|
|
output_summary=input_data.output_summary,
|
|
output_text=input_data.output_text,
|
|
)
|
|
yield "check_run", result
|
|
except Exception as e:
|
|
yield "error", str(e)
|