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 >>> >>> class Story01(aiviro.BaseScenario): ... 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()
- 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 ✅)
- 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 ofcreate_*_robot
methods, see Create Robot section.- Parameters:
robot_name – name of the robot
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 - 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: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, useset_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) 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.
- 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") ... def process_item(self, item: int): ... if item % 2 == 0: ... raise StepWarningException(f"Item {item} is even") ... ... @staticmethod ... @step ... def static_step(): ... pass
- 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_id: str
Returns id of the current step. This id is unique for each step in the scenario.