jgt_common package

General helper functions.

Inheritance diagram of jgt_common

The functions and classes in this module fall into several general categories:

Classes Defined in this Module

CommonAttributeList()

A list for similar objects that can be operated on as a group.

Flag()

A settable, nameable Boolean flag.

IncompleteAtTimeoutException()

Exception for check_until results that timeout still pending validation.

NotEmptyList()

A list that fails to iterate if it is empty.

ResponseInfo()

Keep track of needed info about test operation responses (or any info really).

ResponseList()

A list specialized for testing, w/ResponseInfo object items.

Dictionary related functions

dict_from()

Build a new dict based on turning values from an iterable into key/value pairs.

dict_strip_value()

Return a new dict based on stripping out any key with the given value.

dict_transform()

Return a new dict based on transforming the keys and/or values of a_dict.

filter_dict()

Filter a dictionary based on truthiness.

must_get_key()

Either return the value for the key, or raise an exception.

must_get_keys()

Either return the value found for the keys provided, or raise an exception.

Documentation support

build_classification_rst_string()

Create rST for all the items in from_dict that are part of for_module.

classify()

Add glossary subject category classification meta-data to it’s target.

first_line_of_doc_string()

Get the first non-empty line of the doc string for thing, if there is one.

Environment related functions

must_be_in_virtual_environment()

Ensure the current process is running in a virtual environment.

var_from_env()

Try to get a value from an environment variable.

Exceptions and exception handling

IncompleteAtTimeoutException()

Exception for check_until results that timeout still pending validation.

assert_if_values()

Assert if any truthy values are returned (or yielded) from the decorated function.

retry_on_exceptions()

Retry a function based on provided parameters.

Exitting the process

cleanup_and_exit()

Cleanup a directory tree that was created and exit.

error_if()

Exit the program if a provided check is true.

exit()

Exit the program and optionally print a message to standard error.

must_be_in_virtual_environment()

Ensure the current process is running in a virtual environment.

safe_run()

Run the given list of commands, only return if no error.

File and file contents related functions

display_name()

Create a human-readable name for a given project.

filter_lines()

Filter the given lines based on the given filter function.

get_file_contents()

Get the contents of a file as a Python string.

get_file_docstring()

Get the full docstring of a given python file.

Filtering and Transforming functions

dict_from()

Build a new dict based on turning values from an iterable into key/value pairs.

dict_strip_value()

Return a new dict based on stripping out any key with the given value.

dict_transform()

Return a new dict based on transforming the keys and/or values of a_dict.

filter_dict()

Filter a dictionary based on truthiness.

filter_lines()

Filter the given lines based on the given filter function.

Looping / Retry related items

IncompleteAtTimeoutException()

Exception for check_until results that timeout still pending validation.

check_until()

Periodically call a function until its result validates or the timeout is exceeded.

retry_on_exceptions()

Retry a function based on provided parameters.

Meta-data related functions

classify()

Add glossary subject category classification meta-data to it’s target.

display_name()

Create a human-readable name for a given project.

get_file_docstring()

Get the full docstring of a given python file.

Miscellaneous functions

Flag()

A settable, nameable Boolean flag.

accumulator_for()

Accumulate the results of calling fun into a unique list, returning that list.

always_false()

Return False regardless of any provided arguments.

always_true()

Return True regardless of any provided arguments.

assert_if_values()

Assert if any truthy values are returned (or yielded) from the decorated function.

default_if_none()

Return default if value is None else value.

fib_or_max()

Get the nth Fibonacci number or max_number, which ever is smaller.

format_if()

Return a message string with a formatted value if any content value is present.

identity()

Build simple identity function based on provided parameters.

is_iterable()

Return True if item is iterable, False otherwise, using an iter(item) test.

no_op()

Reusable no-op function.

percent_diff()

Get percentage difference, out to precision places.

ticketing_system_for()

Return the Ticketing System/Type for the given ticket, or the empty string.

url_for_ticket()

Return a URL for the given ticketing_system, ticket pair.

url_if_ticket()

Return a Ticket URL if a ticket is found.

Random data related functions

generate_random_string()

Generate a random string of the specified size.

Classes/functions for working with the requests library

ResponseInfo()

Keep track of needed info about test operation responses (or any info really).

ResponseList()

A list specialized for testing, w/ResponseInfo object items.

Subprocesses/Commands related functions

safe_run()

Run the given list of commands, only return if no error.

Sequences/Lists helper classes and functions

CommonAttributeList()

A list for similar objects that can be operated on as a group.

NotEmptyList()

A list that fails to iterate if it is empty.

accumulator_for()

Accumulate the results of calling fun into a unique list, returning that list.

index_or_default()

Return the index of a value from a list, or a default if not found.

is_iterable()

Return True if item is iterable, False otherwise, using an iter(item) test.

list_from()

Generate a list from a single item or an iterable.

no_nones()

Return a list of the non-None values in iterable.

only_item_of()

Assert item_sequence has only one item, and return that item.

padded_list()

Generate a fixed-length list from an iterable, padding as needed.

string_to_list()

.split() a string into a list and .strip() each piece.

truths_from()

Return a list of the truthy values in iterable.

String related functions

format_if()

Return a message string with a formatted value if any content value is present.

generate_random_string()

Generate a random string of the specified size.

string_to_list()

.split() a string into a list and .strip() each piece.

Ticketing System related functions

ticketing_system_for()

Return the Ticketing System/Type for the given ticket, or the empty string.

url_for_ticket()

Return a URL for the given ticketing_system, ticket pair.

url_if_ticket()

Return a Ticket URL if a ticket is found.


class jgt_common.CommonAttributeList[source]

Bases: list

A list for similar objects that can be operated on as a group.

