filip.utils package

Utility module for helper functions

Submodules

filip.utils.cleanup module

Functions to clean up a tenant within a fiware based platform.

Decorator to clean up the server before and after the test

Note

This does not substitute a proper TearDown method, because a failing test will not execute the clean up after the error. Since this would mean an unnecessary error handling. We actually want a test to fail with proper messages.

Parameters:

fiware_service – tenant
fiware_servicepath – tenant path
cb_url – url of context broker service
iota_url – url of IoT-Agent service
ql_url – url of quantumleap service
cb_client – enables TLS communication if created with Session object, only needed for self-signed certificates
iota_client – enables TLS communication if created with Session object, only needed for self-signed certificates
ql_client – enables TLS communication if created with Session object, only needed for self-signed certificates

Returns:

Decorator for clean tests

Clears all services that a url is provided for

Parameters:

fiware_header –
cb_url – url of the context broker service
iota_url – url of the IoT-Agent service
ql_url – url of the QuantumLeap service
cb_client – enables TLS communication if created with Session object, only needed for self-signed certificates
iota_client – enables TLS communication if created with Session object, only needed for self-signed certificates
ql_client – enables TLS communication if created with Session object, only needed for self-signed certificates

Returns:

None

filip.utils.cleanup.clear_context_broker(url: str | None = None, fiware_header: FiwareHeader | None = None, clear_registrations: bool = False, cb_client: ContextBrokerClient | None = None)[source]

Function deletes all entities, registrations and subscriptions for a given fiware header. To use TLS connection you need to provide the cb_client parameter as an argument with the Session object including the certificate and private key.

Note

Always clear the devices first because the IoT-Agent will otherwise through errors if it cannot find its registration anymore.

Parameters:

url – Url of the context broker service
fiware_header – header of the tenant
cb_client – enables TLS communication if created with Session object, only needed for self-signed certificates
clear_registrations – Determines whether registrations should be deleted. If registrations are deleted while devices with commands still exist, these devices become unreachable. Only set to true once such devices are cleared.

Returns:

None

filip.utils.cleanup.clear_iot_agent(url: Url | str | None = None, fiware_header: FiwareHeader | None = None, iota_client: IoTAClient | None = None)[source]

Function deletes all device groups and devices for a given fiware header. To use TLS connection you need to provide the iota_client parameter as an argument with the Session object including the certificate and private key.

Parameters:

url – Url of the iot agent service
fiware_header – header of the tenant
iota_client – enables TLS communication if created with Session object, only needed for self-signed certificates

Returns:

None

filip.utils.cleanup.clear_quantumleap(url: str | None = None, fiware_header: FiwareHeader | None = None, ql_client: QuantumLeapClient | None = None)[source]

Function deletes all data for a given fiware header. To use TLS connection you need to provide the ql_client parameter as an argument with the Session object including the certificate and private key. :param url: Url of the quantumleap service :param fiware_header: header of the tenant :param ql_client: enables TLS communication if created with Session object, only needed for self-signed certificates

Returns:: None

filip.utils.data module

filip.utils.data.load_datapackage(url: str, package_name: str) → Dict[str, DataFrame][source]

Downloads data package from online source and stores it as hdf-file in filip.data named by the <filename>.hdf.

Parameters:

url (str) – Valid url to where the data package is hosted
package_name (str) – name of the cached file.

Returns:

Dict of dataframes

filip.utils.datetime module

filip.utils.datetime.convert_datetime_to_iso_8601_with_z_suffix(dt: datetime) → str[source]

Converts datetime object to iso8601 notation with z-suffix

Parameters:: dt – datetime object
Returns:: String in iso 8601 notation with z-suffix

filip.utils.datetime.transform_to_utc_datetime(dt: datetime) → datetime[source]

Converts datetime object to utc datetime object with zone

Parameters:: dt –

Returns:

filip.utils.filter module

Filter functions to keep client code clean and easy to use.

Filter the given device list based on conditions

Parameters:

devices – device list that need to be filtered
device_ids – A list of device_id as filter condition
entity_names – A list of entity_name (e.g. entity_id) as filter condition.
entity_types – A list of entity_type as filter condition

Returns:

List of matching devices

filip.utils.filter.filter_group_list(group_list: List[ServiceGroup], resources: str | List[str] | None = None, apikeys: str | List[str] | None = None) → List[ServiceGroup][source]

Filter service group based on resource and apikey.

Parameters:

group_list – The list of service groups that need to be filtered
resources – see ServiceGroup model
apikeys – see ServiceGroup

Returns: a single service group or Not Found Error

filip.utils.filter.filter_subscriptions_by_entity(entity_id: str, entity_type: str, url: str | None = None, fiware_header: FiwareHeader | None = None, subscriptions: List[Subscription] | None = None) → List[Subscription][source]

Function that filters subscriptions based on the entity id or id pattern and entity type or type pattern. The function can be used in two ways, wither pass list of subscriptions to filter based on entity or directly pass client information to filter subscriptions in a single request.

Parameters:

entity_id – Id of the entity to be matched
entity_type – Type of the entity to be matched
url – Url of the context broker service
fiware_header – Fiware header of the tenant
subscriptions – List of subscriptions to filter

Returns:

list of subscriptions by entity

filip.utils.geo_ql module

Implementation of geo query language

filip.utils.iot module

Helper functions to handle the devices related with the IoT Agent

Filter the given device list based on conditions

Parameters:

devices – device list that need to be filtered
device_ids – A list of device_id as filter condition
entity_names – A list of entity_name (e.g. entity_id) as filter condition.
entity_types – A list of entity_type as filter condition

