Helpers

View Licence Agreement

sosw - Serverless Orchestrator of Serverless Workers

The MIT License (MIT)
Copyright (C) 2025  sosw core contributors <info@sosw.app>

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

Static helper methods which you can use in any Lambdas. Must be completely independent with no specific requirements.

sosw.components.helpers.camel_case_to_slug(name)

Convert input from camel case to slug case

Parameters

name –

• str - CamleCase string

Returns

• str - slug-case string

sosw.components.helpers.camel_case_to_underscore(name)

Converts attribute to string and formats it as underscored.

Parameters

name –

• str - CamelCase string (or something convertable to CamelCase with __str__() method.

Returns

• str - underscore_formatted_value

sosw.components.helpers.chunks(l, n)

Yield successive n-sized chunks from l.

sosw.components.helpers.construct_dates_from_event(event: dict) → tuple

Processes given event dictionary for start and end points of time. Otherwise takes the default settings.

The end date of the period may be specified as en_date in the event. The default value is today.

Also the event should have either st_date or days_back numeric parameter. If provided the days_back it will be substracted from end date.

Both st_date and en_date might be either date, datetime or string (‘YYYY-MM-DD’) types. In case of datetime, the hours/minutes/etc are ignored.

Parameters

event (dict) – Lambda payload.

Returns

start_date, end_date as datetime.date

sosw.components.helpers.convert_string_to_words(string)

Convert string to comma separated words.

Parameters

string (str) – String to convert into words.

Return type

str

Returns

Comma separated words.

sosw.components.helpers.dict_to_dunder(d, *, parent: str = '', separator: str = '__') → dict

Converts the nested dict to a flat dict with keys using dunder notation. Reverse of dunder_to_dict.

d = {
       'a': 'v1',
       'b': {
           'c': 'v2',
           'd': {'e': 'v3'}
       }
    }
result = dict_to_dunder(d)

# result:
{'a': 'v1', 'b__c': 'v2', 'b__d__e': 'v3'}

Parameters

separator (str) – Custom separator for recursive extraction. Default: ‘__’

sosw.components.helpers.dunder_to_dict(data: dict, separator=None)

Converts the flat dict with keys using dunder notation for nesting elements to regular nested dictionary.

E.g.:

data = {'a': 'v1', 'b__c': 'v2', 'b__d__e': 'v3'}
result = dunder_to_dict(data)

# result:
{
    'a': 'v1',
    'b': {
        'c': 'v2',
        'd': {'e': 'v3'}
    }
}

Parameters

• data – A dictionary that is converted to Nested.

• separator (str) – Custom separator for recursive extraction. Default: ‘__’

sosw.components.helpers.first_or_none(items: Iterable, condition: Callable = None)

Return first element in iterable to match condition or None

sosw.components.helpers.get_list_of_multiple_or_one_or_empty_from_dict(input, name, vtype=None)

Extracts objects by ‘name’ from the ‘input’ and returns as a list. Tries both plural and singular names from the input. If vtype is specified, tries to convert each of the elements in the result to this type.

Parameters

• input –

  • dict - Input dictionary. Event of Lambda for example.

• name –

  • str - Name of attribute (in plural form).

• vtype –

  • type - Type to be converted to. Must be callable. Tested types: str, int, float

Returns

• list - List of vtypes, or list of whatever was in input, or empty list.

Raises

ValueError – In all cases something is wrong.

sosw.components.helpers.get_message_dict_from_sns_event(event: Dict) → Dict

Extract SNS event message and return it loaded as a dict.

Parameters

event – Lambda SNS event (payload). Must be a JSON document.

Returns

The SNS message, converted to dict

sosw.components.helpers.get_one_from_dict(input, name, vtype=None)

Extracts object by ‘name’ from the ‘input’. Tries also plural name in case not found by single ‘name’. In case found an iterable by plural name, validates that it has exactly one value in it. If vtype is specified, tries to convert result to it.

Parameters

• input –

  • dict - Input dictionary. Event of Lambda for example.

• name –

  • str - Name of attribute (in singular form).

• vtype –

  • type - Type to be converted to. Must be callable. Tested types: str, int, float

Returns

• instance of vtype | something else | None

Raises

ValueError –

• In all cases something is wrong.

sosw.components.helpers.get_one_or_none_from_dict(input, name, vtype=None)

Extracts object by ‘name’ from the ‘input’. Tries also plural name in case not found by single ‘name’. In case found an iterable by plural name, validates that it has one or zero values in it. If vtype is specified, tries to convert result to it.

Parameters

• input (dict) – Input dictionary. Event of Lambda for example.

• name (str) – Name of attribute (in singular form).

• vtype (type) – Type to be converted to. Must be callable. Tested types: str, int, float

Returns

• instance of vtype | something else | None

Raises

ValueError – In all cases something is wrong.

sosw.components.helpers.is_event_from_sns(event)

Check if the lambda invocation was by SNS.

Parameters

event (dict) – Lambda Event (payload)

Return type

bool

sosw.components.helpers.is_valid_date(date_str, date_formats)

Validate string to be at least one of the given datetime formats.

Parameters

• date_str (str) – a date or time or both, Example: ‘2018/09/16’

• date_formats (list) – List of datetime format, that is acceptable for datetime.strptime. Example: ‘%Y/%m/%d’

Return type

bool

Returns

True if the date string is valid for any of the datetime formats, False otherwise.

sosw.components.helpers.make_hash(o)

Makes a hash from a dictionary, list, tuple or set to any level, that contains only other hashable types (including any lists, tuples, sets, and dictionaries).

Original idea from this user: https://stackoverflow.com/users/660554/jomido

Plus some upgrades to work with sets and dicts having different types of keys appropriately. See source unittests of this function for some more details.

sosw.components.helpers.nested_dict_from_keys(keys: List, value: Optional = None) → Dict

Constructs a nested dictionary using a list of keys to embed recursively. If value is provided it is assigned to the last subkey.

Examples:

nested_dict_from_keys(['a', 'b', 'c']) == {'a': {'b': {'c': None}}}
nested_dict_from_keys(['a', 'b', 'c'], value=42) == {'a': {'b': {'c': 42}}}

Parameters

• keys – List of keys to embed.

• value – Optional value to set to lowest level

sosw.components.helpers.recursive_insert(d: dict, path: str, value, separator: str = '.') → dict

Insert the value to the input dictionary d nested according to the path.

By default, the separator is ., but can be changed in the separator argument. Returns the input d with modifications applied.

Example:

d = {'a': {'b': {'z': 42}}}
result = recursive_insert(d, 'a.b.ccc.ddd', 123)

# Result
{'a': {'b': {'z': 42, 'ccc': {'ddd': 123}}}}

sosw.components.helpers.recursive_matches_extract(src, key, separator=None, **kwargs)

Searches the ‘src’ recursively for nested elements provided in ‘key’ with dot notation. In case some levels are iterable (list, tuple) it checks every element in it till finds it.

Returns the first found element or None. In case the full path is inaccessible also returns None.

If you are just checking if some elements exist, you might be interested in recursive_exists_strict() or recursive_exists_soft() helpers.

Warning

Please be aware that this method does not check for duplicates in iterable elements on neither level during extraction.

Parameters

• src (dict) – Input dictionary. Can contain nested dictionaries and lists.

• key (str) – Path to search with dot notation.

• separator (str) – Custom separator for recursive extraction. Default: ‘.’

In order to filter out some specific elements, you might want to use the optional ‘exclude’ attributes. If attributes are specified and the last level element following the path (dot notation) will have a key-value, the check for the main key-value will be skipped. See unittests to understand the bahaviour better.

Parameters

• exclude_key (str) – Key to check in last level element to exclude.

• exclude_val (str) – Value to match in last level element to exclude.

Returns

Value from structure extracted by specified path

sosw.components.helpers.recursive_matches_soft(src, key, val, **kwargs)

Searches the ‘src’ recursively for nested elements provided in ‘key’ with dot notation. In case some levels are iterable (list, tuple) it checks every element. In case the full path is inaccessible returns False. If any of the elements addressed by ‘key’ matches the ‘val’ - Bingo! Return True.

You might also be interested in recursive_exists_strict() helper.

Parameters

• src (dict) – Input dictionary. Can contain nested dictionaries and lists.

• key (str) – Path to search with dot notation.

• val (any) – Value to match in some elements specified by path.

In order to check not just that some element exists, but to check for duplicates, you might want to use optional ‘exclude’ attributes. If attributes are specified and the last level element following the path (dot notation) will have a key-value, the check for the main key-value will be skipped. See unittests to understand the bahaviour better.

Parameters

• exclude_key (str) – Key to check in last level element to exclude.

• exclude_val (srt) – Value to match in last level element to exclude.

Return type

bool

sosw.components.helpers.recursive_matches_strict(src, key, val, **kwargs)

Searches the ‘input’ recursively for nested elements provided in ‘key’ with dot notation. In case some levels are iterable (list, tuple) it checks every element. In case the full path is inaccessible raises AttributeError or KeyError.

Parameters

• src (dict) – Input dictionary. Can contain nested dictionaries and lists.

• key (str) – Path to search with dot notation.

• val (any) – Value to match in some elements specified by path.

Return type

bool

sosw.components.helpers.recursive_update(d: Dict, u: Mapping) → Dict

Recursively updates the dictionary d with another one u. Values of u overwrite in case of type conflict.

Examples (in comparison with dict.update([other])):

d = {'a': 42, 'b': {'b1': 33, 'b2': 44}}
u = {'a': 43, 'b': {'b1': 22, 'b3': 33}}

recursive_update(d, u)

# result:
{'a': 43, 'b': {'b1': 22, 'b2': 44, 'b3': 33}}

d.update(u)

# result:
{'a': 43, 'b': {'b1': 22, 'b3': 33}}

List, set and tuple values of d and u are merged, preserving only unique values. Returned as List.

sosw.components.helpers.rstrip_all(input, patterns)

Strips all of the patterns from the right of the input. Order and spaces do not matter.

Parameters

• input –

  • str - String to modify

• patterns –

  • list|set|tuple|str - Pattern[-s] to remove.

Returns

• str

sosw.components.helpers.slug_to_camel_case(text: str) → str

Split the input text by hyphens to get individual words Convert each word to capitalize the first letter and join them to form CamelCase

Returns

Resulting CamelCase string

sosw.components.helpers.small_int_from_string(input_string: str, num_digits: int = 2) → int

Generate a small integer based on the input string using its MD5 hash. This value is reproducible, so it could be useful for example if you use it for some kind of partitioning or unsorted batching in order to be able to query based on it later on.

Examples:

small_int_from_string("hello world")
91
small_int_from_string("hello world", num_digits=3)
291

Returns

The generated small integer.

Raises

ValueError: If num_digits is not a positive integer.

sosw.components.helpers.trim_arn_to_account(arn: str) → str

Extract just the ACCOUNT_ID from full ARN. Supports versions, aliases or raw name (without ARN).

More information about ARN Format: https://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html#genref-arns

sosw.components.helpers.trim_arn_to_name(arn: str) → str

Extract just the name of function from full ARN. Supports versions, aliases or raw name (without ARN).

More information about ARN Format: https://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html#genref-arns

sosw.components.helpers.underscore_to_camel_case(name)

Convert input from underscore case to camel case

Parameters

name –

• str - snake_case string

Returns

• str - SnakeCase string

sosw.components.helpers.unwrap_event_recursively(event: Dict, sources: Optional[List[str]] = None) → List[Dict]

Recursively unwraps lambda event from SQS and/or SNS event skeletons. Supported sources: ‘sqs’, ‘sns’. Will unwrap recursively until the event is not wrapped anymore, or up to depth of 10

unwrapped_messages = unwrap_event_recursively(event, sources=['sns', 'sqs']):

Parameters

• event – Lambda event

• sources – List of strings describing what the event might be wrapped by. If empty, will unwrapped from all.

Returns

List of dictionaries - unwrapped messages from the event

sosw.components.helpers.validate_account_to_dashed(account)

Validates the the provided string is in valid AdWords account format and converts it to dashed format.

Parameters

account (str) – AdWords Account

Return type

str

Returns

Dashed format

sosw.components.helpers.validate_account_to_int(account)

Validates the provided string is in valid AdWords account format and converts it to integer format.

Parameters

account ((str, int)) – AdWords Account

Returns

Account ID as integer

sosw.components.helpers.validate_date_from_something(d)

Convert valid input to datetime.date() or raise either AttributeError or ValueError.

Parameters

d – Some input. Supported types: * datetime.datetime * datetime.date * int - Epoch or Epoch milliseconds * float - Epoch or Epoch milliseconds * str (YYYY-MM-DD) * str (YYYY-MM-DD HH:MM:SS)

Returns

Transformed d

Return type

datetime.date

Raises

ValueError

sosw.components.helpers.validate_date_list_from_event_or_days_back(input, days_back=0, key_name='date_list')

Takes from input the date_list and extracts date_list. Validates and converts to datetime.date. Input should have date_list as list of strings or comma-separated string.

• Format: YYYY-MM-DD

• Examples:

['2018-01-01', '2018-02-01']
'2018-01-01, 2018-02-01'

Parameters

• input (dict) – This is supposed to be your whole Lambda event.

• days_back (int) – Optional Number of days to take back from today. Ex: days_back=1 is yesterday. Default: today.

• key_name (str) – Optional custom name of key to extract from ‘input’.

Returns

list(datetime.date)

sosw.components.helpers.validate_date_timestamp_from_something(d) → float

Converts the input d to timestamp of the date rounded.

Parameters

d – Some input. Supported types: * datetime.datetime * datetime.date * int - Epoch or Epoch milliseconds * float - Epoch or Epoch milliseconds * str (YYYY-MM-DD) * str (YYYY-MM-DD HH:MM:SS) * str(epoch time seconds as string) * str(epoch time seconds (float) as string)

Returns

Transformed d

Raises

ValueError

sosw.components.helpers.validate_datetime_from_something(d)

Converts the input d to datetime.datetime.

Parameters

d – Some input. Supported types: * datetime.datetime * datetime.date * int - Epoch or Epoch milliseconds * float - Epoch or Epoch milliseconds * str (epoch time seconds as string) * str (epoch time seconds (float) as string) * str: Many different formats are supported, but the order only from more specific to less specific, (e.g., YYYY-MM-DD, in favor of DD-MM-YYYY).

Returns

Transformed d

Return type

datetime.datetime

Raises

ValueError

sosw.components.helpers.validate_list_of_numbers_from_csv(data)

Converts a comma separated string of numeric values to a list of sorted unique integers. The values that do not match are skipped.

Parameters

data ((str, iterable)) –

• str | iterable

Returns

• list(int)

sosw.components.helpers.validate_list_of_words_from_csv_or_list(data: (<class 'str'>, <class 'list'>)) → list

Splits a CSV string to list of stripped words. In case the data is already a list of strings - splits it’s elements and flattens the result.

All resulting elements must be single words, if any of the elements contains spaces (i.e. multiple words) the validation fails with ValueError.

Parameters

data – CSV string of list of strings (possibly CSV themselves)

Returns

List of stripped and split words

sosw.components.helpers.validate_string_matches_datetime_format(date_str, date_format, field_name='date')

Validate string, make sure it’s of the given datetime format

Parameters

• date_str (str) – a date or time or both, Example: ‘2018/09/16’

• date_format (str) – datetime format, that is acceptable for datetime.strptime. Example: ‘%Y/%m/%d’ (https://docs.python.org/3.6/library/datetime.html#strftime-and-strptime-behavior)

• field_name (str) – name of the field (for the error)

Raises

ValueError

sosw.components.helpers.validate_uuid4(uuid_string)

Validate that a UUID string is in fact a valid uuid4. Happily, the uuid module does the actual checking for us. It is vital that the ‘version’ kwarg be passed to the UUID() call, otherwise any 32-character hex string is considered valid.