Accessing an attribute on this list instead returns a list of that attribute’s value from each member. (unless the attribute is defined here or in the base class) If any member of this list does not have that attribute, AttributeError is raised.

Setting an attribute on this list instead sets the attribute on each member.

classify_data = ('sequence', 'class')
update_all(**kwargs)[source]

Update all attributes based on the provided kwargs.

class jgt_common.Flag(initial_value=False, name='Flag')[source]

Bases: object

A settable, nameable Boolean flag.

Instances can be used in a boolean context (if/while/etc.)

Call the instance to set a new value, returning the previous value.

.value to access the value. .name to access the name.

Motivation: Importing a boolean value from another module makes a new binding; changing the imported binding does not change the imported-from-modules value. Additionally, this gives you a nice named value for str() and repr(), and a clean way to handle assigning new values and toggling existing values.

Typical use might be:

limit_thing = Flag(initial_value=False, name="Limit size of thing")

In the Command line processing code:

limit_thing(args.limit_thing)

Off in some module:

def do_thing():
    ...
    loop_size = LIMITED_SIZE if limit_thing else MAX_SIZE
    ...

Use an instance of this class where you want semantic control over code without having to expose implementation details of that code to the control point; keeping the determination of, say, “running smoke tests,” or “running in a deployment pipeline,” or “running in a PR checker,” separate from how the code will react to that condition.

classify_data = ('misc', 'class')
property name
toggle()[source]

Flip to the opposite value, return the old value.

property value
jgt_common.HEX_DIGIT_RE = '[\\da-fA-F]'

Regular expression for matching a single hex digit.

exception jgt_common.IncompleteAtTimeoutException(msg, call_result=None, timeout=None)[source]

Bases: Exception

Exception for check_until results that timeout still pending validation.

Parameters
  • msg (str) – Human readable string describing the exception.

  • call_result (any) – the final result of the call, which failed validation.

  • timeout (int,float) – the timeout at which the result was still failing.

Atributes:

call_result (any): the final result of the call, which failed validation. timeout (int,float): the timeout at which the result was still failing.

classify_data = ('looping', 'exceptions', 'class')
class jgt_common.NotEmptyList[source]

Bases: list

A list that fails to iterate if it is empty.

Iterating on this list will fail if the list is empty. This simplifies code from having to check for empty lists everywhere. Empty lists are a problem because for loop bodies don’t execute on empyt lists. thus any checks/tests/etc in a loop body would not run. In a testing context, the loop would “succeed” by not doing anything (it would fail to have checked anything) and that would be a false-positive.

classify_data = ('sequence', 'class')
static error_on_empty()[source]

Is called to return the error message when the NotEmptyList is empty.

jgt_common.OBSOLETE_TICKETING_SYSTEMS = ['JIRA', 'SNOW', 'VersionOne']

Systems for which we still support identifying Tickets, but no longer can access.

class jgt_common.ResponseInfo(response=None, description=None, response_callback=None, response_data_extract=None, **kwargs)[source]

Bases: object

Keep track of needed info about test operation responses (or any info really).

Originally created to augment/track info about requests' responses, it’s not limited or bounded by that use case. Please read response more generally as any kind of object useful for catpuring some kind of response from the system under test. In addition, arbitrary other attributes can be set on this object. To make that easier, kwargs is processed as attribute / value pairs to be set on the object, for whatever attributes make sense for your application.

Sometimes this object is keeping track of a response that is needed, but isn’t available yet. In these cases, response_callback can be set to a parameter-less function that can be called to obtain the response when the response_data property is used.

For ease of use, when the data is buried in the response or otherwise needs to be decoded, response_data_extract can be used. It should take one parameter (the response) and return the desired data from it. This function should not have any side-effects. response_data_extract is also used by the response_data property, see that property documentation for details.

Parameters
  • response (any, optional) – Whatever kind of response object needs tracking.

  • description (str, optional) – Description of this particular response.

  • response_callback (function w/no parameters, optional) – A callback to use in place of the response field.

  • response_data_extract (function w/1 parameter, optional) – A callback used to extract wanted data from the response.

  • kwargs (dict, optional) – any additional attributes to set on this object, based on the name/value pairs in kwargs.

classify_data = ('requests', 'class')
property response_data

Property that returns the data from a response.

  1. If response_callback is set, that is used to obtain the response, otherwise the response attribute is used. If the response_callback is called, it’s result is assigned to the .response attribute and the response_callback attribute is set to None. This is to prevent the callback from being invoked more than once.

  2. If response_data_extract is set, it is called on the response from step 1, and it’s return value is the value of this property. Otherwise the result of step 1 is returned as is. Note that in this step the .response attribute is not changed, the response_data_extract callback is expected to have no side-effects.

run_response_callback()[source]

Run the response_callback, if any, and set response to the result.

Set response_callback to None so iit isn’t run more than once.

class jgt_common.ResponseList[source]

Bases: jgt_common.NotEmptyList, jgt_common.CommonAttributeList

A list specialized for testing, w/ResponseInfo object items.

To best understand this class, it is important to have a strong understanding of CommonAttributeList and ResponseInfo. The common workflow for ResponseList relies greatly on those other connected pieces.

For example, you might utilize this class in a ways such as this:

>>> responses = ResponseList()
>>> responses.build_and_set(response=client.get_thing())
>>>
>>> # All the `.response` fields, see `CommonAttributeList`
>>> for response in responses.response:
...     assert response.json()["thing"]
>>>
>>>
>>> responses.set(
...     ResponseInfo(response=client.get_thing(param),
                     description=f"Getting {param}...")
...     for param in my_params
... )
build_and_set(*args, **kwargs)[source]

Create object and then set it on the list.

Create ResponseInfo object with args & kwargs, then .set it on this ResponseList.

