Scenario Structure

Section is a comprehensive guide to constructing and managing scenarios within the Aiviro package. At its core is the BaseScenario class, facilitating scenario initialization, configuration, and execution.

This section provides insights into key elements such as configuration retrieval, logging, error handling, robot instantiation, and email operations. Additionally, it covers essential methods like scenario setup, execution, and cleanup.

The section also introduces decorators and exceptions to enhance scenario structure, including step-wise execution step().

Use this section to understand the fundamental structure of Aiviro scenarios and efficiently build, configure, and manage your automation workflows.

class aiviro.modules.template.BaseScenario(config: YAMLConfig, logging_enabled: bool = True)

Base object to inherit from for scenario initialization. It helps to create robots and other basic objects based on the provided yaml-config file.

Parameters:

• config – configuration file containing all necessary parameters

• logging_enabled – if True then logging will be initialized

Example:

>>> import aiviro
>>> from aiviro.modules.config import YAMLConfig
>>> from src.story01.helios import HeliosHandler
>>>
>>> class Story01(aiviro.BaseScenario):
...     def __init__(self, yaml_config: YAMLConfig):
...         super().__init__(config=yaml_config)
...         self._helios_handler = HeliosHandler()
...
...     def _before_run(self):
...         # some logic before scenario run
...         pass
...
...     def _after_run(self):
...         # some logic after scenario run
...         pass
...
...     def _run(self):
...         r = self.robot("rdp")
...         r.click(aiviro.Button("Start"))
...         # additional logic of the scenario
>>>
>>> if __name__ == "__main__":
...     # initialize and merge configs
...     main_config = YAMLConfig("main_config.yaml")
...     story_config = YAMLConfig("story01.yaml")
...
...     # initialize scenario with merged config
...     story = Story01(main_config | story_config)
...     story.start()
...     story.close()
...
...     # you can also use scenario with context manager which will close the scenario automatically
...     with Story01(main_config | story_config) as story:
...         story.start()

property config: YAMLConfig

Returns configuration object.

property logger: AiviroLoggerAdapter

Returns main logger of the scenario.

property error_handler: ErrorHandler | None

Returns error handler object.

>>> import aiviro
>>> class Story01(aiviro.BaseScenario):
...     def _run(self):
...         r = self.robot("rdp")
...         r.click(aiviro.Button("Start"))
...         try:
...             r.click(aiviro.Button("Stop"))
...         except aiviro.SearchObjectError as e:
...             self.error_handler.collect(e)

property steps_run_handler: StepsRunHandler

Returns steps run handler object.

property steps_tree: str

Returns tree structure of the steps run by the scenario.

[0] "__root__" (1 ❌) 1/3
├── [1] "_process_data_to_helios" (1 ❌) 1/2
│   ├── [1-1] "_start_helios" (0 ✅) 1/1
│   │   └── [1-1-1] "read_data" (0 ✅)
│   └── [1-2] "process_invoice_header" (1 ❌) 3/4 exp=AlreadyExist('my exception message')
│       ├── [1-2-1] "_one_order_process" (0 ✅)
│       ├── [1-2-2] "_one_order_process" (0 ✅)
│       ├── [1-2-3] "_one_order_process" (0 ✅)
│       └── [1-2-4] "_one_order_process" (1 ❌) exp=AlreadyExist('my exception message')
├── [2] "_process_data_to_helios" (0 ✅) 2/2
│   ├── [2-1] "_start_helios" (0 ✅) 1/1
│   │   └── [2-1-1] "read_data" (0 ✅)
│   └── [2-2] "process_invoice_header" (0 ✅) 6/6
│       ├── [2-2-1] "_one_order_process" (0 ✅)
│       ├── [2-2-2] "_one_order_process" (0 ✅)
│       ├── [2-2-3] "_one_order_process" (0 ✅)
│       ├── [2-2-4] "_one_order_process" (0 ✅)
│       ├── [2-2-5] "_one_order_process" (0 ✅)
│       └── [2-2-6] "_one_order_process" (0 ✅)
└── [3] "_process_data_to_helios" (1 ❌) 1/2
    ├── [3-1] "_start_helios" (0 ✅) 1/1
    │   └── [3-1-1] "read_data" (0 ✅)
    └── [3-2] "process_invoice_header" (0 ✅) 2/2
        ├── [3-2-1] "_one_order_process" (0 ✅)
        └── [3-2-2] "_one_order_process" (0 ✅)

property tmp_folder: TemporaryFileStorage

Returns temporary folder object.

robot(robot_name: str) → DesktopRobot | WebRobot | RDPRobot

Returns robot by its name. Supported robots are DesktopRobot RDPRobot WebRobot. Name and type of the variables must correspond to the arguments of create_*_robot methods, see Create Robot section.

Parameters:

robot_name – name of the robot

robot:
  orion:
    type: rdp
    remote_address: 10.128.10.20
    username: robot-username
    password: secret-password
    shared_folder: !!python/tuple
        - __auto_folder  # automatically creates temporary folder
        - folder_name_on_remote_machine

