def _run_rta_in_process()

in cortado/rtas/_cli.py [0:0]

• 23 lines of code
• 5 McCabe index (conditional complexity)

def _run_rta_in_process(rta_name: str) -> Exception | None:
    configure_logging()
    log = logging.getLogger(__name__)

    log.debug(f"Loading {rta_name}")

    try:
        # NOTE: we're assuming here that the RTA will be registered in the module
        # named as RTA. This might not be the case in the future.
        load_module(rta_name)
        registry = get_registry()

        rta = registry.get(rta_name)
        if not rta:
            return ValueError(f"Can't find RTA with name `{rta_name}`")

    except Exception as e:
        log.error(f"Can't load RTA `{rta_name}`", exc_info=True)
        return e

    if not isinstance(rta, CodeRta):
        log.warning(f"RTA `{rta_name}` is not executable")
        return ValueError(f"RTA `{rta_name}` is not executable")

    log.debug(f"Running {rta_name}")
    try:
        rta.code_func()
    except Exception as e:
        log.error(f"RTA `{rta_name}` failed during execution", exc_info=True)
        return e

    return None