classify_data = ('requests', 'class')
run_response_callbacks()[source]

Call run_response_callback on each item of this ReponseList.

set(resp)[source]

Clear and set the contents of this list to single object / iterator of objects.

Generators will be converted into a list to allow access more than once.

This method can be handy/useful when transforming this list’s contents from one form to another, such as:

>>> x = CommonAttributeList()
>>> ...
>>> x.set(transform(thing, doo_dad) for thing in x)
property single_item

Property - Assert this list has one item, and return that item.

jgt_common.UUID_BASIC_RE = '[\\da-fA-F]{8}-[\\da-fA-F]{4}-[\\da-fA-F]{4}-[\\da-fA-F]{4}-[\\da-fA-F]{12}'

Regular expression (RE) for matching UUID string forms.

This RE is not anchored, or delimited, for maximum reuse.

jgt_common.UUID_ISOLATED_RE = '\\b[\\da-fA-F]{8}-[\\da-fA-F]{4}-[\\da-fA-F]{4}-[\\da-fA-F]{4}-[\\da-fA-F]{12}\\b'

RE for matching an UUID that is not part of a larger “word”.

jgt_common.accumulator_for(fun)[source]

Accumulate the results of calling fun into a unique list, returning that list.

Everytime the decorated function is called, its results are appended to a list, and that list is returned instead. Each invocation of this function results in a wrapper that uses its own unique list.

NOTE: If you use this is as a decorator on a function definition, ALL invocations of the function will accumulate into a single list. This is probably not what you want.

The intent is to use this decorator “as needed”:

def get_thing_when_stable():
    '''
    Get a thing when the API returns a stable result.

    A stable result is when 3 consecutive payloads match.
    If no stable result is achieved, an IncompleteAtTimeoutException exception
    is raised.
    '''
    results = check_until(accumulator_for(api_get_thing), last_3_payloads_equal)
    return get_thing_from_payload(results[-1])
jgt_common.always_false(*args, **kwargs)[source]

Return False regardless of any provided arguments.

jgt_common.always_true(*args, **kwargs)[source]

Return True regardless of any provided arguments.

jgt_common.assert_if_values(format_if_format, error_fun=<function <lambda>>)[source]

Assert if any truthy values are returned (or yielded) from the decorated function.

Any values that the decorated function yields/returns are passed to error_fun. format_if_format, and the result of the error_fun call, are passed to format_if. The result of the format_if call is used for an assertion.

Parameters
  • format_if_format (str) – the first parameter to format_if.

  • error_fun (callable) – called on the results of the decorated function, return value will be passed as the 2nd parameter to format_if. The default function will only process truthy values. That can be changed by passing in your own custom error_fun

Returns

None - If no values are returned, or yielded, or the result of error_fun is falsey, return None.

Raises

AssertionError – as described above.

This decorator is meant to simplify code that otherwise has to manually keep track of a list of things to complain about:

errors = []
for thing in things:
   if bad(thing):
       errors.append(description_of(thing))
err_msg = format_if("blah blah blah: {}", "".join(errors))
assert not err_msg, err_msg

into simpler code:

@assert_if_values("blah blah blah: {}")
...
    for thing in things:
        if bad(thing):
            yield description_of(thing)
jgt_common.build_classification_rst_string(from_dict, for_module, category_name_mappings)[source]

Create rST for all the items in from_dict that are part of for_module.

Example

__doc__ += build_classification_rst_string(globals(), __name__, <category

mappings dict>)

The for_module parameter is needed because often globals() contains symbols imported from other modules not be documented here.

Parameters
  • from_dict (dict) – A dictionary of name to function/class mappings, such as from locals() or globals()

  • for_module (str) – The module for which rST documentation is desired.

  • category_name_mappings (str) – Map from the short hand names used in the classify() decorator to the desired category table label.

Returns

multiline rST string.

Return type

str

jgt_common.check_until(function_call, is_complete_validator, timeout=300, cycle_secs=5, logger=<Logger jgt_common (WARNING)>, fn_args=None, fn_kwargs=None)[source]

Periodically call a function until its result validates or the timeout is exceeded.

Parameters
  • function_call (function) – The function to be called

  • is_complete_validator (function) – a fn that will accept the output from function_call and return False if the call should continue repeating (still pending result), or True if the checked result is complete and may be returned.

  • timeout (int) – maximum number of seconds to “check until” before raising an exception.

  • cycle_secs (int) – how long to wait (in seconds) in between calls to function_call.

  • logger (logging.logger, optional) – a logging instance to be used for debug info, or None to suppress logging by this function.

  • fn_args (tuple, optional) – tuple of positional args to be provided to function_call

  • fn_kwargs (dict, optional) – keyword args to be provided to function_call

Returns

the result of function_call when the is_complete_validator returns any True

value.

Return type

any

Raises

jgt_common.IncompleteAtTimeoutException – if function_call’s result never satisfies the is_complete_validator before timeout is reached.

jgt_common.class_lookup = {}

This dictionary is to allow code that needs to have very late binding of a class to look up the class here, and code that needs to control another module’s use of a late bound class to set it in this dictionary.

Code accessing this dictionary should use .get() with a default value so that this module doesn’t have to import lots of things here to set up all the defaults. Code accessing this dictionary should publish which key(s) it will be using so that modules wishing to retarget those classes will know which keys to set.

Code setting values in this dictionary so do so before importing modules that will use that value. Setting values in this dictionary is only needed when the default values need to be changed.

(Motivation: the requests client logging code needs to be able to use a custom class instead of just requests.Session when being used by testing code with the Locust test runner, and it uses this dictionary to accomplish this.)

jgt_common.classify(*args)[source]

Add glossary subject category classification meta-data to it’s target.

