Python typing 模块，Callable() 实例源码

def apply(func,         # type: Callable[..., bytes]
              args=(),      # type: Sequence[AnyStr]
              exe=None,     # type: Optional[str]
              depfiles=(),  # type: Sequence[str]
              cache=None    # type: Optional[Cache]
              ):
        """Applies func(*args) when the result is not present in the cache.
        The result of func(*args) must be bytes and must not be None which is used as
        cache-miss indicator. After evaluation of func the result is stored in the cache.
        """
        key, value = None, None
        if cache is not None:
            hashobj = cache.mixtohash(args, exe=exe, depfiles=depfiles)
            key = hashobj.hexdigest()
            value = cache.get(key)
        if value is None:
            value = func(*args)
            if key is not None:
                cache.set(key, value)
        return value

def register_command(handler: Callable[[argparse.Namespace], None],
                     main_parser: Optional[ArgParserType]=None,
                    ) -> Callable[[argparse.Namespace], None]:
    if main_parser is None:
        main_parser = global_argparser
    if id(main_parser) not in _subparsers:
        subparsers = main_parser.add_subparsers(title='commands',
                                                dest='command')
        _subparsers[id(main_parser)] = subparsers
    else:
        subparsers = _subparsers[id(main_parser)]

    @functools.wraps(handler)
    def wrapped(args):
        handler(args)

    doc_summary = handler.__doc__.split('\n\n')[0]
    inner_parser = subparsers.add_parser(handler.__name__.replace('_', '-'),
                                         description=handler.__doc__,
                                         help=doc_summary)
    inner_parser.set_defaults(function=wrapped)
    wrapped.register_command = functools.partial(register_command,
                                                 main_parser=inner_parser)
    wrapped.add_argument = inner_parser.add_argument
    return wrapped

def __init__(self,
                 get_raw_audio: Callable[[], ndarray],
                 sample_rate: int = 16000,
                 id: Optional[str] = None,
                 label: Optional[str] = "nolabel",
                 fourier_window_length: int = 512,
                 hop_length: int = 128,
                 mel_frequency_count: int = 128,
                 label_with_tags: str = None,
                 positional_label: Optional[PositionalLabel] = None):
        super().__init__(id=id, label=label)

        # The default values for hop_length and fourier_window_length are powers of 2 near the values specified in the wave2letter paper.
        self.get_raw_audio = get_raw_audio
        self.sample_rate = sample_rate
        self.fourier_window_length = fourier_window_length
        self.hop_length = hop_length
        self.mel_frequency_count = mel_frequency_count
        self.label_with_tags = label_with_tags
        self.positional_label = positional_label

def randomly_grouped_by(key_from_example: Callable[[LabeledExample], Any], training_share: float = .9) -> Callable[
        [List[LabeledExample]], Tuple[List[LabeledExample], List[LabeledExample]]]:
        def split(examples: List[LabeledExample]) -> Tuple[List[LabeledExample], List[LabeledExample]]:
            examples_by_directory = group(examples, key=key_from_example)
            directories = examples_by_directory.keys()

            # split must be the same every time:
            random.seed(42)
            keys = set(random.sample(directories, int(training_share * len(directories))))

            training_examples = [example for example in examples if key_from_example(example) in keys]
            test_examples = [example for example in examples if key_from_example(example) not in keys]

            return training_examples, test_examples

        return split

def dispatch(method: Callable[[Any, Type[T], Mapping[str, Any], PipelineContext], Any]) -> Callable[[Any, Type[T], Mapping[str, Any], PipelineContext], Any]:
        dispatcher = singledispatch(method)
        provides = set()

        def wrapper(self: Any, type: Type[T], query: Mapping[str, Any], context: PipelineContext = None) -> Any:
            call = dispatcher.dispatch(type)
            try:
                return call(self, query, context=context)
            except TypeError:
                raise DataSource.unsupported(type)

        def register(type: Type[T]) -> Callable[[Any, Type[T], Mapping[str, Any], PipelineContext], Any]:
            provides.add(type)
            return dispatcher.register(type)

        wrapper.register = register
        wrapper._provides = provides
        update_wrapper(wrapper, method)
        return wrapper

