github/AutoGPT
1
1
Fork 0
mirror of https://github.com/Significant-Gravitas/AutoGPT.git synced 2026-04-08 03:00:28 -04:00
Code Issues Packages Projects Releases Wiki Activity

best practice improvements

Browse Source
This commit is contained in:
SwiftyOS
2025-03-11 10:47:43 +01:00
committed by Swifty
parent 98a1adc397
commit a588cf1dc5
2 changed files with 67 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
autogpt_platform/backend/backend/blocks/example/_api.py
View File

@@ -8,6 +8,8 @@ This module provides a example of how to create a client for an API.
from json import JSONDecodeError
from typing import Any, Dict, Optional
from pydantic import BaseModel
from backend.data.model import APIKeyCredentials
# This is a wrapper around the requests library that is used to make API requests.
@@ -20,6 +22,16 @@ class ExampleAPIException(Exception):
self.status_code = status_code
class CreateResourceResponse(BaseModel):
message: str
is_funny: bool
class GetResourceResponse(BaseModel):
message: str
is_funny: bool
class ExampleClient:
"""Client for the Example API"""
@@ -41,7 +53,6 @@ class ExampleClient:
self._requests = Requests(
extra_headers=headers,
trusted_origins=["https://api.example.com"],
raise_for_status=False,
)
@@ -73,6 +84,8 @@ class ExampleClient:
response_data = response.json()
if "errors" in response_data:
# This is an example error and needs to be
# replaced with how the real API returns errors
error_messages = [
error.get("message", "") for error in response_data["errors"]
]
@@ -83,7 +96,7 @@ class ExampleClient:
return response_data
def get_resource(self, resource_id: str) -> Dict:
def get_resource(self, resource_id: str) -> GetResourceResponse:
"""
Fetches a resource from the Example API.
@@ -91,7 +104,7 @@ class ExampleClient:
resource_id: The ID of the resource to fetch.
Returns:
The resource data as a dictionary.
The resource data as a GetResourceResponse object.
Raises:
ExampleAPIException: If the API request fails.
@@ -100,13 +113,11 @@ class ExampleClient:
response = self._requests.get(
f"{self.API_BASE_URL}/resources/{resource_id}"
)
return self._handle_response(response)
except ExampleAPIException:
raise
return GetResourceResponse(**self._handle_response(response))
except Exception as e:
raise ExampleAPIException(f"Failed to get resource: {str(e)}", 500)
def create_resource(self, data: Dict) -> Dict:
def create_resource(self, data: Dict) -> CreateResourceResponse:
"""
Creates a new resource via the Example API.
@@ -114,15 +125,13 @@ class ExampleClient:
data: The resource data to create.
Returns:
The created resource data as a dictionary.
The created resource data as a CreateResourceResponse object.
Raises:
ExampleAPIException: If the API request fails.
"""
try:
response = self._requests.post(f"{self.API_BASE_URL}/resources", json=data)
return self._handle_response(response)
except ExampleAPIException:
raise
return CreateResourceResponse(**self._handle_response(response))
except Exception as e:
raise ExampleAPIException(f"Failed to create resource: {str(e)}", 500)

83
autogpt_platform/backend/backend/blocks/example/example.py
View File

@@ -1,13 +1,9 @@
import logging
from typing import Any
from backend.data.block import Block, BlockCategory, BlockOutput, BlockSchema, BlockType
from backend.data.model import (
APIKeyCredentials,
ContributorDetails,
CredentialsField,
SchemaField,
)
from pydantic import BaseModel
from backend.data.block import Block, BlockCategory, BlockOutput, BlockSchema
from backend.data.model import APIKeyCredentials, CredentialsField, SchemaField
from ._api import ExampleClient
from ._auth import TEST_CREDENTIALS, TEST_CREDENTIALS_INPUT, ExampleCredentialsInput
@@ -15,6 +11,11 @@ from ._auth import TEST_CREDENTIALS, TEST_CREDENTIALS_INPUT, ExampleCredentialsI
logger = logging.getLogger(__name__)
class GreetingMessage(BaseModel):
message: str
is_funny: bool
class ExampleBlock(Block):
class Input(BlockSchema):
@@ -25,7 +26,18 @@ class ExampleBlock(Block):
description="The greetings to display", default=["Hello", "Hi", "Hey"]
)
is_funny: bool = SchemaField(
description="Whether the block is funny", placeholder="True", default=True
description="Whether the block is funny",
placeholder="True",
default=True,
# Advanced fields are not shown in the UI by default
advanced=True,
)
greeting_context: str = SchemaField(
description="The context of the greeting",
placeholder="Enter a context",
default="The user is looking for an inspirational greeting",
# Hidden fields are not shown in the UI by default
hidden=True,
)
# Only if the block needs credentials
credentials: ExampleCredentialsInput = CredentialsField(
@@ -33,10 +45,10 @@ class ExampleBlock(Block):
)
class Output(BlockSchema):
response: dict[str, Any] = SchemaField(
response: GreetingMessage = SchemaField(
description="The response object generated by the example block."
)
all_responses: list[dict[str, Any]] = SchemaField(
all_responses: list[GreetingMessage] = SchemaField(
description="All the responses from the example block."
)
greeting_count: int = SchemaField(
@@ -52,9 +64,6 @@ class ExampleBlock(Block):
id="380694d5-3b2e-4130-bced-b43752b70de9",
# The description of the block, explaining what the block does.
description="The example block",
# The list of contributors who contributed to the block.
# Each contributor is represented by a ContributorDetails object.
contributors=[ContributorDetails(name="Craig Swift")],
# The set of categories that the block belongs to.
# Each category is an instance of BlockCategory Enum.
categories={BlockCategory.BASIC},
@@ -73,15 +82,21 @@ class ExampleBlock(Block):
# The list or single expected output if the test_input is run.
# Each output is a tuple of (output_name, output_data).
test_output=[
("response", {"message": "Hello, world!"}),
("response", {"message": "Hello, world!"}), # We mock the function
("response", {"message": "Hello, world!"}), # We mock the function
("response", GreetingMessage(message="Hello, world!", is_funny=True)),
(
"response",
GreetingMessage(message="Hello, world!", is_funny=True),
), # We mock the function
(
"response",
GreetingMessage(message="Hello, world!", is_funny=True),
), # We mock the function
(
"all_responses",
[
{"message": "Hello, world!"},
{"message": "Hello, world!"},
{"message": "Hello, world!"},
GreetingMessage(message="Hello, world!", is_funny=True),
GreetingMessage(message="Hello, world!", is_funny=True),
GreetingMessage(message="Hello, world!", is_funny=True),
],
),
("greeting_count", 3),
@@ -89,23 +104,18 @@ class ExampleBlock(Block):
# Function names on the block implementation to mock on test run.
# Each mock is a dictionary with function names as keys and mock implementations as values.
test_mock={
"my_function_that_can_be_mocked": lambda *args, **kwargs: "Hello, world!"
"my_function_that_can_be_mocked": lambda *args, **kwargs: GreetingMessage(
message="Hello, world!", is_funny=True
)
},
# The credentials required for testing the block.
# This is an instance of APIKeyCredentials with sample values.
test_credentials=TEST_CREDENTIALS,
# The type of the block, which is an instance of BlockType Enum.
block_type=BlockType.STANDARD,
)
@staticmethod
def my_static_method(post_greeting: str) -> str:
logger.info("my_static_method called with input: %s", post_greeting)
return f"Hello, {post_greeting}!"
def my_function_that_can_be_mocked(
self, name: str, credentials: APIKeyCredentials
) -> str:
) -> GreetingMessage:
logger.info("my_function_that_can_be_mocked called with input: %s", name)
# Use the ExampleClient from _api.py to make an API call
@@ -113,8 +123,10 @@ class ExampleBlock(Block):
# Create a sample resource using the client
resource_data = {"name": name, "type": "greeting"}
response = client.create_resource(resource_data)
return f"API response: {response.get('message', 'Hello, world!')}"
# If your API response object matches the return type of the function,
# there is no need to convert the object. In this case we have a different
# object type for the response and the return type of the function.
return GreetingMessage(**client.create_resource(resource_data).model_dump())
def run(
self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs
@@ -130,14 +142,13 @@ class ExampleBlock(Block):
node_exec_id: The ID of the current node execution.
user_id: The ID of the user executing the block.
"""
rtn_all_responses: list[dict[str, Any]] = []
rtn_all_responses: list[GreetingMessage] = []
# Here we deomonstrate best practice for blocks that need to yield multiple items.
# We yield each item from the list to allow for operations on each element.
# We also yield the complete list for situations when the full list is needed.
for greeting in input_data.greetings:
full_greeting = ExampleBlock.my_static_method(greeting)
message = self.my_function_that_can_be_mocked(full_greeting, credentials)
rtn_all_responses.append({"message": message})
yield "response", {"message": message}
message = self.my_function_that_can_be_mocked(greeting, credentials)
rtn_all_responses.append(message)
yield "response", message
yield "all_responses", rtn_all_responses
yield "greeting_count", len(input_data.greetings)