jgt_common.cleanup_and_exit(dir_name=None, status=0, message=None)[source]

Cleanup a directory tree that was created and exit.

Parameters
  • dir_name (string) – Full path to a directory to remove (optional)

  • status (int) – Exit code to use for exit (optional)

  • message (string) – Message to print to standard error (optional)

jgt_common.default_if_none(value, default)[source]

Return default if value is None else value.

Use because:
  • no chance of the value stutter being mistyped, speeds up code reading time.

  • easier to read when value or default are complex expressions.

  • can save having to create local variable(s) to shorten the if .. is None ... form.

jgt_common.dict_from(iterable, key_transform=<function identity>, value_transform=<function identity>)[source]

Build a new dict based on turning values from an iterable into key/value pairs.

Example

A common use case would be to build a dict from a list of values, using the list values as keys, and building dict values as a transform from the list values.:

> l = [1, 2, 3, 4, 5]
> result = dict_from(l, value_transform=lambda n: n**2)
# {1: 1, 2: 4, 3: 9, 4: 16, 5: 25}
Parameters
  • iterable (iterable) – the source for the dictionary to be created

  • key_transform (function) – Takes a key and returns a new key to use.

  • value_transform (function) – Take a value and returns a new value to use.

Returns

A new dictionary with key and values as transformed.

Return type

dict

jgt_common.dict_strip_value(a_dict, value=None)[source]

Return a new dict based on stripping out any key with the given value.

Note

The default value None is chosen because it is a common case. Unlike other functions, value None is literally the value None.

Parameters
  • a_dict (dict) – A dictionary to strip values from.

  • value – Any value that should be stripped from the dictionary.

Returns

A new dictionary without key/value pairs for the given value.

Return type

dict

jgt_common.dict_transform(a_dict, key_transform=<function identity>, value_transform=<function identity>)[source]

Return a new dict based on transforming the keys and/or values of a_dict.

Parameters
  • a_dict (dict) – the source dictionary to process

  • key_transform (function) – Takes a key and returns a new key to use.

  • value_transform (function) – Takes a value and returns a new value to use.

Returns

A new dictionary with keys and values as transformed.

Return type

dict

jgt_common.display_name(path, package_name='')[source]

Create a human-readable name for a given project.

Determine the display name for a project given a path and (optional) package name. If a display_name.txt file is found, the first line is returned. Otherwise, return a title-cased string from either the base directory or package_name (if provided).

Parameters
  • path (str) – Path for searching

  • package_name (str) – Sphinx-style, dot-delimited package name (optional)

Returns

A display name for the provided path

Return type

str

jgt_common.error_if(check, status=None, message='')[source]

Exit the program if a provided check is true.

Exit the program if the check is true. If a status is provided, that code is used for the exit code; otherwise the value from the check is used. An optional message for standard error can also be provided.

Parameters
  • check – Anything with truthiness that can check if the program should exit or not

  • status (int) – Exit code to use for exit (optional)

  • message (string) – Message to print to standard error if check is True (optional)

jgt_common.execute_command_list(commands_to_run, verbose=True)[source]

Execute each command in the list.

If any command fails, print a helpful message and exit with that status.

jgt_common.exit(status=0, message=None)[source]

Exit the program and optionally print a message to standard error.

Parameters
  • status (int) – Exit code to use for exit (optional)

  • message (string) – Message to print to standard error (optional)

jgt_common.fib_or_max(fib_number_index, max_number=None)[source]

Get the nth Fibonacci number or max_number, which ever is smaller.

This can be used for retrying failed operations with progressively longer gaps between retries, cap’d at the max_number paraemter if given.

jgt_common.filter_dict(a_dict, keep_key=<function always_true>, keep_value=<function always_true>)[source]

Filter a dictionary based on truthiness.

Return a new dict based on keeping only those keys _and_ values whose function returns True.

Parameters
  • a_dict (dict) – A dictionary to filter values from.

  • keep_key (function) – Return True if the key is to be kept.

  • keep_value (function) – Return True if the value is to be kept.

Returns

A new dictionary with only the desired key, value pairs.

Return type

dict

jgt_common.filter_lines(line_filter, lines, return_type=None)[source]

Filter the given lines based on the given filter function.

This function by default will return the same type that it is given. If you’d like, you may return a different type by providing the return_type parameter.

The expected values for the the lines parameter type and the return_type value are str and list. Any other types/values may cause unexpected behavior.

Parameters
  • line_filter (Callable) – The callable function to be used to filter each line. It should take a single string parameter and return a boolean.

  • lines (Union[str, List[str]]) – Either a string with newline characters splitting the lines, or a list of lines as strings.

  • return_type (type) – The desired return type. If not provided, the type of the lines parameter will be used.

Returns

The filtered lines.

Return type

Union[str, List[str]]

jgt_common.first_line_of_doc_string(thing)[source]

Get the first non-empty line of the doc string for thing, if there is one.

If thing is a class and has no doc string, return the first line of the __init__ method doc string if there is one.

jgt_common.format_if(format_str, content)[source]

Return a message string with a formatted value if any content value is present.

Useful for error-checking scenarios where you want a prepared error message if failures are present (passed in via content), or no message if no failures.

Parameters
  • format_str (str) – A message string with a single format brace to be filled

  • content (str) – A value to be filled into the format_str if present

Returns

either the format_str with content included if content present, or an empty string if no content.

Return type

str

jgt_common.generate_random_string(prefix='', suffix='', size=8, choose_from=None)[source]

Generate a random string of the specified size.

Parameters
  • prefix (str, optional) – String to prepend to the beginning of the random string.

  • suffix (str, optional) – String to append to the end of the random string.

  • size (int, optional) – Number of characters the random string should have. (defaults to 8)

  • choose_from (str) – A string containing the characters from which the randomness will be chosen. If not provided, it will choose from lowercase letters and digits.