def dispatch(method: Callable[[Any, Type[T], Any, PipelineContext], None]) -> Callable[[Any, Type[T], Any, PipelineContext], None]:
        dispatcher = singledispatch(method)
        accepts = set()

        def wrapper(self: Any, type: Type[T], items: Any, context: PipelineContext = None) -> None:
            call = dispatcher.dispatch(type)
            try:
                return call(self, items, context=context)
            except TypeError:
                raise DataSink.unsupported(type)

        def register(type: Type[T]) -> Callable[[Any, Type[T], Any, PipelineContext], None]:
            accepts.add(type)
            return dispatcher.register(type)

        wrapper.register = register
        wrapper._accepts = accepts
        update_wrapper(wrapper, method)
        return wrapper

def _transform(self, source_type: Type[S], target_type: Type[T]) -> Tuple[Callable[[S], T], int]:
        try:
            LOGGER.info("Searching type graph for shortest path from \"{source_type}\" to \"{target_type}\"".format(source_type=source_type.__name__, target_type=target_type.__name__))
            path = dijkstra_path(self._type_graph, source=source_type, target=target_type, weight="cost")
            LOGGER.info("Found a path from \"{source_type}\" to \"{target_type}\"".format(source_type=source_type.__name__, target_type=target_type.__name__))
        except (KeyError, NetworkXNoPath):
            raise NoConversionError("Pipeline can't convert \"{source_type}\" to \"{target_type}\"".format(source_type=source_type, target_type=target_type))

        LOGGER.info("Building transformer chain from \"{source_type}\" to \"{target_type}\"".format(source_type=source_type.__name__, target_type=target_type.__name__))
        chain = []
        cost = 0
        for source, target in _pairwise(path):
            transformer = self._type_graph.adj[source][target][_TRANSFORMER]
            chain.append((transformer, target))
            cost += transformer.cost
        LOGGER.info("Built transformer chain from \"{source_type}\" to \"{target_type}\"".format(source_type=source_type.__name__, target_type=target_type.__name__))

        if not chain:
            return _identity, 0

        return partial(_transform, transformer_chain=chain), cost

def _best_transform_to(self, target_type: Type[T], source_types: Iterable[Type]) -> Tuple[Callable[[T], Any], Type, int]:
        best = None
        best_cost = _MAX_TRANSFORM_COST
        from_type = None
        for source_type in source_types:
            try:
                transform, cost = self._transform(source_type, target_type)
                if cost < best_cost:
                    best = transform
                    best_cost = cost
                    from_type = source_type
            except NoConversionError:
                pass
        if best is None:
            raise NoConversionError("Pipeline can't convert from any of \"{source_types}\" to \"{target_type}\"".format(source_types=source_types, target_type=target_type))
        return best, from_type, best_cost

def with_default(self, value: Union[Any, Callable[[MutableMapping[str, Any]], Any]], supplies_type: Type = None) -> "QueryValidator":
        if self._current is None or self._current.child is not None:
            raise QueryValidatorStructureError("No key is selected! Try using \"can_have\" before \"with_default\".")

        if self._current.required:
            raise QueryValidatorStructureError("Can't assign a default value to a required key! Try using \"can_have\" instead of \"have\".")

        if supplies_type:
            expected_type = supplies_type
        else:
            expected_type = type(value)

        default_node = _DefaultValueNode(self._current.key, value, supplies_type)
        result = self.as_(expected_type)
        result._current.child.child = default_node
        return result

def dispatch(method: Callable[[Any, Type[T], F, PipelineContext], T]) -> Callable[[Any, Type[T], F, PipelineContext], T]:
        dispatcher = singledispatch(method)
        transforms = {}

        def wrapper(self: Any, target_type: Type[T], value: F, context: PipelineContext = None) -> T:
            call = dispatcher.dispatch(TypePair[value.__class__, target_type])
            try:
                return call(self, value, context=context)
            except TypeError:
                raise DataTransformer.unsupported(target_type, value)

        def register(from_type: Type[F], to_type: Type[T]) -> Callable[[Any, Type[T], F, PipelineContext], T]:
            try:
                target_types = transforms[from_type]
            except KeyError:
                target_types = set()
                transforms[from_type] = target_types
            target_types.add(to_type)

            return dispatcher.register(TypePair[from_type, to_type])

        wrapper.register = register
        wrapper._transforms = transforms
        update_wrapper(wrapper, method)
        return wrapper

