Function _is_queue_like_object

Lib/logging/config.py:500–525

Check that *obj* implements the Queue API.

(obj)

        return value

def _is_queue_like_object(obj):
    """Check that *obj* implements the Queue API."""
    if isinstance(obj, (queue.Queue, queue.SimpleQueue)):
        return True
    # defer importing multiprocessing as much as possible
    from multiprocessing.queues import Queue as MPQueue
    if isinstance(obj, MPQueue):
        return True
    # Depending on the multiprocessing start context, we cannot create
    # a multiprocessing.managers.BaseManager instance 'mm' to get the
    # runtime type of mm.Queue() or mm.JoinableQueue() (see gh-119819).
    #
    # Since we only need an object implementing the Queue API, we only
    # do a protocol check, but we do not use typing.runtime_checkable()
    # and typing.Protocol to reduce import time (see gh-121723).
    #
    # Ideally, we would have wanted to simply use strict type checking
    # instead of a protocol-based type checking since the latter does
    # not check the method signatures.
    #
    # Note that only 'put_nowait' and 'get' are required by the logging
    # queue handler and queue listener (see gh-124653) and that other
    # methods are either optional or unused.
    minimal_queue_interface = ['put_nowait', 'get']
    return all(callable(getattr(obj, method, None))
               for method in minimal_queue_interface)

class DictConfigurator(BaseConfigurator):
    """