Returns

A randomly generated string.

Return type

str

Raises

AssertionError – if the given length is incompatible with prefix/suffix length

Examples

>>> generate_random_string()
'vng345jn'
>>> generate_random_string(prefix='Lbs-', suffix='-test', size=15)
'Lbs-js7eh9-test'
>>> generate_random_string(prefix='Lbs-', size=15)
'Lbs-js7eh98sfnk'
>>> generate_random_string(suffix='-test', size=15)
'8sdfjs7eh9-test'
>>> generate_random_string(choose_from="aeiou")
'uiiaueea'
jgt_common.get_file_contents(*paths)[source]

Get the contents of a file as a Python string.

Parameters

*paths – All paths that lead to the file whole contents are to be retrieved.

Returns

The contents of the file.

Return type

str

jgt_common.get_file_docstring(file_path)[source]

Get the full docstring of a given python file.

Parameters

file_path (str) – The path to the file whose docstring should be gathered and returned.

Returns

The python file’s docstring.

Return type

str

jgt_common.identity(x, *args)[source]

Build simple identity function based on provided parameters.

A single parameter is returned as is, multiple parameters are returned as a tuple.

From https://stackoverflow.com/questions/8748036/is-there-a-builtin-identity-function-in-python # noqa: E501 Not the top voted answer, but it handles both single and multiple parameters.

jgt_common.index_or_default(a_list, value, default=- 1)[source]

Return the index of a value from a list, or a default if not found.

Parameters
  • a_list (list) – A list from which to find the index

  • value (any) – the list item whose index is sought

  • default (any) – the value to return if the value is not in the list

Returns

an index value (int) for the list item, or default value (any)

Return type

int,any

jgt_common.is_iterable(item)[source]

Return True if item is iterable, False otherwise, using an iter(item) test.

From the Python documentation for class collections.abc.Iterable:

Checking isinstance(obj, Iterable) detects classes that are registered as Iterable or that have an __iter__() method, but it does not detect classes that iterate with the __getitem__() method. The only reliable way to determine whether an object is iterable is to call iter(obj).

jgt_common.list_from(item)[source]

Generate a list from a single item or an iterable.

Any item that is “false-y”, will result in an empty list. Strings and dictionaries will be treated as single items, and not iterable.

Parameters

item – A single item or an iterable.

Returns

A list from the item.

Return type

list

Examples

>>> list_from(None)
[]
>>> list_from('abcd')
['abcd']
>>> list_from(1234)
[1234]
>>> list_from({'abcd': 1234})
[{'abcd': 1234}]
>>> list_from(['abcd', 1234])
['abcd', 1234]
>>> list_from({'abcd', 1234})
['abcd', 1234]
jgt_common.must_be_in_virtual_environment(exit_code=1, message='Must be running in a Python virtual environment, aborting!')[source]

Ensure the current process is running in a virtual environment.

Parameters
  • exit_code (int, optional) – Exit code to use if not running in a virtual environment.

  • message (string, optional) – Message to print if not running in a virtual environment.

jgt_common.must_get_key(a_dict, key)[source]

Either return the value for the key, or raise an exception.

The exception will indicate what the valid keys are. Inspired by Gherkin steps so that a typo in the Gherkin will result in a more helpful error message than the stock KeyError.

Parameters
  • a_dict (dict) – Dictionary with the values

  • key (str) – The key whose value is desired

Returns

The value found on the key

Raises

KeyError – if the given key is not present in the dictionary.

jgt_common.must_get_keys(a_dict, *keys)[source]

Either return the value found for the keys provided, or raise an exception.

Parameters
  • a_dict (dict) – Dictionary with the values

  • keys (str) – The key or keys whose value is desired

Returns

The value found on the final key

Raises

KeyError – if any of the given keys are not present in the dictionary.

jgt_common.no_nones(iterable)[source]

Return a list of the non-None values in iterable.

jgt_common.no_op(*args, **kwargs)[source]

Reusable no-op function.

jgt_common.only_item_of(item_sequence, label='')[source]

Assert item_sequence has only one item, and return that item.

jgt_common.padded_list(iterable, size, padding=None)[source]

Generate a fixed-length list from an iterable, padding as needed.

Parameters
  • iterable (iterable) – Any iterable that needs padding

  • size (int) – The length for the returned list

  • padding – Any value that should be used to pad an iterable that is too short

Returns

The iterable parameter converted to a list, up to size, padded as needed.

Return type

list

jgt_common.percent_diff(a, b, precision=2)[source]

Get percentage difference, out to precision places.

jgt_common.re_for_hex_digits(length)[source]

Return regular expression for matching exactly length hex digits.

jgt_common.retry_on_exceptions(max_retry_count, exceptions, max_retry_sleep=30)[source]

Retry a function based on provided parameters.

A wrapper to retry a function max_retry_count times if any of the given exceptions are raised.

In the event the exception/exceptions are raised, this code will sleep for ever increasing amounts of time (using the fibonacci sequence) but capping at max_retry_sleep seconds.

Parameters
  • max_retry_count (int) – The maximum number of retries, must be > 0..

  • exceptions (exception or tuple of exceptions) – The exceptions to catch and retry on.

  • max_retry_sleep (int, float) – The maximum time to sleep between retries.

jgt_common.safe_run(commands, cwd=None)[source]

Run the given list of commands, only return if no error.

If there is an error in attempting or actually running the commands, error messages are printed to stdout and sys.exit() will be called.

jgt_common.string_to_list(source, sep=',', maxsplit=- 1, chars=None)[source]

.split() a string into a list and .strip() each piece.