def test_suggestPayout():
    MockedReportWrapper = NamedTuple('MockedReportWrapper', [('getReportBody', Callable),
                                                             ('getReportWeakness', Callable),
                                                             ('getVulnDomains', Callable)])
    MockedReportWrapperXSS = MockedReportWrapper(getReportBody=lambda: 'XSS',
                                                 getReportWeakness=lambda: 'XSS',
                                                 getVulnDomains=lambda: [])
    assert payout.suggestPayout(MockedReportWrapperXSS) == config.payoutDB['xss']['average']
    for vulnType in config.payoutDB:
        for domain in config.payoutDB[vulnType]:
            MockedReportWrapperVuln = MockedReportWrapper(getReportBody=lambda: vulnType,
                                                          getReportWeakness=lambda: vulnType,
                                                          getVulnDomains=lambda: [domain])
            assert payout.suggestPayout(MockedReportWrapperVuln) == config.payoutDB[vulnType][domain]
    MockedReportWrapperNone = MockedReportWrapper(getReportBody=lambda: '',
                                                  getReportWeakness=lambda: '',
                                                  getVulnDomains=lambda: [])
    assert payout.suggestPayout(MockedReportWrapperNone) is None

def retry(func: Callable[[], T]) -> T:
    """ Retry the function with 30 second timeouts until it works
        - I've observed the getFirefoxDriver() without this freeze once (out of hundreds of runs...) so adding this
          as a safety measure. """
    for i in range(10):
        if config.DEBUG and i > 0:
            print("Retry #%s" % str(i))

        def timeoutHandler(signum, frame):
            raise TimeoutException("Timeout!")
        signal.signal(signal.SIGALRM, timeoutHandler)
        signal.alarm(delayTime)
        try:
            t = func()
            signal.alarm(0)
            return t
        except TimeoutException:
            pass
    signal.alarm(0)
    raise TimeoutException("Retried 10 times... Failed!")

def __init__(self,
                 connection: Connection,
                 consumer: BrightsideConsumerConfiguration,
                 consumer_factory: Callable[[Connection, BrightsideConsumerConfiguration, logging.Logger], BrightsideConsumer],
                 command_processor_factory: Callable[[str], CommandProcessor],
                 mapper_func: Callable[[BrightsideMessage], Request]) -> None:
        """
        The configuration parameters for one consumer - can create one or more performers from this, each of which is
        a message pump reading froma queue
        :param connection: The connection to the broker
        :param consumer: The consumer we want to create (routing key, queue etc)
        :param consumer_factory: A factory to create a consumer to read from a broker, a given implementation i.e. arame
        the command processor factory creates a command procesoor configured for a pipeline
        :param mapper_func: Maps between messages on the queue and requests (commnands/events)
        """
        self._connection = connection
        self._consumer = consumer
        self._consumer_factory = consumer_factory
        self._command_processor_factory = command_processor_factory
        self._mapper_func = mapper_func

def register(self, request_class: Request, handler_factory: Callable[[], Handler]) -> None:
        """
        Register the handler for the command
        :param request_class: The command or event to dispatch. It must implement getKey()
        :param handler_factory: A factory method to create the handler to dispatch to
        :return:
        """
        key = request_class.__name__
        is_command = request_class.is_command()
        is_event = request_class.is_event()
        is_present = key in self._registry
        if is_command and is_present:
            raise ConfigurationException("A handler for this request has already been registered")
        elif is_event and is_present:
            self._registry[key].append(handler_factory)
        elif is_command or is_event:
            self._registry[key] = [handler_factory]

def _initializer_wrapper(init_function: Callable[..., None]) -> Type[Initializer]:
    class Init(Initializer):
        def __init__(self, **kwargs):
            self._init_function = init_function
            self._kwargs = kwargs
        def __call__(self, tensor: torch.autograd.Variable) -> None:
            self._init_function(tensor, **self._kwargs)
        def __repr__(self):
            return 'Init: %s, with params: %s' % (self._init_function, self._kwargs)
        @classmethod
        def from_params(cls, params: Params):
            return cls(**params.as_dict())
    return Init


# There are no classes to decorate, so we hack these into Registrable._registry

def _last_dimension_applicator(function_to_apply: Callable[[torch.Tensor, Optional[torch.Tensor]], torch.Tensor],
                               tensor: torch.Tensor,
                               mask: Optional[torch.Tensor] = None):
    """
    Takes a tensor with 3 or more dimensions and applies a function over the last dimension.  We
    assume the tensor has shape ``(batch_size, ..., sequence_length)`` and that the mask (if given)
    has shape ``(batch_size, sequence_length)``.  We first unsqueeze and expand the mask so that it
    has the same shape as the tensor, then flatten them both to be 2D, pass them through
    the function and put the tensor back in its original shape.
    """
    tensor_shape = tensor.size()
    reshaped_tensor = tensor.view(-1, tensor.size()[-1])
    if mask is not None:
        while mask.dim() < tensor.dim():
            mask = mask.unsqueeze(1)
        mask = mask.expand_as(tensor).contiguous().float()
        mask = mask.view(-1, mask.size()[-1])
    reshaped_result = function_to_apply(reshaped_tensor, mask)
    return reshaped_result.view(*tensor_shape)

