How to use hook_spec method in localstack

Best Python code snippet using localstack_python

opts.py

Source:opts.py

1from copy import copy2from functools import partial, reduce3from typing import Any, Union4import panel as pn5import pandas as pd6import param7import proplot as pplt8from matplotlib.colors import Colormap, Normalize9from lumflux.support import apply_cmap10# todo baseclass widget generating thingy11class OptsBase(param.Parameterized):12    _type = None13    updated = param.Event()14    def __init__(self, **params):15        super().__init__(**params)16        self._excluded_from_opts = ["name"]  # todo remove this and opts property17        self.widgets = self.generate_widgets()18    @property19    def panel(self) -> pn.Column:20        return pn.Column(*self.widgets.values())21    @property22    def opts(self) -> dict:23        opts = {24            name: self.param[name]25            for name in self.param26            if name not in self._excluded_from_opts27        }28        return opts29    def generate_widgets(self, **kwargs) -> dict:30        """returns a dict with keys parameter names and values default mapped widgets"""31        # todo base class?32        names = [33            p34            for p in self.param35            if self.param[p].precedence is None or self.param[p].precedence > 136        ]37        widgets = pn.Param(38            self.param, show_name=False, show_labels=True, widgets=kwargs39        )40        return {k: v for k, v in zip(names[1:], widgets)}41class GenericOpts(OptsBase):42    _type = "generic"43    hooks = param.List()44    def __init__(self, **params):45        self.kwargs = {k: v for k, v in params.items() if k not in self.param}46        super().__init__(**{k: v for k, v in params.items() if k in self.param})47    def hooks_factory(self):48        def hook(hooks, plot, element):49            for hook_spec in hooks:50                handle = plot.handles[hook_spec["handle"]]51                rsetattr(handle, hook_spec["attr"], hook_spec["value"])52        f = partial(hook, self.hooks)53        return f54    @property55    def opts(self) -> dict:56        """Returns final opts dict which is passed to holoviews.57        Returns:58        """59        return {"hooks": [self.hooks_factory()], **self._parse_kwargs(self.kwargs)}60    @staticmethod61    def _parse_kwargs(kwargs):62        out = {}63        for k, v in kwargs.items():64            if k in ["xlim", "ylim"] and isinstance(v, list):65                out[k] = tuple(v)66            elif k in ["padding"] and isinstance(v, list):67                out[k] = to_tuple(v)68            else:69                out[k] = v70        return out71# https://stackoverflow.com/questions/27049998/convert-a-mixed-nested-list-to-a-nested-tuple/27050037#2705003772def to_tuple(lst):73    return tuple(to_tuple(i) if isinstance(i, list) else i for i in lst)74# https://stackoverflow.com/questions/31174295/getattr-and-setattr-on-nested-subobjects-chained-properties75def rsetattr(obj: Any, attr: str, val: Any):76    """Recursive set attr77    Args:78        obj:79        attr:80        val:81    Returns:82    """83    pre, _, post = attr.rpartition(".")84    return setattr(rgetattr(obj, pre) if pre else obj, post, val)85def rgetattr(obj, attr, *args):86    def _getattr(obj, attr):87        return getattr(obj, attr, *args)88    return reduce(_getattr, [obj] + attr.split("."))89class HooksOpts(OptsBase):90    _type = "hooks"91    hooks = param.List()92    #TODO: this code is duplicate with GenericOpts93    def hooks_factory(self):94        def hook(hooks, plot, element):95            for hook_spec in hooks:96                handle = plot.handles[hook_spec["handle"]]97                rsetattr(handle, hook_spec["attr"], hook_spec["value"])98        f = partial(hook, self.hooks)99        return f100    @property101    def opts(self):102        opts = {"hooks": [self.hooks_factory()]}103        return opts104class CmapOpts(OptsBase):105    _type = "cmap"106    cmap = param.ClassSelector(default=None, class_=Colormap)107    norm = param.ClassSelector(default=None, class_=Normalize)108    # the stored norm here is the unscaled one109    # scale factor is applied to clim and norm_scaled110    clim = param.Tuple((0.0, 1.0), length=2)111    sclf = param.Number(1.0, doc="scaling factor to apply")  # curent: 1e+3112    field = param.String(doc="field on which cmap works")113    def __init__(self, rename=True, invert=False, **params):114        # todo from_spec constructor method for this kind of logic115        cmap = params.pop("cmap", None)116        cmap = pplt.Colormap(cmap) if cmap else cmap117        params["cmap"] = cmap118        super().__init__(**params)119        self._excluded_from_opts += [120            "norm",121            "sclf",122        ]  # perhaps use leading underscore to exclude?123        self.norm = self.norm or pplt.Norm("linear", 0.0, 1.0)124        self._cmap = self.cmap or pplt.Colormap("viridis")  # unreversed cmap125        if rename:126            cmap.name = self.field + "_default"127        if invert:128            cmap = cmap.reversed()129        self.cmap = cmap130    @property131    def opts(self):132        names = ["cmap", "clim"]133        opts = {name: self.param[name] for name in names}134        return opts135    @property136    def norm_scaled(self):137        norm = copy(self.norm)138        norm.vmin *= self.sclf139        norm.vmax *= self.sclf140        return norm141    @norm_scaled.setter142    def norm_scaled(self, norm):143        _norm = copy(norm)144        _norm.vmin /= self.sclf145        _norm.vmax /= self.sclf146        self.norm = _norm147    @param.depends("norm", watch=True)148    def _norm_updated(self):149        self.clim = self.norm.vmin * self.sclf, self.norm.vmax * self.sclf150        # todo invert bool?151        # self.clim = self.norm.vmax*self.sclf, self.norm.vmin*self.sclf,152    def apply(self, data: Union[pd.Series, pd.DataFrame]) -> Union[pd.Series, pd.DataFrame]:153        """apply cmap / norm to data (pd series or df)"""154        # norm = copy(self.norm)155        # norm.vmin *= self.sclf156        # norm.vmax *= self.sclf157        return apply_cmap(data, self.cmap, self.norm)158    @param.depends("norm", "cmap", watch=True)159    def update(self):...