robot:
  bender:
    type: rdp
    remote_address: 10.128.32.51
    username: robot-username
    password: secret-password
    domain: company
    auto_reconnect: True
    shared_folder: !!python/tuple
        - /home/robot/shared  # predefined folder
        - my_folder
  anicka:
    type: rdp
    remote_address: 10.128.32.42
    username: robot-username-2
    password: secret-password-2
    domain: company

email_client() → EmailClient

Returns email client instance. Arguments and auth key of the SMTP or IMAP authorization method must correspond to the following methods:

• setup_imap_basic_auth()

• setup_imap_outlook_oauth2_public()

• setup_imap_outlook_oauth2_confidential()

• setup_smtp_basic_auth()

• setup_smtp_outlook_oauth2_public()

• setup_smtp_outlook_oauth2_confidential()

email:
  client:
    smtp:
      auth: basic_auth
      server: smtp-mail.outlook.com
      username: robot
      password: password
      sender_name: Aiviro Robot
      email_address: robot@aiviro.com
    imap:
      auth: outlook_oauth2_confidential
      username: user@name.com
      client_id: 1234567890
      authority_url: https://login.microsoftonline.com/directory-id
      secret: 1234-asdf-5678-zxcv-9012

set_email_extractor_params(attachment_conditions: List[BaseCondition]) → None

Sets additional parameters for email extractor.

Parameters:

attachment_conditions – argument passed to email-extractor constructor

email_extractor() → EmailExtractor

Returns email extractor instance. Name and type of the variables must correspond to the constructor arguments of EmailExtractor class. For setting additional parameters, use set_email_extractor_params() method.

email:
  extractor:
    source_dir: robot/invoices
    processed_dir: robot/processed
    unprocessed_dir: robot/unprocessed
    max_valid_emails: 10

email_notifier() → EmailNotifier

Returns email notifier instance. Name and type of the variables must correspond to the constructor arguments of EmailNotifier class.

email:
  notifier:
    report_title: My Super Title
    recipients:
      - r1@cmp.com
      - r2@cmp.com
      - r3@cmp.com

start()

Starts the scenario. First calls _before_run(), then _run() and finally _after_run().

close()

Closes all initialized objects.

aiviro.modules.template.step(__fn: Callable | None = None, *, name: str | None = None, description: str | None = None, metric: str | None = None) → Callable

Decorator for marking function as a step in scenario. Use it to separate your scenario into logical blocks.

Parameters:

• name – Name of the step. If not provided, function name will be used.

• description – Description of the step.

• metric – Name of the metric to count with each step call.

Example:

>>> from aiviro import BaseScenario, step, get_run_context, StepWarningException
>>>
>>> class MyScenario(BaseScenario):
...     def _run(self):
...         self.prepare_data()
...         self.import_data()
...
...     @step
...     def prepare_data(self):
...         pass
...
...     @step(name="Import data into system")
...     def import_data(self):
...         for i in range(10):
...             get_run_context().logger.info(f"Processing item {i}")
...             self.process_item(item=i)
...
...     @step(
...         name="Process item",
...         description="Process item and check if it is even",
...         metric="even_items",
...     )
...     def process_item(self, item: int):
...         if item % 2 == 0:
...             raise StepWarningException(f"Item {item} is even")
...
...     @staticmethod
...     @step
...     def static_step():
...         pass

>>> class MyScenario2(BaseScenario):
...     def _run(self):
...         try:
...             self.step_1()
...         except Exception as e:
...             er_step = get_run_context().error_step
...             print(f"Error step: {er_step.name} - {er_step.description}")
...             # Output: Error step: step_11 - This is step with exception
...
...         self.step_2()
...
...     @step
...     def step_1(self):
...         self.step_11()
...
...     @step
...     def step_2(self):
...         pass
...
...     @step(description="This is step with exception")
...     def step_11(self):
...         raise RuntimeError("My exception raise")

exception aiviro.modules.template.StepWarningException

Exception to inherit from when a step should be marked as a warning.

aiviro.modules.template.get_run_context() → RunContext

Returns context info about currently running scenario and step. Context contains information about the scenario and step that is currently running. You can retrieve step name, id, run id and logger from the context.

Raises:

RuntimeError – if no scenario is running

class aiviro.modules.template.RunContext(_scenario: 'BaseScenario', _step: 'StepRun')

property scenario: BaseScenario

Returns currently running scenario.

property step: StepRun

Returns currently running step.

property error_step: StepRun | None

Returns the deepest step that has failed or crashed, if exists.

property logger: Logger

Returns logger for the current step run, or context logger if no step is running.

class aiviro.core.utils.global_context.storage.TemporaryFileStorage

property root_path: Path

Return the root path of the temporary directory.

property unique_folder: Path

Create a new unique sub-folder in the temporary directory.

close()

Cleanup the temporary directory.