def splat(f: Callable[..., A]) -> Callable[[Iterable], A]:
    """Convert a function taking multiple arguments into a function taking a single iterable argument.

    Args:
        f: Any function

    Returns:
        A function that accepts a single iterable argument. Each element of this iterable argument is passed as an
        argument to ``f``.

    Example:
        $ def f(a, b, c):
        $     return a + b + c
        $
        $ f(1, 2, 3)  # 6
        $ g = splat(f)
        $ g([1, 2, 3])  # 6
    """

    def splatted(args):
        return f(*args)

    return splatted

def unsplat(f: Callable[[Iterable], A]) -> Callable[..., A]:
    """Convert a function taking a single iterable argument into a function taking multiple arguments.

    Args:
        f: Any function taking a single iterable argument

    Returns:
        A function that accepts multiple arguments. Each argument of this function is passed as an element of an
        iterable to ``f``.

    Example:
        $ def f(a):
        $     return a[0] + a[1] + a[2]
        $
        $ f([1, 2, 3])  # 6
        $ g = unsplat(f)
        $ g(1, 2, 3)  # 6
    """

    def unsplatted(*args):
        return f(args)

    return unsplatted

def _split_sample(
            split: Callable[[object], bool], X: np.ndarray, y: np.ndarray
    ) -> Tuple[Tuple[np.ndarray, np.ndarray], Tuple[np.ndarray, np.ndarray]]:
        """
        Split X, y sample set in two with a split function
        :return: ((X_left, y_left), (X_right, y_right))
        """
        if split.type is 'numerical':
            left_indexes = X[:, split.attribute] < split.criteria
            right_indexes = ~left_indexes
        else:
            Z = (
                pd.Index(pd.unique(split.criteria))
                .get_indexer(X[:, split.attribute]))
            left_indexes = np.where(Z >= 0)[0]
            right_indexes = np.where(Z < 0)[0]

        left = X[left_indexes], y[left_indexes]
        right = X[right_indexes], y[right_indexes]

        return left, right

def _producer_multi_threads(queue_task, queue_product, worker_function):
    """
    ??????????????
    :type queue_task: multiprocessing.JoinableQueue
    :type queue_product: multiprocessing.JoinableQueue
    :type worker_function: Callable[[Any], Any]
    """
    while True:
        try:
            task = queue_task.get()
            if isinstance(task, _QueueEndSignal):  # ????
                # finally ?? task_done() ?break??????????
                break
            if isinstance(task, dict):
                result = worker_function(**task)
            elif isinstance(task, (tuple, list)):
                result = worker_function(*task)
            else:
                result = worker_function(task)

            queue_product.put((task, result))
        except:
            traceback.print_exc()
        finally:
            queue_task.task_done()

def __init__(self, loop,
                 target_host, target_port,
                 connected_callback: Callable[[BaseProtocol], None],
                 data_callback: Callable[[bytes], None],
                 user=None,
                 password=None):

        super(SOCKS5ConnectProtocol, self).__init__()
        self.loop = loop

        self.data_buffer = b''
        self.connected_callback = connected_callback
        self.data_callback = data_callback
        self.target_host = target_host
        self.target_port = target_port

        self.user = user
        self.password = password

        if (self.user and self.password):
            self.auth_method = constants.SOCKS5_METHOD_USERNAME_PASSWORD
        else:
            self.auth_method = constants.SOCKS5_METHOD_NO_AUTHENTICATION_REQUIRED

        self.state = STAGE_SOCKS5_METHOD_SELECT

def __init__(self, loop, transport,
                 tcp_connect_coroutine: Callable[[Socks5AddrHeader], Tuple[str, int]],
                 udp_connect_coroutine: Callable[[Socks5AddrHeader], Tuple[str, int]],
                 auth=constants.SOCKS5_METHOD_NO_AUTHENTICATION_REQUIRED,
                 username_passwords: Dict = None):

        self.loop = loop
        self.transport = transport

        self.tcp_connect_coroutine = tcp_connect_coroutine
        self.udp_connect_coroutine = udp_connect_coroutine

        self.auth = auth
        self.username_passwords = username_passwords

        self.state = constants.STAGE_SOCKS5_METHOD_SELECT

