prozorro_sale.tools package

exception prozorro_sale.tools.ApplicationError[source]

Bases: Exception

Base class for Application exceptions.

exception prozorro_sale.tools.ConcurrencyError[source]

Bases: RuntimeError

Base class for Concurrency exceptions.

class prozorro_sale.tools.DefaultTypesEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]

Bases: json.encoder.JSONEncoder

Custom encoder for json

default(obj)[source]

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)
prozorro_sale.tools.check_required_env_vars(required_vars: set)[source]

Method to check required environment variables.

Parameters

required_vars (set) – Data

Raises

ApplicationError – if not defined variables presented

prozorro_sale.tools.check_required_fields(required_fields: set, data: dict)[source]

Method for checking expected values in data.

Parameters

  • required_fields (set) – expected inclusion in data

  • data (dict) – Data

Returns

required fields

Return type

dict

class prozorro_sale.tools.classproperty(getter)[source]

Bases: object

Classproperty wrapper

prozorro_sale.tools.default_type_encoder(data)[source]

Method for json dumps with custom encoder.

Parameters

data (dict) –

Returns

Return type

str

prozorro_sale.tools.retry_on(error, times=5)[source]

Request retries wrapper.

async prozorro_sale.tools.retry_on_concurrency_error_middleware(request, handler)[source]

Middleware for request retries if caught concurrency.