def dispatch()

in src/dispatch/plugins/dispatch_slack/decorators.py [0:0]

• 22 lines of code
• 10 McCabe index (conditional complexity)

    def dispatch(self, *args, **kwargs):
        """Runs all registered functions."""
        for f in self.registered_funcs:
            # only inject the args the function cares about
            func_args = inspect.getfullargspec(inspect.unwrap(f["func"])).args
            injected_args = (kwargs[a] for a in func_args)

            if subject := f["subject"]:
                if subject_meta := kwargs.get("context", {}).get("subject"):
                    if subject_meta:
                        if subject != subject_meta.type:
                            log.debug(
                                f"Skipping dispatch function due to subject exclusion. ({f['name']})"
                            )
                            continue

            if exclude := f["exclude"]:
                subtype: str = kwargs.get("body", {}).get("event", {}).get("subtype", "")
                if subtype in exclude.get("subtype", []):
                    log.debug(f"Skipping dispatched function due to event exclusion. ({f['name']})")
                    continue

            try:
                f["func"](*injected_args)
            except Exception as e:
                log.exception(e)
                log.debug(f"Failed to run dispatched function {f['name']}. Reason: ({e})")