Files
AutoGPT/autogpt_platform/backend/backend/blocks/maths.py
Zamil Majdy 2f8cdf62ba feat(backend): Standardize error handling with BlockSchemaInput & BlockSchemaOutput base class (#11257)
<!-- 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>
2025-10-30 12:28:08 +00:00

131 lines
4.1 KiB
Python

import operator
from enum import Enum
from typing import Any
from backend.data.block import (
Block,
BlockCategory,
BlockOutput,
BlockSchemaInput,
BlockSchemaOutput,
)
from backend.data.model import SchemaField
class Operation(Enum):
ADD = "Add"
SUBTRACT = "Subtract"
MULTIPLY = "Multiply"
DIVIDE = "Divide"
POWER = "Power"
class CalculatorBlock(Block):
class Input(BlockSchemaInput):
operation: Operation = SchemaField(
description="Choose the math operation you want to perform",
placeholder="Select an operation",
)
a: float = SchemaField(
description="Enter the first number (A)", placeholder="For example: 10"
)
b: float = SchemaField(
description="Enter the second number (B)", placeholder="For example: 5"
)
round_result: bool = SchemaField(
description="Do you want to round the result to a whole number?",
default=False,
)
class Output(BlockSchemaOutput):
result: float = SchemaField(description="The result of your calculation")
def __init__(self):
super().__init__(
id="b1ab9b19-67a6-406d-abf5-2dba76d00c79",
input_schema=CalculatorBlock.Input,
output_schema=CalculatorBlock.Output,
description="Performs a mathematical operation on two numbers.",
categories={BlockCategory.LOGIC},
test_input={
"operation": Operation.ADD.value,
"a": 10.0,
"b": 5.0,
"round_result": False,
},
test_output=[
("result", 15.0),
],
)
async def run(self, input_data: Input, **kwargs) -> BlockOutput:
operation = input_data.operation
a = input_data.a
b = input_data.b
operations = {
Operation.ADD: operator.add,
Operation.SUBTRACT: operator.sub,
Operation.MULTIPLY: operator.mul,
Operation.DIVIDE: operator.truediv,
Operation.POWER: operator.pow,
}
op_func = operations[operation]
try:
if operation == Operation.DIVIDE and b == 0:
raise ZeroDivisionError("Cannot divide by zero")
result = op_func(a, b)
if input_data.round_result:
result = round(result)
yield "result", result
except ZeroDivisionError:
yield "result", float("inf") # Return infinity for division by zero
except Exception:
yield "result", float("nan") # Return NaN for other errors
class CountItemsBlock(Block):
class Input(BlockSchemaInput):
collection: Any = SchemaField(
description="Enter the collection you want to count. This can be a list, dictionary, string, or any other iterable.",
placeholder="For example: [1, 2, 3] or {'a': 1, 'b': 2} or 'hello'",
)
class Output(BlockSchemaOutput):
count: int = SchemaField(description="The number of items in the collection")
def __init__(self):
super().__init__(
id="3c9c2f42-b0c3-435f-ba35-05f7a25c772a",
input_schema=CountItemsBlock.Input,
output_schema=CountItemsBlock.Output,
description="Counts the number of items in a collection.",
categories={BlockCategory.LOGIC},
test_input={"collection": [1, 2, 3, 4, 5]},
test_output=[
("count", 5),
],
)
async def run(self, input_data: Input, **kwargs) -> BlockOutput:
collection = input_data.collection
try:
if isinstance(collection, (str, list, tuple, set, dict)):
count = len(collection)
elif hasattr(collection, "__iter__"):
count = sum(1 for _ in collection)
else:
raise ValueError("Input is not a countable collection")
yield "count", count
except Exception:
yield "count", -1 # Return -1 to indicate an error