fromfunctoolsimportlru_cache,singledispatchfromtypingimportAny,Callable,List,Tuple,Unionimportattrfrom.errorsimportStructureHandlerNotFoundError@attr.sclass_DispatchNotFound:"""A dummy object to help signify a dispatch not found."""pass
[docs]classMultiStrategyDispatch:""" MultiStrategyDispatch uses a combination of exact-match dispatch, singledispatch, and FunctionDispatch. """__slots__=("_direct_dispatch","_function_dispatch","_single_dispatch","_generators","dispatch",)def__init__(self,fallback_func):self._direct_dispatch={}self._function_dispatch=FunctionDispatch()self._function_dispatch.register(lambda_:True,fallback_func)self._single_dispatch=singledispatch(_DispatchNotFound)self.dispatch=lru_cache(maxsize=None)(self._dispatch)def_dispatch(self,cl):try:dispatch=self._single_dispatch.dispatch(cl)ifdispatchisnot_DispatchNotFound:returndispatchexceptException:passdirect_dispatch=self._direct_dispatch.get(cl)ifdirect_dispatchisnotNone:returndirect_dispatchreturnself._function_dispatch.dispatch(cl)
[docs]defregister_cls_list(self,cls_and_handler,direct:bool=False)->None:"""Register a class to direct or singledispatch."""forcls,handlerincls_and_handler:ifdirect:self._direct_dispatch[cls]=handlerelse:self._single_dispatch.register(cls,handler)self.clear_direct()self.dispatch.cache_clear()
[docs]defregister_func_list(self,pred_and_handler:List[Union[Tuple[Callable[[Any],bool],Any],Tuple[Callable[[Any],bool],Any,bool],]],):""" Register a predicate function to determine if the handle should be used for the type. """fortupinpred_and_handler:iflen(tup)==2:func,handler=tupself._function_dispatch.register(func,handler)else:func,handler,is_gen=tupself._function_dispatch.register(func,handler,is_generator=is_gen)self.clear_direct()self.dispatch.cache_clear()
[docs]defclear_direct(self):"""Clear the direct dispatch."""self._direct_dispatch.clear()
[docs]defclear_cache(self):"""Clear all caches."""self._direct_dispatch.clear()self.dispatch.cache_clear()
[docs]@attr.s(slots=True)classFunctionDispatch:""" FunctionDispatch is similar to functools.singledispatch, but instead dispatches based on functions that take the type of the first argument in the method, and return True or False. objects that help determine dispatch should be instantiated objects. """_handler_pairs:list=attr.ib(factory=list)
[docs]defdispatch(self,typ):""" returns the appropriate handler, for the object passed. """forcan_handle,handler,is_generatorinself._handler_pairs:# can handle could raise an exception here# such as issubclass being called on an instance.# it's easier to just ignore that case.try:ch=can_handle(typ)exceptException:continueifch:ifis_generator:returnhandler(typ)else:returnhandlerraiseStructureHandlerNotFoundError(f"unable to find handler for {typ}",type_=typ)