hooks.py

Source:hooks.py

...14    def wrapper(fn):15        fn.hook_priority = priority16        return plugin(namespace=namespace, **kwargs)(fn)17    return wrapper18def hook_spec(namespace: str):19    """20    Creates a new hook decorator bound to a namespace.21    on_infra_start = hook_spec("localstack.hooks.on_infra_start")22    @on_infra_start()23    def foo():24        pass25    # run all hooks in order26    on_infra_start.run()27    """28    fn = functools.partial(hook, namespace=namespace)29    # attach hook manager and run method to decorator for convenience calls30    fn.manager = HookManager(namespace)31    fn.run = fn.manager.run_in_order32    return fn33class HookManager(PluginManager):34    def load_all_sorted(self, propagate_exceptions=False):35        """36        Loads all hook plugins and sorts them by their hook_priority attribute.37        """38        plugins = self.load_all(propagate_exceptions)39        # the hook_priority attribute is part of the function wrapped in the FunctionPlugin40        plugins.sort(41            key=lambda _fn_plugin: getattr(_fn_plugin.fn, "hook_priority", 0), reverse=True42        )43        return plugins44    def run_in_order(self, *args, **kwargs):45        """46        Loads and runs all plugins in order them with the given arguments.47        """48        for fn_plugin in self.load_all_sorted():49            fn_plugin(*args, **kwargs)50    def __str__(self):51        return "HookManager(%s)" % self.namespace52    def __repr__(self):53        return self.__str__()54configure_localstack_container = hook_spec(HOOKS_CONFIGURE_LOCALSTACK_CONTAINER)55"""Hooks to configure the LocalStack container before it starts. Executed on the host when invoking the CLI."""56install = hook_spec(HOOKS_INSTALL)57"""Additional programmatic installers."""58prepare_host = hook_spec(HOOKS_PREPARE_HOST)59"""Hooks to prepare the host that's starting LocalStack. Executed on the host when invoking the CLI."""60on_infra_start = hook_spec(HOOKS_ON_INFRA_START)61"""Hooks that are executed right before starting the LocalStack infrastructure."""62on_infra_ready = hook_spec(HOOKS_ON_INFRA_READY)63"""Hooks that are execute after all startup hooks have been executed, and the LocalStack infrastructure has become...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.