Skip to content

yaml_workflow.tasks.basic_tasks

yaml_workflow.tasks.basic_tasks

Basic task functions for demonstration and testing.

Classes

Functions

add_numbers(a: float, b: float) -> float

Add two numbers together.

Parameters:

Name Type Description Default
a float

First number

required
b float

Second number

required

Returns:

Name Type Description
float float

Sum of the numbers

Source code in src/yaml_workflow/tasks/basic_tasks.py
@register_task()
def add_numbers(a: float, b: float) -> float:
    """
    Add two numbers together.

    Args:
        a: First number
        b: Second number

    Returns:
        float: Sum of the numbers
    """
    return a + b

create_greeting(name: str, context: Dict[str, Any]) -> str

Create a greeting message.

Parameters:

Name Type Description Default
name str

Name to greet

required
context Dict[str, Any]

Template context

required

Returns:

Name Type Description
str str

Greeting message

Raises:

Type Description
TemplateError

If template resolution fails

Source code in src/yaml_workflow/tasks/basic_tasks.py
@register_task()
def create_greeting(name: str, context: Dict[str, Any]) -> str:
    """
    Create a greeting message.

    Args:
        name: Name to greet
        context: Template context

    Returns:
        str: Greeting message

    Raises:
        TemplateError: If template resolution fails
    """
    try:
        template = Template("Hello {{ name }}!", undefined=StrictUndefined)
        return template.render(name=name, **context)
    except Exception as e:
        raise TemplateError(f"Failed to create greeting: {str(e)}")

echo(message: str) -> str

Echo back the input message.

Parameters:

Name Type Description Default
message str

Message to echo

required

Returns:

Name Type Description
str str

The input message

Source code in src/yaml_workflow/tasks/basic_tasks.py
@register_task()
def echo(message: str) -> str:
    """
    Echo back the input message.

    Args:
        message: Message to echo

    Returns:
        str: The input message
    """
    return message

fail(message: str = 'Task failed') -> None

A task that always fails.

Parameters:

Name Type Description Default
message str

Error message

'Task failed'

Raises:

Type Description
RuntimeError

Always raises this error

Source code in src/yaml_workflow/tasks/basic_tasks.py
@register_task()
def fail(message: str = "Task failed") -> None:
    """
    A task that always fails.

    Args:
        message: Error message

    Raises:
        RuntimeError: Always raises this error
    """
    raise RuntimeError(message)

hello_world(name: str = 'World') -> str

A simple hello world function.

Parameters:

Name Type Description Default
name str

Name to include in greeting. Defaults to "World".

'World'

Returns:

Name Type Description
str str

The greeting message

Source code in src/yaml_workflow/tasks/basic_tasks.py
@register_task()
def hello_world(name: str = "World") -> str:
    """
    A simple hello world function.

    Args:
        name: Name to include in greeting. Defaults to "World".

    Returns:
        str: The greeting message
    """
    return f"Hello, {name}!"

join_strings(*strings: str, separator: str = ' ') -> str

Join multiple strings together.

Parameters:

Name Type Description Default
*strings str

Variable number of strings to join

()
separator str

String to use as separator. Defaults to space.

' '

Returns:

Name Type Description
str str

Joined string

Source code in src/yaml_workflow/tasks/basic_tasks.py
@register_task()
def join_strings(*strings: str, separator: str = " ") -> str:
    """
    Join multiple strings together.

    Args:
        *strings: Variable number of strings to join
        separator: String to use as separator. Defaults to space.

    Returns:
        str: Joined string
    """
    return separator.join(strings)