pyoaev.utils
============

.. py:module:: pyoaev.utils






Module Contents
---------------

.. py:class:: _StdoutStream

.. py:function:: get_content_type(content_type: Optional[str]) -> str

.. py:function:: response_content(response: requests.Response, streamed: bool, action: Optional[Callable[[bytes], None]], chunk_size: int, *, iterator: bool) -> Optional[Union[bytes, Iterator[Any]]]

.. py:function:: copy_dict(*, src: Dict[str, Any], dest: Dict[str, Any]) -> None

.. py:class:: EncodedId

   Bases: :py:obj:`str`


.. py:function:: remove_none_from_dict(data: Dict[str, Any]) -> Dict[str, Any]

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

   Bases: :py:obj:`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``).



   .. py:method:: 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)




.. py:class:: RequiredOptional

   .. py:attribute:: required
      :type:  Tuple[str, Ellipsis]
      :value: ()



   .. py:attribute:: optional
      :type:  Tuple[str, Ellipsis]
      :value: ()



   .. py:attribute:: exclusive
      :type:  Tuple[str, Ellipsis]
      :value: ()



   .. py:method:: validate_attrs(*, data: Dict[str, Any], excludes: Optional[List[str]] = None) -> None


.. py:class:: CustomJsonFormatter(*args, json_default: pythonjsonlogger.core.OptionalCallableOrStr = None, json_encoder: pythonjsonlogger.core.OptionalCallableOrStr = None, json_serializer: Union[Callable, str] = json.dumps, json_indent: Optional[Union[int, str]] = None, json_ensure_ascii: bool = True, **kwargs)

   Bases: :py:obj:`pythonjsonlogger.json.JsonFormatter`


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


   .. py:method:: 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})`



.. py:function:: setup_logging_config(level, json_logging=True)

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

   .. py:attribute:: log_level


   .. py:attribute:: json_logging
      :value: True



   .. py:attribute:: local_logger


   .. py:method:: prepare_meta(meta=None)
      :staticmethod:



   .. py:method:: setup_logger_level(lib, log_level)
      :staticmethod:



   .. py:method:: debug(message, meta=None)


   .. py:method:: info(message, meta=None)


   .. py:method:: warning(message, meta=None)


   .. py:method:: error(message, meta=None)


.. py:function:: logger(level, json_logging=True)

.. py:class:: PingAlive(api, config, logger, ping_type)

   Bases: :py:obj:`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.



   .. py:attribute:: ping_type


   .. py:attribute:: api


   .. py:attribute:: config


   .. py:attribute:: logger


   .. py:attribute:: in_error
      :value: False



   .. py:attribute:: exit_event


   .. py:method:: ping() -> None


   .. py:method:: 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.




   .. py:method:: stop() -> None