Decorator for hook to specify minimum supported release
544618
def os_requires_version(ostack_release, pkg):
    
    def wrap(f):
        @wraps(f)
        def wrapped_f(*args):
            if os_release(pkg) < ostack_release:
                raise Exception("This hook is not supported on releases"
                                " before %s" % ostack_release)
            f(*args)
        return wrapped_f
    return wrap
105252
def hook(self, *hook_names):
        """"""
        def wrapper(decorated):
            for hook_name in hook_names:
                self.register(hook_name, decorated)
            else:
                self.register(decorated.__name__, decorated)
                if '_' in decorated.__name__:
                    self.register(
                        decorated.__name__.replace('_', '-'), decorated)
            return decorated
        return wrapper
196081
def minmax(low, high):
    """"""
    def decorator(function):
        """Decorate a function with args."""
        @functools.wraps(function)
        def wrapper(*args, **kwargs):
            """Wrap the function."""
            series = function(*args, **kwargs)
            lo_pass = low <= series
            hi_pass = series <= high

            return lo_pass & hi_pass

        return wrapper

    return decorator
802569
def create_release_hook(name, source_path):
    """"""
    from rez.plugin_managers import plugin_manager
    return plugin_manager.create_instance('release_hook',
                                          name,
                                          source_path=source_path)
807662
def hook_fn(self, module:nn.Module, input:Tensors, output:Tensors):
        ""
        if self.detach:
            input  = (o.detach() for o in input ) if is_listy(input ) else input.detach()
            output = (o.detach() for o in output) if is_listy(output) else output.detach()
        self.stored = self.hook_func(module, input, output)
216867
def release_api_class(self):
        """"""
        cls = current_app.config['GITHUB_RELEASE_CLASS']
        if isinstance(cls, string_types):
            cls = import_string(cls)
        assert issubclass(cls, GitHubRelease)
        return cls
1369743
def fn_minimum_argcount(callable):
    """"""
    fn = get_fn(callable)
    available_argcount = fn_available_argcount(callable)
    try:
        return available_argcount - len(fn.__defaults__)
    except TypeError:
        return available_argcount
388155
def _updater_wrapper(updater):
    """"""
    def updater_handle(key, lhs_handle, rhs_handle, _):
        """ ctypes function """
        lhs = _ndarray_cls(NDArrayHandle(lhs_handle))
        rhs = _ndarray_cls(NDArrayHandle(rhs_handle))
        updater(key, lhs, rhs)
    return updater_handle
318992
def register_hooks(func, hooks, obj):
    
    for name, args, kwargs in hooks:
        hook = getattr(obj, name)
        force_call = kwargs.pop("_force_call", False)
        if force_call or len(args) > 0 or len(kwargs) > 0:
            hook = hook(*args, **kwargs)
        hook(func)
686086
def handle(cls, value, context, **kwargs):
        
        try:
            hook_name, key = value.split("::")
        except ValueError:
            raise ValueError("Invalid value for hook_data: %s. Must be in "
                             "<hook_name>::<key> format." % value)

        return context.hook_data[hook_name][key]
829017
def _add_hookimpl(self, hookimpl):
        
        if hookimpl.hookwrapper:
            methods = self._wrappers
        else:
            methods = self._nonwrappers

        if hookimpl.trylast:
            methods.insert(0, hookimpl)
        elif hookimpl.tryfirst:
            methods.append(hookimpl)
        else:
            # find last non-tryfirst method
            i = len(methods) - 1
            while i >= 0 and methods[i].tryfirst:
                i -= 1
            methods.insert(i + 1, hookimpl)

        if "__multicall__" in hookimpl.argnames:
            warnings.warn(
                "Support for __multicall__ is now deprecated and will be"
                "removed in an upcoming release.",
                DeprecationWarning,
            )
            self.multicall = _legacymulticall
625926
def hook(self, hook_name):
        
        def wrapper(fn):
            self.register_hook(hook_name, fn)
            return fn
        return wrapper
944837
def post_exec_hook(hook):
    
    #post_setupnode hook
    module_name = '.'.join([env.project_package_name,'deploy'])
    funcs_run = []
    try:
        imported = import_module(module_name)
        func = vars(imported).get(hook)
        if func:
            func()
            funcs_run.append(func)
    except ImportError:
        return

   #run per app
    for app in env.INSTALLED_APPS:
        if app == 'woven': continue
        module_name = '.'.join([app,'deploy'])
        try:
            imported = import_module(module_name)
            func = vars(imported).get(hook)
            if func and func not in funcs_run:
                func()
                funcs_run.append(func)
        except ImportError:
            pass
    import woven.deploy
    func = vars(woven.deploy).get(hook)
    if func and func not in funcs_run: func()
23550
function(hook, rules) {
			if (cfg.AVAILABLE_HOOKS.indexOf(hook) < 0 && !this.ignore) {
				this.error('hook "' + hook + '" not support');
			}

			if (!hooks[hook]) {
				hooks[hook] = new Githook(rules);
			}
			return hooks[hook];
		}
1310930
def _hook(self, hook_name, doc_uri=None, **kwargs):
        """"""
        doc = self.workspace.get_document(doc_uri) if doc_uri else None
        hook_handlers = self.config.plugin_manager.subset_hook_caller(hook_name, self.config.disabled_plugins)
        return hook_handlers(config=self.config, workspace=self.workspace, document=doc, **kwargs)
648448
def os_workload_status(configs, required_interfaces, charm_func=None):
    
    def wrap(f):
        @wraps(f)
        def wrapped_f(*args, **kwargs):
            # Run the original function first
            f(*args, **kwargs)
            # Set workload status now that contexts have been
            # acted on
            set_os_workload_status(configs, required_interfaces, charm_func)
        return wrapped_f
    return wrap
935081
def extern(obj, timeout=200):
    
    global installed

    # Register it just once.
    if not installed: 
        install_hook(obj, timeout)
    installed = True