opengate_data.utils.expressions

Expressions Objects

class Expressions()

Expressions

The filter conditions a search accepts, as static methods returning plain dicts: eq, neq, like, gt, lt, gte, lte, in_, nin and exists. FilterBuilder extends this class, so in practice you reach them through client.new_filter_builder() and combine them with and_ / or_ before calling build().

Example:

filter = client.new_filter_builder().and_(
Expressions.eq("provision.administration.organization", "organization"),
Expressions.exists("provision.device.identifier", True),
).build()

eq

def eq(field, value)

Adds an equality condition to the filter.

Arguments:

  • field str - The field to compare.
  • value any - The value to compare against.

Returns:

  • dict - The equality condition.

Example:

eq(“device.operationalStatus”, “NORMAL”)

neq

def neq(field, value)

Adds a not-equal condition to the filter.

Arguments:

  • field str - The field to compare.
  • value any - The value to compare against.

Returns:

  • dict - The not-equal condition.

Example:

neq(“device.operationalStatus”, “ERROR”)

like

def like(field, value)

Adds a like condition to the filter.

Arguments:

  • field str - The field to compare.
  • value str - The regex pattern to match.

Returns:

  • dict - The like condition.

Example:

like(“device.name”, “device_.*”)

gt

def gt(field, value)

Adds a greater-than condition to the filter.

Arguments:

  • field str - The field to compare.
  • value any - The value to compare against.

Returns:

  • dict - The greater-than condition.

Example:

gt(“device.batteryLevel”, 50)

lt

def lt(field, value)

Adds a less-than condition to the filter.

Arguments:

  • field str - The field to compare.
  • value any - The value to compare against.

Returns:

  • dict - The less-than condition.

Example:

lt(“device.batteryLevel”, 20)

gte

def gte(field, value)

Adds a greater-than-or-equal condition to the filter.

Arguments:

  • field str - The field to compare.
  • value any - The value to compare against.

Returns:

  • dict - The greater-than-or-equal condition.

Example:

gte(“device.batteryLevel”, 80)

lte

def lte(field, value)

Adds a less-than-or-equal condition to the filter.

Arguments:

  • field str - The field to compare.
  • value any - The value to compare against.

Returns:

  • dict - The less-than-or-equal condition.

Example:

lte(“device.batteryLevel”, 30)

in_

def in_(field, values)

Adds an in condition to the filter.

Arguments:

  • field str - The field to compare.
  • values list - The list of values to compare against.

Returns:

  • dict - The in condition.

Example:

in_(“device.name”, [“device_1”, “device_2”])

nin

def nin(field, values)

Adds a not-in condition to the filter.

Arguments:

  • field str - The field to compare.
  • values list - The list of values to compare against.

Returns:

  • dict - The not-in condition.

Example:

nin(“device.name”, [“device_3”, “device_4”])

exists

def exists(field, value)

Adds an existence condition to the filter.

Arguments:

  • field str - The field to check for existence.
  • value bool - True to match the entities that have the field, False to match the ones that do not.

Returns:

  • dict - The existence condition.

Example:

exists("device.location", True)

opengate_data.utils.utils

validate_type

def validate_type(variable: Any, expected_type: Any,
                  variable_name: str) -> None

Validates that the given variable is of the expected type or types.

This function checks if the variable matches the expected type or any type in a tuple of expected types. It raises a TypeError if the variable does not match the expected type(s).

Arguments:

  • variable Any - The variable to be checked.
  • expected_type Any - The expected type or a tuple of expected types.
  • variable_name str - The name of the variable, used in the error message to identify the variable.

Raises:

  • TypeError - If the variable is not of the expected type(s).

Returns:

  • None - This function does not return a value; it raises an exception if the type check fails.

set_method_call

def set_method_call(method)

Decorates a method to ensure it is properly registered and tracked within the builder’s workflow.

This decorator adds the method’s name to a set that tracks method calls

Arguments:

  • method function - The method to be decorated.

Returns:

  • function - The wrapped method with added functionality to register its call.

Raises:

  • None - This decorator does not raise exceptions by itself but ensures the method call is registered.

parse_json

def parse_json(value)

Attempts to convert a string into a Python object by interpreting it as JSON.

Arguments:

  • value str | Any - The value to attempt to convert. If the value is not a string, it is returned directly without attempting conversion.

Returns:

  • Any - The Python object resulting from the JSON conversion if value is a valid JSON string. If the conversion fails due to a formatting error (ValueError), the original value is returned. If value is not a string, it is returned as is.

send_request

