Utils

opengate_data.utils

opengate_data.utils.expressions

Expressions Objects

class Expressions()

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 existed condition to the filter.

Arguments:

  • field str - The field to check for existence.

Returns:

  • dict - The existed condition.
  • value any - The value to compare against.

Example:

exists(“device.location”, False)

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,
    files: [str, [str, bytes, str]] = None,
    json_payload: dict[str, Any] = None,
    params: Any = None,
    stream=False
) -> Response | dict[str, str] | dict[str, str] | dict[str, str] | dict[str,
                                                                        str]

Helper function to make HTTP requests.

This function simplifies making HTTP requests by handling different HTTP methods (POST, GET, PUT, DELETE) and managing common exceptions. It supports sending payloads and files as part of the request.

Arguments:

  • method str - The HTTP method to use for the request (‘post’, ‘get’, ‘put’, ‘delete’).
  • headers dict, optional - The headers to include in the request. Defaults to None.
  • url str, optional - The URL to which the request is sent. Defaults to ‘’.
  • data dict[str, Any] | str, optional - The payload to include in the request body. Defaults to None.
  • files [str, [str, bytes, str]], optional - The files to include in the request. Defaults to None.
  • params dict, optional - The params to include in the request. Defaults to None.
  • json_payload dict[str, Any], optional - The JSON payload to include in the request body. Defaults to None.

Returns:

Response | str: The response object from the request if successful, or an error message string if an exception occurs.

Raises:

  • ValueError - If an unsupported HTTP method is provided.

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.

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.

  • field_aliases: maps internal fields to public setter names (e.g., ‘path’ → ‘with_path’)
  • method_aliases: maps logical method names to public method names (e.g., ’list_one’ → ’list_one()’)

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.