fastiot.util package

Submodules

fastiot.util.case_conversions module

fastiot.util.case_conversions.kebab_case_to_snake_case(d)

Converts keys of dictionary recursively from kebab case, e.g. my-key to snake case, e.g. my_key.

fastiot.util.classproperty module

fastiot.util.classproperty.classproperty(fn)

fastiot.util.config_helper module

Providing methods to import yaml configuration per service

pydantic model fastiot.util.config_helper.FastIoTConfigModel

You may use this as a base class to provide a proper data model for your configurations. This is a proper alternative to relying on reading in YAML-files and working with dictionaries as done with fastiot.util.config_helper.read_config().

Please consult Configuration for a service for more information about handling configurations and a full example.

classmethod from_yaml_file(filename)

Does the magic of import yaml to pydantic model, provided by a filename.

classmethod from_service(service)

Read in the configuration for your service, so just use it with MyServiceConfig.from_service(self) inside your service.

It is possible to read the configuration for a service instantiated multiple times. If both service.yaml and service_id.yaml exist, service_id.yaml will be preferred. Use the environment variable FASTIOT_SERVICE_ID to set the individual service id for a service.

Also see Configuration for a service for more information about handling configurations.

fastiot.util.config_helper.read_config(service)

Load YAML-configuration files based on file (provide string) or, preferably, service.

It is possible to read the configuration for a service instantiated multiple times. If both service.yaml and service_id.yaml exist, service_id.yaml will be preferred. Use the environment variable FASTIOT_SERVICE_ID to set the individual service id for a service.

Also see Configuration for a service for more information about handling configurations.

Example passing your service to get a filename configuration automatically:

>>> from fastiot.core import FastIoTService
>>> from fastiot.util.config_helper import read_config
>>>
>>> class MyService(FastIoTService)
>>>
>>>     def __init__(self, **kwargs)
>>>         super().__init__(**kwargs)
>>>         my_config = read_config(self)

Parameters:

service (Union[FastIoTService, str]) – service to load the config for (preferred) or name of the config file.

Return type:

Dict

Returns:

Dictionary with the loaded configuration. May be empty if no configuration was found.

fastiot.util.csv_reader module

class fastiot.util.csv_reader.CSVReader(filename, required_fields=None, optional_fields=None, checks=None, do_allow_arbitrary_fields=False)

A class for csv file reading. It uses python-lib’s csv package and does more strict checks on how the csv file must be formatted. Raises a CSVError if the file is incorrectly formatted.

Example usage:

with CSVReader('my_file.csv',
               required_fields=['my_field1', 'my_field2'],
               optional_fields=['my_optional_field2']) as reader:
    for data_row in reader:
        print(data_row['my_field1'])
        print(data_row['my_field2'])
        print(data_row.get('my_optional_field2', 'unset'))

__init__(filename, required_fields=None, optional_fields=None, checks=None, do_allow_arbitrary_fields=False)

Constructor for csv reader.

Parameters:

• filename (str) – The filename of the csv file. Must be relative to workdir.

• required_fields (Optional[List[str]]) – Specify required field names which must be in the csv header line. If not, it will raise a CSVError during parsing.

• optional_fields (Optional[List[str]]) – Specify optional field names. If the csv file contains field names, which are not specified, it will raise a CSVError during parsing.

• checks (Optional[Dict[str, Callable[[str], bool]]]) – Specify checks for fields as a mapping of field names to callables. The callables should return true if a given value is valid, false otherwise.

• do_allow_arbitrary_fields (bool) – If true, it will consider all possible field names as optional. Use this option with caution because it will prevent possible errors from being detected.

fastiot.util.object_helper module

fastiot.util.object_helper.parse_object(dic, data_model)

This function will help you convert a dictionary to an instance, of which the class inherits FastIoTData or BaseModel.

my_dict = {'name': 'test_dict', 'value': 123}

obj = parse_object(my_dict, MyDataModel)
>>> MyDataModel(name='test_dict', value=123)

Return type:

Optional[Type[Union[FastIoTData, BaseModel]]]

fastiot.util.object_helper.parse_object_list(dict_list, data_model)

This function helps you to convert a list of dictionaries to a list of instance, of which the class inherits FastIoTData or BaseModel

my_dict_list = [{'name': 'test_dict_1', 'value': 123}, {'name': 'test_dict_2', 'value': 23}]

obj_list = parse_object_list(my_dict_list, MyDataModel)
>>> [MyDataModel(name='test_dict_1', value=123), MyDataModel(name='test_dict_2', value=23)]

Return type:

Optional[List[Type[Union[FastIoTData, BaseModel]]]]

fastiot.util.ports module

fastiot.util.ports.get_local_random_port()

Will use sockets to find an open port offered by the operating system

Return type:

int

fastiot.util.read_yaml module

Deprecated since version 0.9.29: This module is deprecated. Please refer to fastiot.util.config_helper instead.

fastiot.util.read_yaml.read_config(service)

Load YAML-configuration files based on file (provide string) or, preferably, service. :rtype: Dict

Deprecated since version 0.9.29: Please use fastiot.util.config_helper.read_config() instead now.