def getElementsBy(start_node: ParentNode,
                  cond: Callable[['Element'], bool]) -> NodeList:
    """Return list of child elements of start_node which matches ``cond``.

    ``cond`` must be a function which gets a single argument ``Element``,
    and returns boolean. If the node matches requested condition, ``cond``
    should return True.
    This searches all child elements recursively.

    :arg ParentNode start_node:
    :arg cond: Callable[[Element], bool]
    :rtype: NodeList[Element]
    """
    elements = []
    for child in start_node.children:
        if cond(child):
            elements.append(child)
        elements.extend(child.getElementsBy(cond))
    return NodeList(elements)

def throttle_request(throttling_rules: [RuleList, ThrottlingRule],
                     throttling_arguments_func: Callable=None,
                     throttling_options: ThrottlingOptions=None) -> Callable:
    """
    ????????? ??? view-???????, ?????????? ?????????.
    :param throttling_rules: ???? ????????? ThrottlingRule ??? ??????
    :param throttling_arguments_func: ???????, ?? ??????? ??????????? ????? ?????????? ??? ???????.
                                      ???? ?? ??????, ??????? ?-?? ?? ?????????.
    :param throttling_options: ????????? ThrottlingOptions, ???? ????? ????????? ?????????.
    """
    def decorator(func):
        func.throttling_rules = throttling_rules
        func.throttling_arguments_func = throttling_arguments_func
        func.throttling_options = throttling_options
        return func
    return decorator

def property(load_group_or_method: Union[Callable[[Any], Any], Any]) -> Union[property, Callable[[Callable[[Any], Any]], property]]:
        # Default behavior when no load group is provided. Set the load group to the method name.
        if isinstance(load_group_or_method, Callable) and not isinstance(load_group_or_method, type):
            method = load_group_or_method
            load_group = method.__name__
            method._Ghost__load_group = load_group

        # Set the load group as provided in the decorator
        else:
            load_group = load_group_or_method

        # The load_group and method variables need to be set correctly before this function defintion.
        def decorator(method: Callable) -> property:
            method._Ghost__load_group = load_group
            prop = Ghost.__property(method)
            prop._Ghost__load_group = load_group
            return prop

        if isinstance(load_group_or_method, Callable) and not isinstance(load_group_or_method, type):
            return decorator(method)
        else:
            return decorator

def __init__(self, producer: Callable[[], 'Actor'] = None,
                 spawner: Callable[[str, 'Props', pid.PID], pid.PID] = default_spawner,
                 mailbox_producer: Callable[
                     [invoker.AbstractInvoker,
                      'AbstractDispatcher'], mailbox.AbstractMailbox] = default_mailbox_producer,
                 dispatcher: 'AbstractDispatcher' = dispatcher.ThreadDispatcher(),
                 supervisor_strategy: supervision.AbstractSupervisorStrategy = None,
                 middleware: List[Callable[[context.AbstractContext], Task]] = None,
                 middleware_chain: Callable[[context.AbstractContext], Task] = None) -> None:
        self.__producer = producer
        self.__spawner = spawner
        self.__mailbox_producer = mailbox_producer
        self.__supervisor_strategy = supervisor_strategy
        self.__dispatcher = dispatcher
        self.__middleware = middleware
        self.__middleware_chain = middleware_chain

def __init__(
        self, *, cache_sketches: bool=True,
        source_encoding: str="utf-8",
            custom_escape_fns: Mapping[str, Callable[[Any], str]]={}) -> None:

        self._source_encoding = source_encoding

        escape_fns = escaping.builtin_escape_fns.copy()
        if custom_escape_fns:
            escape_fns.update(custom_escape_fns)
        self._escape_fns = types.MappingProxyType(escape_fns)

        self._stmt_classes = list(statements.builtin_stmt_classes)

        class OutputStmt(statements.BaseOutput):
            _filter_fn_names = list(self.escape_fns.keys())

        self._stmt_classes.append(OutputStmt)

        self._cache_sketches = cache_sketches