def send_request(method: str,
                 headers: dict | None = None,
                 url: str = "",
                 data: dict[str, Any] | str | None = None,
                 files: Any = None,
                 json_payload: dict[str, Any] | None = None,
                 params: Any = None,
                 verify: bool = False,
                 timeout: int = 3000,
                 stream: bool = False) -> Response | dict[str, Any]

Helper function to make HTTP requests using the requests library.

Arguments:

  • method str - HTTP method (e.g., ‘GET’, ‘POST’, ‘PUT’, ‘DELETE’, ‘PATCH’).
  • headers dict, optional - Request headers.
  • url str - The URL for the request.
  • data dict | str, optional - Payload for the request body.
  • files Any, optional - Files for multipart encoding upload.
  • json_payload dict, optional - JSON payload for the request body.
  • params dict | Any, optional - URL parameters.
  • verify bool - Whether to verify SSL certificates. Defaults to False.
  • timeout int - Request timeout in seconds. Defaults to 3000.
  • stream bool - Whether to keep the response body unread until it is consumed, which is what a file download needs. Defaults to False.

Returns:

Response | dict[str, Any]: The response object if successful, or an error dictionary.

handle_basic_response

def handle_basic_response(response: requests.Response) -> dict[str, Any]

Handle basic HTTP response.

This function processes the HTTP response and returns a dictionary containing the status code. If the response indicates an error, it also includes the error message.

Arguments:

  • response requests.Response - The response object to process.

Returns:

dict[str, Any]: A dictionary containing the status code and, if applicable, the error message.

handle_error_response

def handle_error_response(response: requests.Response) -> dict[str, Any]

Turn a failed HTTP response into the dictionary the builders return.

Nothing is raised: the caller gets the status code and, when the platform sent one, the body as the error.

Arguments:

  • response requests.Response - The response object to process.

Returns:

dict[str, Any]: The status code and, if the response carried a body, the error.

handle_exception

def handle_exception(e)

Turn a transport failure into the dictionary the builders return.

A request that never reached the platform has no status code, so the result names the kind of failure instead: connection, timeout, request or unexpected.

Arguments:

  • e Exception - The exception raised while sending the request.

Returns:

dict[str, str]: The kind of failure and its details.

validate_build_method_calls_execute

def validate_build_method_calls_execute(method_calls)

Check that a chain was closed properly before execute() runs.

Every builder has to be closed with build() or with build_execute(), and build() has to be the last call before execute() — otherwise a setter called after it would never be validated.

Arguments:

  • method_calls list[str] - The methods called on the builder, in order.

Raises:

  • Exception - If build() was called more than once, if it was not the last call before execute(), or if neither build() nor build_execute() was called.

validate_build

def validate_build(*,
                   method: str,
                   state: dict,
                   spec: dict,
                   used_methods: list[str] | None = None,
                   allowed_method_calls: set[str] | None = None,
                   field_aliases: dict[str, str] | None = None,
                   method_aliases: dict[str, str] | None = None) -> None

Generic builder validator with messages intended for the end user.

The two alias maps exist so the errors name what the caller wrote, not the internals: without them a missing with_destiny_path() would be reported as a missing ‘path’.

Arguments:

  • method str - The operation selected in the chain, which picks its entry in the spec.
  • state dict - The current value of each field of the builder, by internal name.
  • spec dict - The rules per operation: ‘required’ and ‘forbidden’ field names, ‘choices’ mapping a field to its allowed values, and ‘custom’ holding extra checks called with the state.
  • used_methods list[str] | None - The methods called on the builder, in order. Needed to catch two operations in the same chain.
  • allowed_method_calls set[str] | None - The method names that select an operation, of which a chain may use exactly one, once.
  • field_aliases dict[str, str] | None - Maps internal fields to public setter names (e.g., ‘path’ → ‘with_path’).
  • method_aliases dict[str, str] | None - Maps logical method names to public method names (e.g., ’list_one’ → ’list_one()’).

Raises:

  • RuntimeError - If the chain combines several operations, or repeats one.
  • ValueError - If no operation was selected, if it is not in the spec, or if the state misses a required field, carries a forbidden one, or holds a value outside its choices.

build_headers

def build_headers(client_headers: dict | None,
                  *,
                  accept: str | None = None,
                  content_type: str | None = None) -> dict

Build request headers starting from client headers (auth preserved) and optionally overriding Accept / Content-Type.

This function NEVER mutates client headers.

Arguments:

  • client_headers dict | None - The headers of the client, holding the authentication. They are copied, never modified.
  • accept str | None - The Accept header to set. Left untouched if None.
  • content_type str | None - The Content-Type header to set. Left untouched if None.

Returns:

  • dict - A new set of headers.

opengate_data.utils