pyoaev.utils

Module Contents

class pyoaev.utils._StdoutStream

pyoaev.utils.get_content_type(content_type: str | None) → str

pyoaev.utils.response_content(response: requests.Response, streamed: bool, action: Callable[[bytes], None] | None, chunk_size: int, *, iterator: bool) → bytes | Iterator[Any] | None

pyoaev.utils.copy_dict(*, src: Dict[str, Any], dest: Dict[str, Any]) → None

class pyoaev.utils.EncodedId

Bases: str

pyoaev.utils.remove_none_from_dict(data: Dict[str, Any]) → Dict[str, Any]

class pyoaev.utils.EnhancedJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: json.JSONEncoder

Extensible JSON <https://json.org> encoder for Python data structures.

Supports the following objects and types by default:

Python	JSON
dict	object
list, tuple	array
str	string
int, float	number
True	true
False	false
None	null

To extend this to recognize other objects, subclass and implement a .default() method with another method that returns a serializable object for o if possible, otherwise it should call the superclass implementation (to raise TypeError).

default(o)

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 super().default(o)

class pyoaev.utils.RequiredOptional

required: Tuple[str, Ellipsis] = ()

optional: Tuple[str, Ellipsis] = ()

exclusive: Tuple[str, Ellipsis] = ()

validate_attrs(*, data: Dict[str, Any], excludes: List[str] | None = None) → None

class pyoaev.utils.CustomJsonFormatter(*args, json_default: pythonjsonlogger.core.OptionalCallableOrStr = None, json_encoder: pythonjsonlogger.core.OptionalCallableOrStr = None, json_serializer: Callable | str = json.dumps, json_indent: int | str | None = None, json_ensure_ascii: bool = True, **kwargs)

Bases: pythonjsonlogger.json.JsonFormatter

JSON formatter using the standard library’s [json](https://docs.python.org/3/library/json.html) for encoding

add_fields(log_record, record, message_dict)

Extract fields from a LogRecord for logging

This method can be overridden to implement custom logic for adding fields.

Args:

log_record: data that will be logged record: the record to extract data from message_dict: dictionary that was logged instead of a message. e.g

logger.info({“is_this_message_dict”: True})

pyoaev.utils.setup_logging_config(level, json_logging=True)

class pyoaev.utils.AppLogger(level, json_logging=True, name: str = __name__)

log_level

json_logging = True

local_logger

static prepare_meta(meta=None)

static setup_logger_level(lib, log_level)

debug(message, meta=None)

info(message, meta=None)

warning(message, meta=None)

error(message, meta=None)

pyoaev.utils.logger(level, json_logging=True)

class pyoaev.utils.PingAlive(api, config, logger, ping_type)

Bases: threading.Thread

A class that represents a thread of control.

This class can be safely subclassed in a limited fashion. There are two ways to specify the activity: by passing a callable object to the constructor, or by overriding the run() method in a subclass.

ping_type

api

config

logger

in_error = False

exit_event

ping() → None

run() → None

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

stop() → None