def __getitem__(
        self, name: Union[str, Tuple[str, bool]]) -> Callable[
            ..., Awaitable[str]]:
        if isinstance(name, tuple):
            block_name, defined_here = name

        else:
            block_name = name
            defined_here = False

        if block_name not in self._blocks.keys():
            raise KeyError(f"Unknown Block Name {block_name}.")

        SelectedBlockRuntime = self._blocks[block_name]

        async def wrapper() -> str:
            block_rt = SelectedBlockRuntime(
                self._skt_rt, _defined_here=defined_here)

            await block_rt._draw()

            return block_rt._block_result

        return wrapper

def write_file_update_progress(self, cmd: str, callback: Callable, filesize: int=0, blocksize: int=8192, rest: bool=None) -> str:
        self.ftps.voidcmd('TYPE I')  # type: ignore
        with self.ftps.transfercmd(cmd, rest) as conn:  # type: ignore
            self.current_download['filesize'] = filesize
            self.current_download['downloaded'] = 0
            self.current_download['filename'] = cmd.replace('RETR ', '')
            start = time.clock()
            while 1:
                data = conn.recv(blocksize)
                if not data:
                    break
                downloaded = len(data)
                self.current_download['downloaded'] += downloaded
                current = time.clock()
                if current > start:
                    self.current_download['speed'] = self.current_download['downloaded'] / ((current - start) * 1024)
                callback(data)
            self.current_download['filename'] = ''
            self.current_download['speed'] = 0
            self.current_download['filesize'] = 0
            # shutdown ssl layer
            if _SSLSocket is not None and isinstance(conn, _SSLSocket):
                conn.unwrap()
        return self.ftps.voidresp()  # type: ignore

def _get_provider_submodule_method(module_name: str, submodule_name: str, method_name: str) -> Optional[Callable]:
    sub_module = "{}.{}".format(module_name, submodule_name)
    try:
        importlib.import_module(module_name, package='__path__')
    except ImportError:
        return None
    if importlib.util.find_spec(sub_module):
        site = importlib.import_module(sub_module, package=module_name)
        if hasattr(site, method_name):
            obj = getattr(site, method_name)
            if inspect.isfunction(obj):
                return obj
    return None


# We should only create one ProviderContext over the program lifetime,
# to avoid having to search the file system every time it's created.
# This is why this should be outside Settings

def read(self, callback: Callable[[bytearray], Any]):
        try:
            for downloader in self._downloaders:

                # Wait until downloader is not in a downloaded/cancelled state.
                async with self._state_condition:
                    while downloader.state not in (DOWNLOADED, CANCELLED):
                        await self._state_condition.wait()
                    if downloader.state != DOWNLOADED:
                        self._debug('Downloader not in `DOWNLOADED` state, but in `{!s}`.'.format(downloader.state))
                        raise CancelledError()

                # Open file and send all its bytes it to back.
                await read_from_file_by_chunks(downloader.buffer_file_path, callback, self._chunk_size,
                                               lambda: self._state != CANCELLED, loop=self._loop)
        except Exception as exc:
            raise ReadError(exc)

def read_from_file_by_chunks(
        file_path: str,
        callback: Callable[[bytearray], None],
        chunk_size: int = DEFAULT_CHUNK_SIZE,
        condition: Callable[[], bool] = lambda: True,
        *,
        loop):
    chunk = bytearray(chunk_size)
    with open(file_path, 'rb') as f:
        while condition():
            r = await loop.run_in_executor(None, lambda: f.readinto(chunk))
            if not r:
                break
            if r < chunk_size:
                callback(memoryview(chunk)[:r].tobytes())
            else:
                callback(chunk)

def requires_bop(func) -> 'typing.Callable[[BaseOperator, BaseOperator], typing.Any]':
    """
    A decorator that marks a magic method as requiring another BaseOperator.

    :param func: The function to decorate.
    :return: A function that returns NotImplemented when the class required isn't specified.
    """

    @functools.wraps(func)
    def inner(self, other: 'BaseOperator'):
        if not isinstance(other, BaseOperator):
            return NotImplemented

        return func(self, other)

    return inner

def _get_delete_sql(self, emitter: typing.Callable[[], str], session: 'md_session.Session') \
            -> typing.Tuple[str, typing.Any]:
        """
        Gets the DELETE sql for this row.
        """
        if self._session is None:
            self._session = session

        query = io.StringIO()
        query.write("DELETE FROM {} ".format(self.__quoted_name__))
        # generate the where clauses
        wheres = []
        params = {}

        for col, value in zip(self.table.primary_key.columns,
                              md_inspection.get_pk(self, as_tuple=True)):
            name = emitter()
            params[name] = value
            wheres.append("{} = {}".format(col.quoted_fullname, session.bind.emit_param(name)))

        query.write("WHERE ({}) ".format(" AND ".join(wheres)))
        return query.getvalue(), params

    # value loading methods