Returns:

List of matching devices

filip.utils.model_generation module

Code generator for data models from schema.json descriptions

Creates a ContextEntity-Model from a dict:

Parameters:

name – name of the model
data – dictionary containing the data structure
validators (optional) – validators for the new model
path – if present the model will written to *.py file if file ending *.py is used and to json-schema if *.json is used.

Example

>>> def username_alphanumeric(cls, value):
        assert v.value.isalnum(), 'must be numeric'
        return value

>>> model = create_context_entity_model(
                name='MyModel',
                data={
                    'id': 'MyId',
                    'type':'MyType',
                    'temp': 'MyProperty'}
                {'validate_test': validator('temperature')(
                    username_alphanumeric)})

Returns:: ContextEntity

This will create a data model from data model definitions. The schemas can either downloaded from a url or passed as str or dict. Allowed input types are defined but the underlying toolbox.

Many data models suited for FIWARE are located here: https://github.com/smart-data-models/data-models

Parameters:

path – path where the generated code should saved
url – url to download the definition from
schema_type (str) – auto, openapi, jsonschema, json, yaml, dict, csv
class_name – classname for the model class

Returns:

None

Examples:

{
    "type": "object",
    "properties": {
        "number": {
            "type": "number"
            },
        "street_name": {
            "type": "string"
           },
        "street_type": {
            "type": "string",
            "enum": ["Street", "Avenue", "Boulevard"]
        }
    }
}

filip.utils.simple_ql module

The Simple Query Language provides a simplified syntax to retrieve entities which match a set of conditions. A query is composed by a list of statements separated by the ‘;’ character. Each statement expresses a matching condition. The query returns all the entities that match all the matching conditions (AND logical operator).

For further details of the language please refer to:

https://telefonicaid.github.io/fiware-orion/api/v2/stable/

class filip.utils.simple_ql.Operator(value=<no_arg>, names=None, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: str, Enum

The list of operators (and the format of the values they use) is as follows:

EQUAL = '=='

GREATER_OR_EQUAL = '>='

GREATER_THAN = '>'

LESS_OR_EQUAL = '<='

LESS_THAN = '<'

MATCH_PATTERN = '~='

UNEQUAL = '!='

classmethod list()[source]

Returns:: list of all valid values

class filip.utils.simple_ql.QueryStatement(left: str, op: str | Operator, right: Any)[source]

Bases: Tuple

Simple query statement

classmethod parse_str(string: str)[source]

Generates QueryStatement form string

Parameters:: string –
Returns:: QueryStatement

to_str()[source]

Parses QueryStatement to String

Returns:: String

classmethod validate(value)[source]

Validates statements

Parameters:: value –

Returns:

class filip.utils.simple_ql.QueryString(qs: Tuple | QueryStatement | List[QueryStatement | Tuple])[source]

Bases: object

Class for validated QueryStrings that can be used in api clients

classmethod parse_str(string: str)[source]

Creates QueryString from string

Parameters:: string –
Returns:: QueryString

remove(qs: Tuple | QueryStatement | List[QueryStatement])[source]

Remove Statement from QueryString :param qs:

Returns:

to_str()[source]

Parsing self.qs to string object

Returns:: query string that can be added to requests as parameter
Return type:: String

update(qs: Tuple | QueryStatement | List[QueryStatement])[source]

Adds or updates QueryStatement within QueryString. First to arguments must match an existing argument for update. This redundant rules

Parameters:: qs – Query statement to add to the string object
Returns:: None

classmethod validate(v)[source]: validate QueryString

filip.utils.validators module

Helper functions to prohibit boiler plate code

class filip.utils.validators.FiwareRegex(value=<no_arg>, names=None, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: str, Enum

Collection of Regex expression used to check if the value of a Pydantic field, can be used in the related Fiware field.

standard = '(^((?![?&#/\\"\' ])[\\x00-\\x7F])*$)'

string_protect = '(?!^id$)(?!^type$)(?!^geo:location$)(^((?![?&#/\\"\' ])[\\x00-\\x7F])*$)'

filip.utils.validators.ignore_none_input(func)[source]

filip.utils.validators.match_regex(value: str, pattern: str)[source]

filip.utils.validators.validate_escape_character_free(value: Any) → Any[source]

Function that checks whether a value contains a string part that starts or end with ‘ or “. the function iterates to break down each complex data-structure to its fundamental string parts. Each value of a list is examined Of dictionaries each value is examined, keys are skipped, as they are ok for Fiware

Parameters:: value – the string to check
Returns:: validated string

filip.utils.validators.validate_expression_language(cls, expressionLanguage)[source]

filip.utils.validators.validate_fiware_datatype_standard(arg)[source]

filip.utils.validators.validate_fiware_datatype_string_protect(arg)[source]

filip.utils.validators.validate_fiware_service(arg)[source]

filip.utils.validators.validate_fiware_service_path(arg)[source]

filip.utils.validators.validate_fiware_standard_regex(vale: str)[source]

filip.utils.validators.validate_fiware_string_protect_regex(vale: str)[source]

filip.utils.validators.validate_http_url(url: Url) → str[source]

Function checks whether the host has “http” added in case of http as protocol.

Parameters:: url (AnyHttpUrl) – the url for the host / port
Returns:: validated url

filip.utils.validators.validate_jexl_expression(expression, attribute_name, device_id)[source]

filip.utils.validators.validate_mqtt_topic(arg)[source]

filip.utils.validators.validate_mqtt_url(url: Url) → str[source]

Function that checks whether a url is valid mqtt endpoint

Parameters:: url – the url for the target endpoint
Returns:: validated url