Rules

opengate_data.rules.rules

RulesBuilder

RulesBuilder Objects

class RulesBuilder()

Rules builder

with_organization_name

def with_organization_name(organization_name: str) -> "RulesBuilder"

Specify the organization name.

Arguments:

organization_name str - The name of the organization.

Returns:

RulesBuilder - Returns self for chaining.

Example:

builder.with_organization_name('organization_name')

with_identifier

def with_identifier(identifier: str) -> "RulesBuilder"

Specify the identifier for the rule.

Arguments:

identifier str - The identifier for the rules.

Returns:

RulesBuilder - Returns self for chaining.

Example:

builder.with_identifier('identifier')

with_actions

def with_actions(actions: dict[str, Any]) -> "RulesBuilder"

Specify the actions in rules.

Arguments:

actions dict - Actions

Returns:

RulesBuilder - Returns self for chaining.

Example:

rule = {
- `"actions"` - {
- `"close"` - [
{
- `"enabled"` - False,
- `"ruleToClose"` - "rule_name",
- `"alarmToClose"` - "alarmToClose"
}
]
}
}
with_actions = rule["actions"]
builder.with_actions(with_actions)

with_actions_delay

def with_actions_delay(actions_delay: int) -> "RulesBuilder"

Waiting threshold before actions are executed. Allows cancellation of the execution of actions if another rule exists with a subsequent delay cancellation action.

Arguments:

actions_delay int - Delay option in milliseconds

Returns:

RulesBuilder - Returns self for chaining.

Example:

builder.with_actions(1000)

with_active

def with_active(active: bool) -> "RulesBuilder"

Specify the active in rules.

Arguments:

active bool - Activate or deactivate action

Returns:

RulesBuilder - Returns self for chaining.

Example:

builder.with_active(False)
builder.with_active(True)

with_channel

def with_channel(channel: str) -> "RulesBuilder"

Specify the channel in rules.

Arguments:

channel str - Channel

Returns:

RulesBuilder - Returns self for chaining.

Example:

builder.with_channel('default_channel')

with_condition

def with_condition(condition: dict[str, Any]) -> "RulesBuilder"

Specify the condition for the rule. Mandatory for EASY mode and not applicable for ADVANCED mode. JSON filter which follows the same filter structure of the Opengate platform. It can contain rule parameters.

Arguments:

condition dict - Specify the identifier for the rule.

Returns:

RulesBuilder - Returns self for chaining.

Example:

filter_builder_build = FilterBuilder().and_(FilterBuilder().eq("provision.administration.organization","Organization")).build()
builder.with_condition('filter_builder_build)
builder.with_condition({device.cpu.usage._current.value:
"$datastream:device.cpu.usage._current.value"})

with_mode

def with_mode(mode: str) -> "RulesBuilder"

Specify rule type mode

Arguments:

mode str - Advanced or basic.

Returns:

RulesBuilder - Returns self for chaining.

Example:

builder.with_mode('EASY')
builder.with_mode('ADVANCED')

with_name

def with_name(name: str) -> "RulesBuilder"

Specify the name for the rule.

Arguments:

name str - The identifier for the rules.

Returns:

RulesBuilder - Returns self for chaining.

Example:

builder.with_mode('name_rule')

with_description

def with_description(description: str) -> "RulesBuilder"

Specify the description.

Arguments:

description str - The description for the rules.

Returns:

RulesBuilder - Returns self for chaining.

Example:

builder.with_organization_name('description')

with_type

def with_type(rule_type: dict[str, Any]) -> "RulesBuilder"

Specify the description.

Arguments:

rule_type dict - The description for the rules.

Returns:

RulesBuilder - Returns self for chaining.

Example:

- `"rule"` - {
- `"type"` - {
- `"datastreams"` - [
{
- `"name"` - "device.cpu.usage",
- `"fields"` - [
{
- `"field"` - "value",
- `"alias"` - "CPU usage"
}
],
- `"prefilter"` - False
}
],
- `"name"` - "DATASTREAM"
}
}
type = rule["type"]
builder.with_type(type)

with_parameters

def with_parameters(parameters: list[dict[str, str]]) -> "RulesBuilder"

Specify the parameters for rules.

Arguments:

parameters dict - The parameters for the rules.

Returns:

RulesBuilder - Returns self for chaining.

Example:

parameters = [
{
- `"name"` - "name",
- `"schema"` - "string",
- `"value"` - "2"
}
]
builder.with_parameters(parameters)

with_code

def with_code(code: str) -> "RulesBuilder"

Specify the JavaScript code for advanced rules.

Arguments:

code str - The JavaScript code to be used in the rule.

Returns:

RulesBuilder - Returns self for chaining.

Raises:

Exception - If the mode is not set to ‘ADVANCED’.

Example:

code = '''
function add(a, b) {
// This is a comment
return a + b;
}
'''
builder.with_mode('ADVANCED').with_code(code).with_code_file('code')

with_code_file

def with_code_file(code_file: str) -> "RulesBuilder"

Specify the JavaScript code for advanced rules from a file.

This method reads the JavaScript code from a specified file and converts it into a single line to be used in the rule.

Arguments:

code_file str - The path to the file containing the JavaScript code.

Returns:

RulesBuilder - Returns self for chaining.

Raises:

ValueError - If the file does not exist or is not a valid file.
ValueError - If the mode is not set to ‘ADVANCED’.

Example:

builder.with_mode('ADVANCED').with_code_file('path/to/code.js')

with_env

def with_env(data_env: str) -> "RulesBuilder"

Specify the environment variable.

Arguments:

data_env str - The environment variable.

Returns:

RulesBuilder - Returns self for chaining.

Example:

builder.with_env('RULE_ID')

with_config_file

def with_config_file(config_file: str, section: str,
                     config_key: str) -> "RulesBuilder"

Sets up the configuration file (.ini).

This method allows specifying a configuration file, a section within that file, and a key to retrieve a specific value from the section.

Arguments:

config_file str - The path to the.ini configuration file.
section str - The section name within the.ini file where the desired configuration is located.
config_key str - The key within the specified section whose value will be retrieved.

Returns:

RulesBuilder - Returns itself to allow for method chaining.

Example:

[id]
rule_id = afe07216-14ec-4134-97ae-c483b11d965a
builder.with_config_file('model_config.ini', 'id', 'rule_id')

with_find_by_name

def with_find_by_name(find_name: str) -> "RulesBuilder"

Specify the name to find.

Arguments:

find_name str - The name of the pipeline.

Returns:

RulesBuilder - Returns self for chaining.

Example:

builder.with_find_by_name('name_rule')

find_all

def find_all() -> "RulesBuilder"

Retrieve all models.

This method sets up the RulesBuilder instance to retrieve all rules associated with the specified organization.

Returns:

RulesBuilder - The instance of the RulesBuilder class itself, allowing for method chaining.

Example:

builder.with_organization_name("organization_name").find_all()

find_one

def find_one() -> "RulesBuilder"

Retrieve all models.

This method sets up the RulesBuilder instance to retrieve all rules associated with the specified organization.

Returns:

RulesBuilder - The instance of the RulesBuilder class itself, allowing for method chaining.

Example:

builder.with_organization_name('organization').with_channel(
'default_channel'
).with_identifier('4ae733b0-2dc6-4ad4-9d2c-9ab426c9f32d').find_one().build().execute()

create

def create() -> "RulesBuilder"

Initiates the creation process of a new model.

This method prepares the RulesBuilder instance to create a new model by setting up the necessary parameters such as the organization name and the file to be associated with the model. It also specifies the URL endpoint for creating the model and sets the operation type to ‘create’.

Returns:

RulesBuilder - The instance of the RulesBuilder class itself, allowing for method chaining.

Example:

builder.with_organization_name("organization").with_channel(
"default_channel"
).with_name("rule_name").with_mode("EASY").create().build().execute()

update

def update() -> "RulesBuilder"

Update an existing rule.

This method sets up the RulesBuilder instance to update a specific rule associated with the specified organization and identifier.

Returns:

RulesBuilder - The instance of the RulesBuilder class itself, allowing for method chaining.

Example:

rule = {
- `"identifier"` - "9482a13d-ade5-46ec-b6c9-1cfc23f1f2c6",
- `"name"` - "avanzado22",
- `"active"` - False,
- `"mode"` - "ADVANCED",
- `"type"` - {
- `"datastreams"` - [
{
- `"name"` - "device.cpu.usage",
- `"fields"` - [
{
- `"field"` - "value",
- `"alias"` - "CPU usage"
}
],
- `"prefilter"` - False
}
],
- `"name"` - "DATASTREAM"
},
- `"actionsDelay"` - 1000
}

with_type = rule["type"]
with_type(with_type).
with_actions_delay(rule["actionsDelay"]).
with_code_file('reglas_update.js').
update()

delete

def delete() -> "RulesBuilder"

Delete an existing model within the organization.

This method sets up the RulesBuilder instance to delete a specific model associated with the specified organization and identifier. It configures the URL endpoint for the delete operation and sets the operation type to ‘delete’.

Returns:

RulesBuilder - The instance of the RulesBuilder class itself, allowing for method chaining.

Example:

builder.with_organization_name('organization_name').with_channel(
'default_channel'
).with_identifier("45ddc1a9-4de2-4d5f-a9fe-7972aa4555a4").delete().build().execute()

update_parameters

def update_parameters() -> "RulesBuilder"

Updates the parameters of an existing rule.

This function prepares the RulesBuilder instance to update the parameters of a specific rule associated with the specified organization and channel.

Returns:

RulesBuilder - The RulesBuilder instance itself, allowing method chaining.

Example:

builder.with_organization_name("organization_name").with_channel("default_channel").
with_name("rule_name").with_active(True).with_mode("ADVANCED").
- `with_parameters({"parameter1"` - "value1", "parameter2": "value2"}).update_parameters()

