fix(backend): Fix iteration and timer block

This commit is contained in:
Zamil Majdy
2024-11-08 16:53:26 +07:00
parent 91edf08540
commit 57ddeb4574
2 changed files with 48 additions and 24 deletions

View File

@@ -2,13 +2,28 @@ from typing import Any
from backend.data.block import Block, BlockCategory, BlockOutput, BlockSchema
from backend.data.model import SchemaField
from backend.util.json import json
class StepThroughItemsBlock(Block):
class Input(BlockSchema):
items: list | dict = SchemaField(
items: list = SchemaField(
advanced=False,
description="The list or dictionary of items to iterate over",
placeholder="[1, 2, 3, 4, 5] or {'key1': 'value1', 'key2': 'value2'}",
default=[],
)
items_object: dict = SchemaField(
advanced=False,
description="The list or dictionary of items to iterate over",
placeholder="[1, 2, 3, 4, 5] or {'key1': 'value1', 'key2': 'value2'}",
default={},
)
items_str: str = SchemaField(
advanced=False,
description="The list or dictionary of items to iterate over",
placeholder="[1, 2, 3, 4, 5] or {'key1': 'value1', 'key2': 'value2'}",
default="",
)
class Output(BlockSchema):
@@ -39,14 +54,20 @@ class StepThroughItemsBlock(Block):
)
def run(self, input_data: Input, **kwargs) -> BlockOutput:
items = input_data.items
if isinstance(items, dict):
# If items is a dictionary, iterate over its values
for item in items.values():
yield "item", item
yield "key", item
else:
# If items is a list, iterate over the list
for index, item in enumerate(items):
yield "item", item
yield "key", index
for data in [input_data.items, input_data.items_object, input_data.items_str]:
if not data:
continue
if isinstance(data, str):
items = json.loads(data)
else:
items = data
if isinstance(items, dict):
# If items is a dictionary, iterate over its values
for item in items.values():
yield "item", item
yield "key", item
else:
# If items is a list, iterate over the list
for index, item in enumerate(items):
yield "item", item
yield "key", index

View File

@@ -139,23 +139,26 @@ class GetCurrentDateAndTimeBlock(Block):
class CountdownTimerBlock(Block):
class Input(BlockSchema):
input_message: Any = SchemaField(
input_data: Any = SchemaField(
advanced=False,
description="Message to output after the timer finishes",
default="timer finished",
)
seconds: Union[int, str] = SchemaField(
description="Duration in seconds", default=0
advanced=False, description="Duration in seconds", default=0
)
minutes: Union[int, str] = SchemaField(
description="Duration in minutes", default=0
advanced=False, description="Duration in minutes", default=0
)
hours: Union[int, str] = SchemaField(
advanced=False, description="Duration in hours", default=0
)
days: Union[int, str] = SchemaField(
advanced=False, description="Duration in days", default=0
)
hours: Union[int, str] = SchemaField(description="Duration in hours", default=0)
days: Union[int, str] = SchemaField(description="Duration in days", default=0)
class Output(BlockSchema):
output_message: str = SchemaField(
description="Message after the timer finishes"
)
output_data: Any = SchemaField(description="Message after the timer finishes")
def __init__(self):
super().__init__(
@@ -166,11 +169,11 @@ class CountdownTimerBlock(Block):
output_schema=CountdownTimerBlock.Output,
test_input=[
{"seconds": 1},
{"input_message": "Custom message"},
{"input_data": "Custom message"},
],
test_output=[
("output_message", "timer finished"),
("output_message", "Custom message"),
("output_data", "timer finished"),
("output_data", "Custom message"),
],
)
@@ -183,4 +186,4 @@ class CountdownTimerBlock(Block):
total_seconds = seconds + minutes * 60 + hours * 3600 + days * 86400
time.sleep(total_seconds)
yield "output_message", input_data.input_message
yield "output_data", input_data.input_data