For handling lists of things, from config files, etc.

Parameters
  • source (str) – the source string to process

  • sep (str, optional) – The .split sep (separator) to use.

  • maxsplit (int, optional) – The .split maxsplit parameter to use.

  • chars (str, optional) – The .strip chars parameter to use.

jgt_common.ticketing_system_for(ticket)[source]

Return the Ticketing System/Type for the given ticket, or the empty string.

jgt_common.truths_from(iterable)[source]

Return a list of the truthy values in iterable.

jgt_common.url_for_ticket(ticketing_system, ticket)[source]

Return a URL for the given ticketing_system, ticket pair.

Does NOT validate that ticket string “looks” like a ticket for the ticketing system.

Raises if ticketing_system is not a supported system.

jgt_common.url_if_ticket(ticket)[source]

Return a Ticket URL if a ticket is found.

Parameters

ticket (string) – Possible ticket ID

Returns

Either a URL if a string is found, otherwise an empty string.

Return type

string

jgt_common.var_from_env(var_name)[source]

Try to get a value from an environment variable.

Get an environment variable and raise an error if not set / has an empty value.

Returns

The value of the environment variable.

Return type

str

Raises

ValueError – if the variable name is not set or has an empty value.

Submodules

jgt_common.assert_ module

Convenience functions for doing asserts with helpful names and helpful messages.

jgt_common.assert_.almost_equal(actual, expected, places=2, msg='')

Assert that actual and expected are within places equal.

jgt_common.assert_.any_in(a_sequence, b_sequence, msg='')

Assert that at least one member of a_sequence is in b_sequence.

jgt_common.assert_.assert_if_truthy(fun)[source]

(Decorator) Assert if fun returns a truthy value (error indicator).

jgt_common.assert_.eq(expected, actual, msg='')

Assert that the values are equal.

jgt_common.assert_.greater(a, b, msg='')

Assert that a > b.

jgt_common.assert_.greater_equal(a, b, msg='')

Assert that a >= b.

jgt_common.assert_.is_close(a, b, msg='', **isclose_kwargs)

Assert that math.isclose returns True based on the given values.

jgt_common.assert_.is_in(value, sequence, msg='')

Assert that value is in the sequence.

jgt_common.assert_.is_instance(value, of_type, msg='')

Assert that value is an instance of of_type.

jgt_common.assert_.is_not_empty(sequence, msg='')

Cheeck that sequence is not empty.

Semantically more descriptive than just testing sequence for truthyness.

Sequences and containers in python are False when empty, and True when not empty. This helper reads better in the test code and in the error message.

jgt_common.assert_.is_not_none(a, msg='')

Assert a is not None.

jgt_common.assert_.is_singleton_list(sequence, item_description='something', msg='')

Assert that the sequence has exactly one item (of item_description).

jgt_common.assert_.less(a, b, msg='')

Assert that a < b.

jgt_common.assert_.less_equal(a, b, msg='')

Assert that a <= b.

jgt_common.assert_.not_eq(expected, actual, msg='')

Assert that the values are not equal.

jgt_common.assert_.not_in(item, sequence, msg='')

Assert that item is not in sequence.

jgt_common.check module

Convenience functions for doing checks with helpful names and helpful messages.

All the functions in this module return the empty string(“”) when the check passes and a description of how the check has failed when it fails. This is intentional so that the truthyness of strings and falsyness of empty strings can be exploited for use with format_if or assert or …

jgt_common.check.almost_equal(actual, expected, places=2, msg='')[source]

Check that actual and expected are within places equal.

jgt_common.check.any_in(a_sequence, b_sequence, msg='')[source]

Check that at least one member of a_sequence is in b_sequence.

jgt_common.check.eq(expected, actual, msg='')[source]

Check that the values are equal.

jgt_common.check.greater(a, b, msg='')[source]

Check that a > b.

jgt_common.check.greater_equal(a, b, msg='')[source]

Check that a >= b.

jgt_common.check.is_close(a, b, msg='', **isclose_kwargs)[source]

Check that math.isclose returns True based on the given values.

jgt_common.check.is_in(value, sequence, msg='')[source]

Check that value is in the sequence.

jgt_common.check.is_instance(value, of_type, msg='')[source]

Check that value is an instance of of_type.

jgt_common.check.is_not_empty(sequence, msg='')[source]

Cheeck that sequence is not empty.

Semantically more descriptive than just testing sequence for truthyness.

Sequences and containers in python are False when empty, and True when not empty. This helper reads better in the test code and in the error message.

jgt_common.check.is_not_none(a, msg='')[source]

Check a is not None.

jgt_common.check.is_singleton_list(sequence, item_description='something', msg='')[source]

Check that the sequence has exactly one item (of item_description).

jgt_common.check.less(a, b, msg='')[source]

Check that a < b.

jgt_common.check.less_equal(a, b, msg='')[source]

Check that a <= b.

jgt_common.check.not_eq(expected, actual, msg='')[source]

Check that the values are not equal.

jgt_common.check.not_in(item, sequence, msg='')[source]

Check that item is not in sequence.

jgt_common.futures module

Helpful functions for using concurrent.futures with a shared thread pool executor.

Terminology:

  • fdict - futures dictionary - as in the examples for the concurrent.futures module documentation, a dictionary whose keys are futures and whose values are related to the work being done. See each function for details on what the values will be.

Purpose:

These helpers are meant to be used for “spot” parallelism, not for whole application parallelism. In places where the code needs to do a whole bunch of slow things that can be done in parallel (API calls, etc) and then “goes synchronous again.” (Testing is one such use case.)

Archicture:

One executor thread pool is used by all of these functions. To support the use of these functions in a “library” kind of way, configuring the size of the thread pool and shutting it down (if ever used) are done by calling separate functions. This is so an application can use configuration data for the set up without having to spread knowledge of that configuration data to any function(s) that want to use these functions.

The executor thread pool defined here is mainly meant for use by the other functions defined here, but users of this module are free to use it themselves directly as needed.

jgt_common.futures.as_completed_item_result(fdict)[source]

Yield (item, future.result()) from fdict, as each future completes.

The order of values in the tuples allow this function to be used to create a dict mapping from an item to its result:

futures = run_each(iterable, func)
results_map = dict(as_completed_item_result(futures))
jgt_common.futures.as_completed_result(futures)[source]

Yield each futures .result() as each future in futures completes.

Because iterating over a dictionary iterates over it’s keys, futures can be an fdict, or a list of futures, … any other iterable of futures.

jgt_common.futures.get_executor()[source]

Get the shared ThreadPoolExecutor.

Returns

the shared thread pool executor.

Return type

ThreadPoolExecutor

Raises

TypeError – if set_thread_pool_size() was not called first.

jgt_common.futures.result_from_each(iterable, func)[source]

Shorthand for as_completed_result(run_each(iterable, func)).

When you want all the results from calling func on the items from iterable, but you don’t need to know which result came from which item or in which order.

jgt_common.futures.run_each(iterable, func)[source]

Call func on each item in iterable, using a future.

The returned fdict’s futures (keys of the fdict) may or may not be completed by the time this function returns. It is up to the caller to decide how to harvest the results from the futures.

Can be used directly, but is mostly used under the covers by the other functions in this module.

Parameters
  • iterable (any) – Any iterable.

  • func (callable) – will be called with one item from iterable.

Returns

Mapping from a future to the item from iterable used to make it.

Return type

fdict

jgt_common.futures.set_each(iterable, field, func)[source]

Set field on each item from iterable to the value of func(item).

Shorthand for set_when_completed(field, run_each(iterable, func)).

A more general form of set_response_on_each.

jgt_common.futures.set_response_on_each(iterable, func)[source]

Shorthand for set_response_when_completed(run_each(iterable, func)).

This is primarily for those using jgt_common’s ResponseList and ResponseInfo objects.

Example

Instead of writing:

for item in work_list:
    item.response = client.do_work(item.input)

Parallelize that work:

set_response_on_each(work_list, lambda item: client.do_work(item.input))
jgt_common.futures.set_response_when_completed(fdict)[source]

Set .response on each value from fdict to its future’s result.

jgt_common.futures.set_thread_pool_size(max_workers)[source]

Set the size for the shared ThreadPoolExecutor.

jgt_common.futures.set_when_completed(field, fdict)[source]

Set the given field on each value from fdict to its future’s result.

jgt_common.futures.shutdown_executor()[source]

If a ThreadPoolExecutor was started, shut it down.

jgt_common.http_helpers module

Tools for simplifying HTTP requests and responses.

JSON related functions

safe_json_from()

Accept a response object and attempts to return the JSON-decoded data.

Logging related functions

call_with_custom_logger()

Context Manager to make a call with a given curl_logger in place of the default.

Requests’ Response object related functions

check_response_status_code()

Check that a response’s status code matches an expected status description.

create_error_message()

Create a detailed error message based on an API call.

get_data_from_response()

Accept a response object and returns a dict of the data contained within.

get_data_list()

Get the full list of data from within a response, if the data is a list.

response_if_status_check()

Validate a response matches the expected status code before returning.

safe_json_from()

Accept a response object and attempts to return the JSON-decoded data.

safe_request_validator()

Wrap a check_request_until validator in additional response checks.

validate_response_status_code()

Assert that a response’s status code matches an expected status code.

HTTP Status code functions

check_response_status_code()

Check that a response’s status code matches an expected status description.

is_status_code()

Determine if a given status code matches the expected result.

response_if_status_check()

Validate a response matches the expected status code before returning.

validate_response_status_code()

Assert that a response’s status code matches an expected status code.

String related functions

create_error_message()

Create a detailed error message based on an API call.

format_items_as_string_tree()

Format the items provided as a string representing a ‘tree’ format.


jgt_common.http_helpers.call_with_custom_logger(call, curl_logger)[source]

Context Manager to make a call with a given curl_logger in place of the default.

If curl_logger is a last-only logger, call .done() on context exit. NOTE: for last-only loggers, the log will not be written until you exit the context.

Example

Create a new callable based on ‘call’ that uses the logger ‘logger_to_use’:

with call_with_custom_logger(call, logger_to_use) as call_using_logger:
    some_result = check_until(call_using_logger, ...)
jgt_common.http_helpers.check_response_status_code(expected_status_description, response, call_description=None, additional_info=None)[source]

Check that a response’s status code matches an expected status description.

Parameters
  • expected_status_description (str, int) – The expected HTTP response status reason or code.

  • response (Response) – The python requests library response to validate.

  • call_description (str) – additional details about the call placed (optional)

  • additional_info (dict) – Additional Info to be passed to create_error_message

Returns

Empty string if response status code matches, or a detailed error message

otherwise

Return type

str

jgt_common.http_helpers.create_error_message(summary_line, request, response_content, additional_info=None)[source]

Create a detailed error message based on an API call.

This message will include information about the request and response, cutting off the response content at the set MAX_ERROR_MESSAGE_CONTENT_LENGTH. Additional information can be included in the message by including the additional_info parameter.

Parameters
  • summary_line (str) – The first line of the message that describes the error.

  • request (Request) – The python requests library Request.

  • response_content (str) – The response content to be included in the response.

  • additional_info (dict) – A dictionary containing additional information to be included in the message. The keys and values will be separated by a ":" in the message. (optional)

Returns

The complete and detailed error message.