catalog

def catalog() -> "RulesBuilder"

Retrieves the rules catalog.

This function prepares the RulesBuilder instance to retrieve the catalog of rules

Returns:

RulesBuilder - The RulesBuilder instance itself, allowing method chaining.

Example:

builder.catalog()

save

def save() -> "RulesBuilder"

Selects the create-or-update operation.

It reads the rule identifier from the .env variable set with with_env(), or from the file set with with_config_file(), and then decides on its own: if a rule with that identifier already exists it is updated, and if it does not, it is created and the resulting identifier written back to the same place. Use it when a script has to be safe to run more than once.

Returns:

RulesBuilder - Returns itself to allow for method chaining.

Example:

builder.with_organization_name("organization").with_config_file(
"rules_config.ini", "id", "rule_id"
).save().build().execute()

set_config_file_identifier

def set_config_file_identifier() -> "RulesBuilder"

Selects the operation that writes the rule identifier into a configuration file.

It stores the identifier given to with_identifier() under the section and key named with with_config_file(), so a later run can pick the rule up from the file instead of carrying its identifier in the code. The key must already exist in the file.

Returns:

RulesBuilder - Returns itself to allow for method chaining.

Example:

builder.with_identifier("rule_identifier").with_config_file(
'rules_config.ini', 'id', 'rule_id'
).set_config_file_identifier().build().execute()

set_env_identifier

def set_env_identifier() -> "RulesBuilder"

Selects the operation that writes the rule identifier into the .env file.

It stores the identifier given to with_identifier() under the variable named with with_env(), so a later run can pick the rule up from the environment. The variable must already exist in the .env file.

Returns:

RulesBuilder - Returns itself to allow for method chaining.

Raises:

ValueError - If there is no identifier to write, or if the variable is not in the .env file.

Example:

builder.with_identifier("rule_identifier").with_env(
"RULE_ID"
).set_env_identifier().build().execute()

build

def build() -> "RulesBuilder"

Finalizes the construction of the IoT collection configuration.

This method prepares the builder to execute the collection by ensuring all necessary configurations are set and validates the overall integrity of the build. It should be called before executing the collection to ensure that the configuration is complete and valid.

The build process involves checking that mandatory fields such as the device identifier are set. It also ensures that method calls that are incompatible with each other (like build and build_execute) are not both used.

Returns:

RulesBuilder - Returns itself to allow for method chaining, enabling further actions like execute.

Raises:

ValueError - If required configurations are missing or if incompatible methods are used together.

Example:

builder.build()

build_execute

def build_execute()

Executes the data sets search immediately after building the configuration.

This method is a shortcut that combines building and executing in a single step.

Returns:

dict - A dictionary containing the execution response which includes the status code and potentially other metadata about the execution.

Raises:

ValueError - If build has already been called on this builder instance.

Example:

builder.build_execute()

execute

def execute()

Execute the configured operation and return the response.

This method executes the operation that has been configured using the builder pattern. It ensures that the build method has been called and that it is the last method invoked before execute. Depending on the configured method (e.g., create, find, update, delete), it calls the appropriate internal execution method.

Returns:

requests.Response - The response object from the executed request.

Raises:

Exception - If the build method has not been called or if it is not the last method invoked before execute.
ValueError - If the configured method is unsupported.

Example:

builder.execute()