filip.utils package

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: str | Url | 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.geo_ql module

Implementation of geo query language

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

Module contents

Utility module for helper functions