def _compare(self, other: 'DebianVersion', method: Callable[[int], bool]) -> bool:
        if not isinstance(other, DebianVersion):
            return NotImplemented

        # special case: zero Epoch is the same as no Epoch
        if self.epoch is not None and other.epoch is not None and \
           int(self.epoch) != int(other.epoch) and int(self.epoch) != 0 and int(other.epoch) != 0:
            return method(int(self.epoch) - int(other.epoch))

        res = debian_versionpart_compare(split_version_parts(self.version, self.version_char_re),
                                         split_version_parts(other.version, self.version_char_re))
        if res == 0:
            return method(debian_versionpart_compare(split_version_parts(self.revision),
                                                     split_version_parts(other.revision)))
        else:
            return method(res)

def __init__(self, port: int, msgHandler: Callable,
                 name: str=None, basedirpath: str=None):
        if name and basedirpath:
            ha = getHaFromLocalEstate(name, basedirpath)
            if ha and ha[1] != port:
                port = ha[1]

        stackParams = {
            "name": name or randomString(8),
            "ha": HA("0.0.0.0", port),
            "main": True,
            "auth_mode": AuthMode.ALLOW_ANY.value,
            "mutable": "mutable",
            "messageTimeout": config.RAETMessageTimeout
        }
        if basedirpath:
            stackParams["basedirpath"] = basedirpath

            SimpleRStack.__init__(self, stackParams, self.tracedMsgHandler)

        self.msgHandler = msgHandler

def __init__(self, port: int, msgHandler: Callable,
                 name: str=None, basedirpath: str=None, seed=None,
                 onlyListener=False):
        stackParams = {
            "name": name or randomString(8),
            "ha": HA("0.0.0.0", port),
            "auth_mode": AuthMode.ALLOW_ANY.value
        }
        if basedirpath:
            stackParams["basedirpath"] = basedirpath

        seed = seed or randomSeed()
        SimpleZStack.__init__(self, stackParams, self.tracedMsgHandler,
                              seed=seed, onlyListener=onlyListener)

        self.msgHandler = msgHandler

def exposeFunction(self, name: str, puppeteerFunction: Callable
                             ) -> None:
        """Execute function on this page."""
        if self._pageBindings[name]:
            raise PageError(f'Failed to add page binding with name {name}: '
                            'window["{name}"] already exists!')
        self._pageBindings[name] = puppeteerFunction

        addPageBinding = '''
function addPageBinding(bindingName) {
  window[bindingName] = async(...args) => {
    const me = window[bindingName];
    let callbacks = me['callbacks'];
    if (!callbacks) {
      callbacks = new Map();
      me['callbacks'] = callbacks;
    }
    const seq = (me['lastSeq'] || 0) + 1;
    me['lastSeq'] = seq;
    const promise = new Promise(fulfill => callbacks.set(seq, fulfill));
    // eslint-disable-next-line no-console
    console.debug('driver:page-binding', JSON.stringify({name: bindingName, seq, args}));
    return promise;
  };
}
        '''  # noqa: E501
        expression = helper.evaluationString(addPageBinding, name)
        await self._client.send('Page.addScriptToEvaluateOnNewDocument',
                                {'source': expression})
        await self._client.send('Runtime.evaluate', {
            'expression': expression,
            'returnByValue': True
        })

def __init__(self, connection: Connection, ignoreHTTPSErrors: bool,
                 closeCallback: Callable[[], None]) -> None:
        """Make new browser object."""
        self._connection = connection
        self._ignoreHTTPSErrors = ignoreHTTPSErrors
        self._closeCallback = closeCallback

def tracebackwrapper(func, args, kwargs):
    # type: (Callable[..., Any], List[Any], Dict[Any, Any]) -> Any
    try:
        return func(*args, **kwargs)
    except Exception as e:
        e.traceback = traceback.format_exc()  # type: ignore
        raise

