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.BaseScenario(config: YAMLConfig, logging_enabled: bool = True, global_context: GlobalContext | None = None)

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

• global_context – global context to use, if not provided, new one will be created automatically

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

Note

For RDP robot, you can use special shared folder __auto_folder which automatically creates a temporary folder for the robot.

Note

New version of RDP robot configuration is supported, for create_rdp_robot_v2() method. To use it, set version: 2 in the robot’s configuration.

robot:
  bender:
    type: rdp
    version: 2
    server:
      host: 10.128.2.31
      username: robot-username
      password: secret-password
      domain: company
      port: 3389
    shared_folder:
      path: __auto_folder
      alias: folder_name_on_remote_machine
    gateway:
      host: 10.129.5.56
      username: gateway-username
      password: secret-password
      domain: company
      port: 3389
    enable_auto_reconnect: True

robot:
  orion:
    type: rdp
    remote_address: 10.128.32.51
    username: robot-username
    password: secret-password
    domain: company
    auto_reconnect: True
    shared_folder: !!python/tuple
        - __auto_folder  # automatically creates temporary folder
        - folder_name_on_remote_machine
  anicka:
    type: rdp
    remote_address: 10.128.32.42
    username: robot-username-2
    password: secret-password-2
    domain: company
    shared_folder: !!python/tuple
        - /home/robot/shared

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

report_builder(email_notifier_configuration: bool = False) → ReportBuilder

Returns report builder instance.

Parameters:

email_notifier_configuration – if True, report title, logging and notification titles are taken from email notifier configuration

report_builder:
  title: Invoice processing Workflow
  message_titles:  # optional, custom titles for the messages, based on their status code
    0: not-active
    10: great success
  logging_level: DEBUG  # optional, logging level using which every message is logged
  metrics_to_display:  # optional, set of sub-metrics to display in the performance metrics section
    - api_reader_pages
    - email_extracted
    - actions_used
    - my_custom_metric

send_report_via_email_client(recipients: ADDR_TYPE, attachments: list[str | Path] | None = None, send_if_empty: bool = False) → str

Sends report via email client. The report is generated by using ReportBuilder.

Parameters:

• recipients – E-mail recipients of the report

• attachments – List of paths to attach to the email

• send_if_empty – If True, report is sent even if there are no messages in the report

Returns:

generated HTML report

Example:

>>> import aiviro
>>> from aiviro.modules.config import YAMLConfig
>>>
>>> class MyScenario(aiviro.BaseScenario):
...     def __init__(self, config: YAMLConfig):
...         super().__init__(config)
...         self.report = self.report_builder()
...
...     def _run(self):
...         self.report.successful("Task completed successfully")
...
...     def _after_run(self):
...         self.send_report_via_email_client(self.config.recipients)

email_notifier() → EmailNotifier

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

Warning

Email notifier is deprecated and will be removed in future version. Please use instead report_builder() to create and send reports. You can reuse the configuration from email notifier, but we recommend to change the configuration to match the new report builder.

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.step(__fn: Callable | None = None, *, name: str | None = None, description: str | None = None, metric: StepMetric | 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 – Union of ‘StepMetric’ containing name and type of the metric, or str (only for backward compatibility) for the metric to count with each step call. E.g. metric=StepMetric(name=”my_metric”, type=”success_rate”)

Example:

>>> from aiviro import (
...     BaseScenario,
...     step,
...     get_run_context,
...     StepMetric,
...     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=StepMetric(name="even_items", type="counter"),
...     )
...     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")

aiviro.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.core.utils.metrics.StepMetric(*, name: str, display_name: str | None = None, type: Literal['counter', 'success_rate'] = 'counter')

Custom metric for the step.

Parameters:

• name – name of the metric, used for aggregation

• display_name – name of the metric to be displayed in the report, if not set, name is used

• type – type of the metric, has to be either “counter” or “success_rate”. Counter represents simple number of occurrences, success_rate represents the ratio of successful occurrences to the total number of occurrences.

class aiviro.core.utils.global_context.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.TemporaryFileStorage

A temporary file storage that creates a temporary directory and cleans it up on exit.

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.