我们从Python开源项目中,提取了以下32个代码示例,用于说明如何使用functools.singledispatch()。
def convert_yielded(yielded): """Convert a yielded object into a `.Future`. The default implementation accepts lists, dictionaries, and Futures. If the `~functools.singledispatch` library is available, this function may be extended to support additional types. For example:: @convert_yielded.register(asyncio.Future) def _(asyncio_future): return tornado.platform.asyncio.to_tornado_future(asyncio_future) .. versionadded:: 4.1 """ # Lists and dicts containing YieldPoints were handled earlier. if isinstance(yielded, (list, dict)): return multi(yielded) elif is_future(yielded): return yielded elif isawaitable(yielded): return _wrap_awaitable(yielded) else: raise BadYieldError("yielded unknown object %r" % (yielded,))
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 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 import_single_dispatch(): try: from functools import singledispatch except ImportError: singledispatch = None if not singledispatch: try: from singledispatch import singledispatch except ImportError: pass if not singledispatch: raise Exception( "It seems your python version does not include " "functools.singledispatch. Please install the 'singledispatch' " "package. More information here: " "https://pypi.python.org/pypi/singledispatch" ) return singledispatch
def convert_yielded(yielded): """Convert a yielded object into a `.Future`. The default implementation accepts lists, dictionaries, and Futures. If the `~functools.singledispatch` library is available, this function may be extended to support additional types. For example:: @convert_yielded.register(asyncio.Future) def _(asyncio_future): return tornado.platform.asyncio.to_tornado_future(asyncio_future) .. versionadded:: 4.1 """ # Lists and dicts containing YieldPoints were handled separately # via Multi(). if isinstance(yielded, (list, dict)): return multi_future(yielded) elif is_future(yielded): return yielded else: raise BadYieldError("yielded unknown object %r" % (yielded,))
def test_mro(self): @functools.singledispatch def g(obj): return "base" class A: pass class C(A): pass class B(A): pass class D(C, B): pass def g_A(a): return "A" def g_B(b): return "B" g.register(A, g_A) g.register(B, g_B) self.assertEqual(g(A()), "A") self.assertEqual(g(B()), "B") self.assertEqual(g(C()), "A") self.assertEqual(g(D()), "B")
def test_c_classes(self): @functools.singledispatch def g(obj): return "base" @g.register(decimal.DecimalException) def _(obj): return obj.args subn = decimal.Subnormal("Exponent < Emin") rnd = decimal.Rounded("Number got rounded") self.assertEqual(g(subn), ("Exponent < Emin",)) self.assertEqual(g(rnd), ("Number got rounded",)) @g.register(decimal.Subnormal) def _(obj): return "Too small to care." self.assertEqual(g(subn), "Too small to care.") self.assertEqual(g(rnd), ("Number got rounded",))
def singledispatch_pipeverb(func): """ Convenience decorator to convert a function to a singledispatch pipeline verb The function is converted to a :func:`singledispatch` function and then converted into an instance of class :class:`PipeVerb` via :func:`pipeverb`. This decorator is equivalent to ``verb = pipeverb(singledispatch(func))`` Please see :func:`pipeverb` for rules which a pipeline verb has to follow! :param func: the function which should be converted :type func: function :return: PipeVerb instance which can be used in rshift `>>` operations :rtype: PipeVerb :Example: >>> @singledispatch_pipeverb >>> def my_verb_impl(input, x=1, y=2): >>> raise NotImplementedError("my_verb is not implemented for data of type %s" % type(input)) >>> @my_verb_impl.register(pd.DataFrame) >>> def my_verb_impl_df(input, x=1, y=2): >>> # do something with input being a Dataframe >>> pass >>> # ensure that pd.DataFrame is useable as a pipe source >>> make_pipesource(pd.DataFrame) .. seealso:: :func:`pipeverb`, :func:`singledispatch` """ return pipeverb(singledispatch(func))
def method_dispatch(func: Callable[..., T]) -> Callable[..., T]: """ Single-dispatch class method decorator Works like functools.singledispatch for none-static class methods. """ dispatcher = functools.singledispatch(func) @functools.wraps(func) def wrapper(*args: Any, **_: Any) -> T: return dispatcher.dispatch(args[1].__class__)(*args, **_) # issue: https://github.com/python/mypy/issues/708 wrapper.register = dispatcher.register # type: ignore return wrapper
def test_simple_overloads(self): @functools.singledispatch def g(obj): return "base" def g_int(i): return "integer" g.register(int, g_int) self.assertEqual(g("str"), "base") self.assertEqual(g(1), "integer") self.assertEqual(g([1,2,3]), "base")
def test_register_decorator(self): @functools.singledispatch def g(obj): return "base" @g.register(int) def g_int(i): return "int %s" % (i,) self.assertEqual(g(""), "base") self.assertEqual(g(12), "int 12") self.assertIs(g.dispatch(int), g_int) self.assertIs(g.dispatch(object), g.dispatch(str)) # Note: in the assert above this is not g. # @singledispatch returns the wrapper.
def test_wrapping_attributes(self): @functools.singledispatch def g(obj): "Simple test" return "Test" self.assertEqual(g.__name__, "g") if sys.flags.optimize < 2: self.assertEqual(g.__doc__, "Simple test")
def has_context(incorrect_msg=None, exact_names=False, state=None): # call _has_context, since the built-in singledispatch can only use 1st pos arg return _has_context(state, incorrect_msg, exact_names)