improved the example

This commit is contained in:
SwiftyOS
2025-03-02 17:19:24 +01:00
committed by Swifty
parent 1aae4e7474
commit c26a96fc6d
4 changed files with 18 additions and 25 deletions

View File

@@ -1,5 +1,10 @@
from __future__ import annotations
"""
API module for Example API integration.
This module provides a example of how to create a client for an API.
"""
# We also have a Json Wrapper library available in backend.util.json
from json import JSONDecodeError
from typing import Any, Dict, Optional
@@ -22,23 +27,6 @@ def _get_headers(credentials: APIKeyCredentials) -> dict[str, str]:
}
def get_api(credentials: APIKeyCredentials) -> Requests:
"""
Creates a configured Requests instance for the Example API.
Args:
credentials: The Example API credentials to use for authentication.
Returns:
A Requests instance configured with the appropriate headers and trusted origins.
"""
return Requests(
trusted_origins=["https://api.example.com"],
extra_headers=_get_headers(credentials),
raise_for_status=False,
)
class ExampleClient:
"""Client for the Example API"""

View File

@@ -84,6 +84,7 @@ class ExampleBlock(Block):
{"message": "Hello, world!"},
],
),
("greeting_count", 3),
],
# 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.

View File

@@ -8,7 +8,7 @@ from backend.data.block import (
BlockSchema,
)
from backend.data.model import SchemaField
from backend.integrations.webhooks.example import ExampleWebhookType
from backend.integrations.webhooks.example import ExampleWebhookEventType
logger = logging.getLogger(__name__)
@@ -42,7 +42,7 @@ class ExampleTriggerBlock(Block):
# It defines which external service can trigger this block and what type of events it responds to
webhook_config=BlockManualWebhookConfig(
provider="example_provider", # The external service that will send webhook events
webhook_type=ExampleWebhookType.EXAMPLE, # The specific event type this block responds to
webhook_type=ExampleWebhookEventType.EXAMPLE_EVENT, # The specific event type this block responds to
),
# Test input for trigger blocks should mimic the payload structure that would be received from the webhook
test_input=[
@@ -53,9 +53,13 @@ class ExampleTriggerBlock(Block):
}
}
],
test_output=[
("event_data", {"event_type": "example", "data": "Sample webhook data"})
],
)
def run(self, input_data: Input, **kwargs) -> BlockOutput:
# For trigger blocks, the run method is called automatically when a webhook event is received
# The payload from the webhook is passed in as input_data.payload
logger.info("Example trigger block run with payload: %s", input_data.payload)
yield "event_data", input_data.payload

View File

@@ -13,9 +13,9 @@ from ._manual_base import ManualWebhookManagerBase
logger = logging.getLogger(__name__)
class ExampleWebhookType(StrEnum):
EXAMPLE = "example"
EXAMPLE_2 = "example_2"
class ExampleWebhookEventType(StrEnum):
EXAMPLE_EVENT = "example_event"
ANOTHER_EXAMPLE_EVENT = "another_example_event"
# ExampleWebhookManager is a class that manages webhooks for a hypothetical provider.
@@ -24,7 +24,7 @@ class ExampleWebhookManager(ManualWebhookManagerBase):
# Define the provider name for this webhook manager.
PROVIDER_NAME = ProviderName.EXAMPLE_PROVIDER
# Define the types of webhooks this manager can handle.
WebhookType = ExampleWebhookType
WebhookEventType = ExampleWebhookEventType
BASE_URL = "https://api.example.com"
@@ -45,7 +45,7 @@ class ExampleWebhookManager(ManualWebhookManagerBase):
# Extract the JSON payload from the request.
payload = await request.json()
# Set the event type based on the webhook type in the payload.
event_type = payload.get("webhook_type", ExampleWebhookType.EXAMPLE)
event_type = payload.get("webhook_type", ExampleWebhookEventType.EXAMPLE_EVENT)
return payload, event_type