Return type

str

jgt_common.http_helpers.format_items_as_string_tree(*items)[source]

Format the items provided as a string representing a “tree” format.

Any lists or nested lists will be flattened, with each level of nested lists getting its own level of indentation. Each item, once flattened with the appropriate level of indentation provided, will be placed on its own line.

Returns

An indented “tree” format of the items.

Return type

str

Example

>>> format_items_as_string_tree('ex1', 'ex2', ['nested1', 'nested2',
>>>                             ['double1'], ['double2']], 'ex3')
    ex1
    ex2
        nested1
        nested2
            double1
            double2
    ex3
jgt_common.http_helpers.get_data_from_response(response, dig_layers=None, check_empty=True, first_only=True)[source]

Accept a response object and returns a dict of the data contained within.

Parameters
  • response (requests.models.Response) – A Response object from a requests call

  • dig_layers (list) – List of keys to “dig” down into the response before returning

  • check_empty (bool) – If True, raises an AssertionError if the payload is empty

  • first_only (bool) – Strip the list wrapping the data and return the first result

Returns

The data payload.

Examples

>>> response.json() == {"data": [{"key": "value"}, {"key": "value2"}],
                        "other": ""}
True
>>>
>>> get_data_from_response(response, dig_layers=['data'])
{'key': 'value'}
>>> get_data_from_response(response, dig_layers=['data'], first_only=False)
[{'key': 'value'}, {'key': 'value2'}]
>>> get_data_from_response(response, dig_layers=['other'])
AssertionError: Payload was empty: ''
>>> get_data_from_response(response, dig_layers=['other'], check_empty=False)
""
jgt_common.http_helpers.get_data_list(response, **kwargs)[source]

Get the full list of data from within a response, if the data is a list.

Helper around get_data_for_response that returns the full list of data if the data is in a list format, rather than the default of returning the first object inside the list.

For **kwargs values and more information, see get_data_from_response().

jgt_common.http_helpers.is_status_code(expected_status_description, actual_status_code)[source]

Determine if a given status code matches the expected result.

Parameters
  • expected_status_description (str, int) – The result to match. Can be: - A Python requests libarary status reason (“OK”) - A phrase matching a _status_code_ranges key (‘a client error’) - An integer for the response status code (404)

  • actual_status_code (int) – the status code from the response to be validated

Returns

Whether or not the actual status code matches the expected description.

Return type

bool

Raises

AssertionError – if an invalid expected_status_description value is provided.

jgt_common.http_helpers.response_if_status_check(call_description, response, target_status='a successful response')[source]

Validate a response matches the expected status code before returning.

Parameters
  • call_description (str) – Text description of the call placed (e.g. “get schedule list”)

  • response (requests.Response) – the response object to validate.

  • target_status (str, int) – The expected HTTP response status reason or code.

Returns

if the response status validation succeeds

Return type

requests.Response

Raises

AssertionError – if the response status validation fails

jgt_common.http_helpers.safe_json_from(response, description='')[source]

Accept a response object and attempts to return the JSON-decoded data.

Parameters
  • response (requests.models.Response) – a Response object from a requests call

  • description (str, optional) – details about the response expected.

Returns

The JSON-decoded data from the response

Raises

AssertionError – if the JSON data cannot be decoded properly.

jgt_common.http_helpers.safe_request_validator(inner_validator, max_failures=5, logger=None)[source]

Wrap a check_request_until validator in additional response checks.

Will “fail fast” in the case of an unauthorized response, and will support retrying calls if a non-200-level reponse status is returned.

Parameters
  • inner_validator (function) – case-specific response validator function, should return False if condition is not met and re-checking should continue, or True if target status is complete and result should be returned.

  • max_failures (int) – total number of retries per cycle a call should be allowed before failing permanently and returning the last response

  • logger (logging.logger, optional) – a logger to handle debug messages from the validator

Returns

response-status-checking decorated version of inner_validator

Return type

function

jgt_common.http_helpers.validate_response_status_code(expected_status_description, response, err_prefix='', **kwargs)[source]

Assert that a response’s status code matches an expected status code.

Parameters
  • expected_status_description (str, int) – The expected HTTP response status reason or code.

  • response (Response) – The python requests library response to validate.

  • err_prefix (str, optional) – Prefix, if any, to the error message, when an assertion is raised.

  • kwargs – additional keyword args to be passed on to check_response_status_code.

Raises

AssertionError – if the response status validation fails

jgt_common.tag_to_url module

Pre-defined tag-to-url patterns.

jgt_common.uuid_replacer module

Replace UUIDs helper.

class jgt_common.uuid_replacer.UUIDLineReplacer(template=None)[source]

Bases: object

Track and Replace UUIDs on a line-by-line basis.

Instances should be called with a line to process and will return the substituted line.

When you are all done, if you want, you can call uuid_mappings to get a list of the substitutions that were made.

Create a new UUID replacer.

Parameters

template (str) – A .format template to use for generating the UUID replacement values. It is expected to have a placeholder to format one numeric parameter.

uuid_mappings()[source]

Return a list of lines of all the substitutions done.

jgt_common.uuid_replacer.UUID_REPLACEMENT_TEMPLATE = ',,UUID-{:03d},,'

Template used to generate shorter version for a UUID.

Is expected to hold exaclty one numeric parameter substitution.

jgt_common.uuid_replacer.main()[source]

Command-line interace for replacing UUIDs with placeholders.

jgt_common.uuid_replacer.uuid_replace(src, dest, template=',,UUID-{:03d},,')[source]

Replace UUIDs in all the lines in src and write to dest.

Parameters
  • src (file) – a file opened for read

  • dest (file) – a file opened for write.

After processing the contents of src into dest, a glossary is then written to dest.