我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用collections.Callable()。
def register_archive_format(name, function, extra_args=None, description=''): """Registers an archive format. name is the name of the format. function is the callable that will be used to create archives. If provided, extra_args is a sequence of (name, value) tuples that will be passed as arguments to the callable. description can be provided to describe the format, and will be returned by the get_archive_formats() function. """ if extra_args is None: extra_args = [] if not isinstance(function, collections.Callable): raise TypeError('The %s object is not callable' % function) if not isinstance(extra_args, (tuple, list)): raise TypeError('extra_args needs to be a sequence') for element in extra_args: if not isinstance(element, (tuple, list)) or len(element) !=2: raise TypeError('extra_args elements are : (arg_name, value)') _ARCHIVE_FORMATS[name] = (function, extra_args, description)
def _check_unpack_options(extensions, function, extra_args): """Checks what gets registered as an unpacker.""" # first make sure no other unpacker is registered for this extension existing_extensions = {} for name, info in _UNPACK_FORMATS.items(): for ext in info[0]: existing_extensions[ext] = name for extension in extensions: if extension in existing_extensions: msg = '%s is already registered for "%s"' raise RegistryError(msg % (extension, existing_extensions[extension])) if not isinstance(function, collections.Callable): raise TypeError('The registered function must be a callable')
def register_archive_format(name, function, extra_args=None, description=''): """Registers an archive format. name is the name of the format. function is the callable that will be used to create archives. If provided, extra_args is a sequence of (name, value) tuples that will be passed as arguments to the callable. description can be provided to describe the format, and will be returned by the get_archive_formats() function. """ if extra_args is None: extra_args = [] if not isinstance(function, collections.Callable): raise TypeError('The %s object is not callable' % function) if not isinstance(extra_args, (tuple, list)): raise TypeError('extra_args needs to be a sequence') for element in extra_args: if not isinstance(element, (tuple, list)) or len(element) !=2 : raise TypeError('extra_args elements are : (arg_name, value)') _ARCHIVE_FORMATS[name] = (function, extra_args, description)
def filter(self, filter): """ Filters this image using the given filter. For a list of available filters, see the :py:mod:`~PIL.ImageFilter` module. :param filter: Filter kernel. :returns: An :py:class:`~PIL.Image.Image` object. """ self.load() if isinstance(filter, collections.Callable): filter = filter() if not hasattr(filter, "filter"): raise TypeError("filter argument should be ImageFilter.Filter " + "instance or class") if self.im.bands == 1: return self._new(filter.filter(self.im)) # fix to handle multiband images since _imaging doesn't ims = [] for c in range(self.im.bands): ims.append(self._new(filter.filter(self.im.getband(c)))) return merge(self.mode, ims)
def __call__(self, func_or_msg=None, **kwargs): """ If the first argument is a function, return a decorator which runs the wrapped function inside Timer's context manager. Otherwise, treat the first argument as a 'msg' string and return an updated Timer instance, referencing the same logger. A 'level' keyword can also be passed to override self.level. """ if isinstance(func_or_msg, collections.Callable): func = func_or_msg # use the function name when no explicit 'msg' is provided if not self.msg: self.msg = "run '%s'" % func.__name__ @wraps(func) def wrapper(*args, **kwds): with self: return func(*args, **kwds) return wrapper else: msg = func_or_msg or kwargs.get("msg") level = kwargs.get("level", self.level) return self.__class__(self.logger, msg, level)
def gazellestr(exp): ''' Convert a Python object back into a Gazelle-readable string. ''' # Bools if isinstance(exp, bool): if exp: return '#t' else: return '#f' # Procedures elif isinstance(exp, collections.Callable): try: return '(lambda (' + ' '.join([str(x) for x in exp.params]) + \ ') (' + ' '.join([gazellestr(x) for x in exp.body]) + '))' except AttributeError: return exp # Lists elif isinstance(exp, list): return '(' + ' '.join(map(gazellestr, exp)) + ')' # Everything else return str(exp)
def compile_node(self, field): """Compile node into a cached function that performs the match. Returns: Callable: taking the object to match. """ # can embed other Q objects if isinstance(field, type(self)): return field # convert Django Q objects in-place. elif isinstance(field, _Q_): field.__class__ = type(self) return field # or it's a key, value pair. lhs, rhs = field lhs, opcode = self.prepare_statement(lhs, rhs) # this creates the new matching function to be added to the stack. return self.compile_op(lhs, rhs, opcode)
def addApplyFunc(self, func, outColumn, inColumn='', *args, **kwargs): """Add function to be applied to dataframe""" # check inputs if not isinstance(func, collections.Callable): self.log().critical('specified function object is not callable') raise AssertionError('functions in ApplyFuncToDf link must be callable objects') if not isinstance(outColumn, str) or not isinstance(inColumn, str): self.log().critical('types of specified column names are "%s" and "%s"', type(outColumn).__name__, type(inColumn).__name__) raise TypeError('column names in ApplyFuncToDf must be strings') if not outColumn: self.log().critical('no output column specified') raise RuntimeError('an output column must be specified to apply function in ApplyFuncToDf') # add function if inColumn == '': self.apply_funcs.append({'func': func, 'colout': outColumn, 'args': args, 'kwargs': kwargs}) else: self.apply_funcs.append({'colin': inColumn, 'func': func, 'colout': outColumn, 'args': args, 'kwargs': kwargs})
def __new__(meta, classname, bases, class_dict): new_class_dict = {} for attr_name, attr_value in list(class_dict.items()): if isinstance(attr_value, collections.Callable) and hasattr(attr_value, 'param_names'): # print("Processing attr_name = %r; attr_value = %r" % ( # attr_name, attr_value)) method = attr_value param_names = attr_value.param_names data = attr_value.data func_name_format = attr_value.func_name_format meta.process_method( classname, method, param_names, data, new_class_dict, func_name_format) else: new_class_dict[attr_name] = attr_value return type.__new__(meta, classname, bases, new_class_dict)
def _metaclass_check_getitem_type_decl(cls, type_decl): # Callable[[T, ...], T] if not check_getitem_tuple(type_decl, 2): return False # return type. if nontype_object(type_decl[1]): return False # types of parameters. if isinstance(type_decl[0], abc.Iterable): # [T, ...] for T in type_decl[0]: if nontype_object(T): return False else: return True else: # special case, Ellipsis. return type_decl[0] is Ellipsis
def _class___new__(cls, instance): # 1. not Callable. if not isinstance(instance, cls.main_cls): raise MagicTypeError( 'instance should be a Callable.', instance=instance, ) # 2. with no specification. if not cls.partial_cls: raise MagicSyntaxError( 'Callable should be specified to wrap the callable.', ) parameters_types, return_type = cls.partial_cls if parameters_types is Ellipsis: parameters_types = [parameters_types] return function_constraints( *parameters_types, # tailing coma is invalid for version < 3.5. return_type=return_type )(instance)
def __init__(self, callback, interval=1): """ :param callback: a function with one arg to execute every so often :type callback: function :param interval: number of seconds at most between the updates :type interval: int """ if isinstance(callback, collections.Callable): # callback is really callable self.update = self.update_real self.callback = callback self.interval = interval else: self.update = self.update_empty self.count = 0 self.oldval = 0 self.oldtime = 0 self.text = "" self.total = 1