Creating and managing Playbook Bricks in Brick Management
Purpose | This documentation outlines key factors to take into consideration when creating and configuring Playbook Bricks. |
---|
Created | November 21, 2023 |
---|
To create a Playbook Brick in Brick Management, follow these steps:
- Navigate to Settings.
- Select Brick Management v2.
- Click on the "CREATE BRICK" button located at the top right corner.
- Fill in the necessary fields.
About Playbook Bricks
Playbook Bricks help with the automation of tasks within the workflow. In the following sections you will find the different dependencies needed to create your playbook Brick.
Dependencies
Ensure the following dependencies are available:
from playbook import BaseBrick, RoutedTask
# If you are using Secrets:
from rivendel import Secrets
Base Brick
The BaseBrick
class provides a foundation for Playbook Bricks. Any Playbook Brick class should inherit from this base class.
Adding Parameters
To customize Playbook Bricks, create a Brick class derived from the base brick and introduce parameters. Parameters can take different forms, including string, boolean, Secrets, and more.
rivendel
Library
The rivendel library is the primary import library for the management of Playbook Bricks.
playbooks.RoutedTask
Playbooks yield routed tasks, including session ID, events, state, and variables. With this feature they are able to provide insights into the ongoing state and variables.
Class Methods
Initialization Method (init
)
The init
method initializes an instance of the Playbook Brick class.
class Brick(BaseBrick):
def init(self, config):
self.config = config
-
Parameters:
self
: The instance of the class being created.config
: A parameter representing the configuration settings for the Playbook Brick. It is passed to the method when a new instance of the class is created.
run
Method
The run
method defines the logic of the Playbook Brick. It takes a task
parameter along with optional in_nodes
and out_nodes
. The method performs validation on task.event
based on a JSON schema loaded from self.config.schema
. If the validation passes, it returns a new RoutedTask
object.
def run(self, task, in_nodes=[], out_nodes=[], *args, **kwargs):
import json
schema = json.loads(self.config.schema)
validate(instance=task.event, schema=schema)
return RoutedTask(
session_id=task.session_id,
event=task.event,
state=task.state,
variables={**task.variables},
nodes=out_nodes,
)
-
Parameters:
self
: The instance of the class that the method is being called on.task
: A parameter representing the task for the Playbook Brick. It contains information such as session ID, event data, state, and variables.in_nodes
(optional): A list of input nodes, which are used for defining the input dependencies for the task.out_nodes
(optional): A list of output nodes, which are used for defining the output dependencies for the task.*args
and**kwargs
(optional): These allow the method to accept a variable number of positional and keyword arguments, respectively.
-
Returns:
- The method returns a
RoutedTask
object, which is an instance of theRoutedTask
class. This object encapsulates the result of the Playbook Brick's execution, including session ID, event data, state, variables, and output nodes.
- The method returns a
Data Management
Playbook Bricks are responsible for overseeing the flow of data, both incoming and outgoing. They define the structure of input and output data, ensuring an easy data flow within the workflow.
Example of Playbook Brick
import json
from jsonschema import validate
from playbook import RoutedTask, BaseBrick
class Brick(BaseBrick):
def __init__(self, config):
self.config = config
def run(self, task, in_nodes=[], out_nodes=[], *args, **kwargs):
schema = json.loads(self.config.schema)
validate(instance=task.event, schema=schema)
return RoutedTask(
session_id=task.session_id,
event=task.event,
state=task.state,
variables={**task.variables},
nodes=out_nodes,
)
def next_nodes(self):
return {}
# The following example showcases a Playbook Brick designed to validate JSON data.
# If the validation succeeds, it passes to the next task.
# The `run` method performs the validation based on a JSON schema, and the `next_nodes` method
# currently returns an empty dictionary as a placeholder for defining next nodes.