def iter_parallel_report(func,  # type: Callable[..., Any]
                         args_lists,  # type: Sequence[CallArgs]
                         ccmode=CC_PROCESSES):
    # type: (...) -> Iterator[Union[ExeResult, ExcInfo]]
    if ccmode == CC_OFF or len(args_lists) <= 1 or not multiprocessing:
        for args, kwargs in args_lists:
            yield func(*args, **kwargs)
        return

    processes = min(len(args_lists), multiprocessing.cpu_count())
    if ccmode == CC_THREADS:
        pool = multiprocessing.pool.ThreadPool(processes=processes)
    else:
        pool = multiprocessing.Pool(processes=processes, initializer=per_process_init)
    try:
        async_results = [pool.apply_async(func, args=args, kwds=kwargs)
                         for args, kwargs in args_lists]
        pool.close()
        while async_results:
            try:
                asyncres = async_results.pop(0)
                yield asyncres.get()
            except (KeyboardInterrupt, GeneratorExit):
                raise
            except Exception as e:
                t, v, tb = sys.exc_info()
                try:
                    # Report the textual traceback of the subprocess rather
                    # than this local exception which was triggered
                    # by the other side.
                    tb = e.traceback  # type: ignore
                except AttributeError:
                    pass
                yield ExcInfo((t, v, tb))
    except GeneratorExit:
        pool.terminate()
    except KeyboardInterrupt:
        pool.terminate()
        raise
    finally:
        pool.join()

def iter_parallel(func,        # type: Callable
                  args_lists,  # type: Sequence[CallArgs]
                  ccmode=CC_PROCESSES):
    # type: (...) -> Iterator[Any]
    if not args_lists:
        return
    if ccmode != CC_OFF:
        args_lists = [((func, args, kwargs), {}) for args, kwargs in args_lists]
        wrappedfunc = tracebackwrapper
    else:
        wrappedfunc = func

    for result in iter_parallel_report(wrappedfunc, args_lists, ccmode=ccmode):
        if ccmode == CC_OFF:
            yield result
        else:
            tbtext = None
            try:
                if isinstance(result, ExcInfo):
                    t, v, tb = result.exc_info
                    if not isinstance(tb, types.TracebackType):
                        tbtext = tb
                        tb = None
                    reraise(t, v, tb)
                else:
                    yield result
            except Exception:
                if tbtext is not None:
                    raise Exception(tbtext)
                else:
                    traceback.print_exc()
                    raise

# ----------------------------------------------------------------------
# The data types option and style.

def get_cache_value(key, func, cachevar):
    # type: (str, Callable[[str], Any], Dict[str, Any]) -> Any
    data = cachevar.get(key)
    if data is None:
        data = func(key)
        cachevar[key] = data
    return data

def create_activation(activation_name: str) -> Callable[[tf.Tensor], tf.Tensor]:
    """
    Create TensorFlow activation function with the given name.

    List of available activation functions is available in
    `TensorFlow docs <https://www.tensorflow.org/versions/r0.12/api_docs/python/nn/activation_functions_>`_.

    :param activation_name: activation function name
    :return: callable activation function
    """
    if activation_name == 'identity':
        return tf.identity
    return get_attribute(TF_ACTIVATIONS_MODULE, activation_name)

def __init__(self, op: Callable[[Memory.MEMORY_IMPL], List[bool]], args: List[Union[str, int]]):
        self.operator = op
        self.register = args[0]
        self.proc = None  # type: Processor
        self.reg_row = None  # type: Memory.MEMORY_IMPL

def __init__(self, op: Callable[[bool, bool], bool], args: List[Union[str, int]]):
        self.operator = op
        self.register = args[0]
        if isinstance(args[1], str):
            self.argument = args[1]
            self.literal = False
        else:
            self.argument = Memory.MEMORY_IMPL(Memory.REGISTER_WIDTH, False)
            self.argument.set_value(args[1])
            self.literal = True

def __init__(self, op: Callable[[List[bool], List[bool]], List[bool]], args: List[Union[str, int]]):
        self.operator = op
        self.register = args[0]
        if isinstance(args[1], str):
            self.argument = args[1]
            self.literal = False
        else:
            self.argument = Memory.MEMORY_IMPL(Memory.REGISTER_WIDTH, False)
            self.argument.set_value(args[1])
            self.literal = True
        self.proc = None  # type: Processor
        self.zero_bits = [False for _ in range(Memory.REGISTER_WIDTH)]  # type: List[bool]

def __init__(self, op: Callable[[int, int], int], args: List[Union[str, int]]):
        self.operator = op
        self.register = args[0]
        self.o_args = args
        self.args = []  # type: List[int]
        self.proc = None  # type: Processor

def __init__(self, op: Callable[[List[Union[str, int]]], None], args: List[Union[str, int]]):
        self.operator = op
        self.register = args[0]
        self.o_args = args
        self.proc = None  # type: Processor