Decorators

illuminate.decorators.cli.adapt(command: str) -> Callable

Adapts Manager's static methods to accept cli arguments.

PARAMETER	DESCRIPTION
command	Command string TYPE: str

RETURNS	DESCRIPTION
Callable	Manager's static method wrapper

Source code in illuminate/decorators/cli.py

def adapt(command: str) -> Callable:
    """
    Adapts Manager's static methods to accept cli arguments.

    :param command: Command string
    :return: Manager's static method wrapper
    """

    def decorator(func: Callable) -> Callable:
        """
        Outer wrapper responsible for control flow.

        :param func: The function to be adapted
        :return: The wrapped function with adapted arguments
        """

        if command == "populate":

            @functools.wraps(func)
            def wrapper(
                fixtures: tuple[str], selector: str, url: str, *args, **kwargs
            ) -> None:
                """
                Adapts Manager's db_populate method to accept cli arguments.

                :param fixtures: Tuple of fixture files
                :param selector: Database name in settings.py module
                :param url: SQLAlchemy Database URL
                :return: None
                """

                try:
                    func(
                        fixtures,
                        selector,
                        url or Assistant._provide_db_url(selector),
                        *args,
                        **kwargs,
                    )
                except NoSuchModuleError:
                    raise BasicManagerException(
                        "Command populate can only be performed on SQL "
                        f"database, {selector} is not supported SQL database"
                    )

        elif command == "revision":

            @functools.wraps(func)
            def wrapper(
                path: str,
                revision: str,
                selector: str,
                url: str,
                *args,
                **kwargs,
            ) -> None:
                """
                Adapts Manager's db_revision method to accept cli arguments.

                :param path: Migration directory path
                :param revision: Parent revision
                :param selector: Database name in settings.py module
                :param url: SQLAlchemy Database URL
                :return: None
                """
                try:
                    config = Assistant.provide_alembic_config(
                        path, selector, url
                    )
                    func(config, revision, *args, **kwargs)
                except NoSuchModuleError:
                    raise BasicManagerException(
                        "Command revision can only be performed on SQL "
                        f"database, {selector} is not supported SQL database"
                    )

        elif command == "upgrade":

            @functools.wraps(func)
            def wrapper(
                path: str,
                revision: str,
                selector: str,
                url: str,
                *args,
                **kwargs,
            ) -> None:
                """
                Adapts Manager's db_upgrade method to accept cli arguments.

                :param path: Migration directory path
                :param revision: Parent revision
                :param selector: Database name in settings.py module
                :param url: SQLAlchemy Database URL
                :return: None
                """
                try:
                    config = Assistant.provide_alembic_config(
                        path, selector, url
                    )
                    func(config, revision, selector, *args, **kwargs)
                except NoSuchModuleError:
                    raise BasicManagerException(
                        "Command upgrade can only be performed on SQL "
                        f"database, {selector} is not supported SQL database"
                    )

        else:
            logger.warning(f"Decorated command {command} is not supported")

            @functools.wraps(func)
            def wrapper(*args, **kwargs) -> None:
                """
                Does nothing.

                :return: None
                """
                func(*args, **kwargs)

        return wrapper

    return decorator

illuminate.decorators.logging.show_info(func: Callable) -> Callable

Displays ETL process information.

PARAMETER	DESCRIPTION
func	Manager's public method TYPE: Callable

RETURNS	DESCRIPTION
Callable	Manager's public method wrapper

Source code in illuminate/decorators/logging.py

def show_info(func: Callable) -> Callable:
    """
    Displays ETL process information.

    :param func: Manager's public method
    :return: Manager's public method wrapper
    """

    @functools.wraps(func)
    def wrapper(self: Manager) -> None:
        """
        Displays ETL process information.

        :param self: Manager object
        :return: None
        """
        logger.info("Process started")
        start = default_timer()
        log_context(self)
        log_settings(self)
        func(self)
        end = default_timer() - start
        log_results(self)
        logger.opt(colors=True).info(
            f"Process finished in <yellow>{end:.2f}</yellow> seconds"
        )

    def log_context(self: Manager) -> None:
        """
        Displays ETL context information.

        :param self: Manager object
        :return: None
        """
        logger.opt(colors=True).info(
            f"Project files for project "
            f"<yellow>{self.name}</yellow> loaded into context"
        )
        logger.info(f"Adapters discovered {[i for i in self.adapters]}")
        logger.info(
            f"Models discovered {[locate(i) for i in self.settings.MODELS]}"
        )
        logger.info(f"Observers discovered {[i for i in self.observers]}")

    def log_results(self: Manager) -> None:
        """
        Displays ETL results information.

        :param self: Manager object
        :return: None
        """
        logger.success("Results gathered")
        logger.opt(colors=True).info(
            f"<yellow>Unsuccessful</yellow> observations: "
            f"<magenta>{len(self.not_observed)}</magenta>"
        )
        logger.debug(f"Unsuccessful attempts {self.not_observed}")
        logger.opt(colors=True).info(
            f"<yellow>Successful</yellow> observations: "
            f"<magenta>{len(self.observed) - len(self.not_observed)}</magenta>"
        )
        logger.opt(colors=True).info(
            f"Number of <yellow>exports</yellow>: "
            f"<magenta>{len(self.exported)}</magenta>"
        )

    def log_settings(self: Manager) -> None:
        """
        Displays ETL settings information.

        :param self: Manager object
        :return: None
        """
        settings_conn = self.settings.CONCURRENCY
        settings_db = self.settings.DB.copy()
        for db in settings_db:
            settings_db[db]["pass"] = "****"  # nosec
        settings_obs = self.settings.OBSERVATION_CONFIGURATION.copy()
        settings_obs["http"]["auth_password"] = "****"  # nosec
        logger.debug(f"Concurrency settings {settings_conn}")
        logger.debug(f"Database settings {settings_db}")
        logger.debug(f"Observation settings {settings_obs}")

    return wrapper

illuminate.decorators.logging.show_logo(func: Callable) -> Callable

Displays framework's logo.

PARAMETER	DESCRIPTION
func	Manager's public method TYPE: Callable

RETURNS	DESCRIPTION
Callable	Manager's public method wrapper

Source code in illuminate/decorators/logging.py

def show_logo(func: Callable) -> Callable:
    """
    Displays framework's logo.

    :param func: Manager's public method
    :return: Manager's public method wrapper
    """

    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> None:
        """
        Displays framework's logo.

        :return: None
        """
        logo = f"<fg {LOGO_COLOR}>{LOGO}</fg {LOGO_COLOR}>"
        logger.opt(colors=True).success(logo)
        func(*args, **kwargs)

    return wrapper

illuminate.decorators.logging.show_observer_catalogue(func: Callable) -> Callable

Displays Observers found in project files.

PARAMETER	DESCRIPTION
func	CLI's function TYPE: Callable

RETURNS	DESCRIPTION
Callable	CLI's function wrapper

Source code in illuminate/decorators/logging.py

def show_observer_catalogue(func: Callable) -> Callable:
    """
    Displays Observers found in project files.

    :param func: CLI's function
    :return: CLI's function wrapper
    """

    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> None:
        """
        Displays Observers found in project files.

        :return: None
        """
        context = func(*args, **kwargs)
        if not context["observers"]:
            logger.info("No observers found")
        outputs = []
        for observer in context["observers"]:
            _observer = observer()
            _io = _observer.initial_observations
            outputs.append(
                f"<yellow>{_observer.__class__}</yellow> -> "
                f"<cyan>{[(i.url, i._callback.__name__)for i in _io]}</cyan>"
            )
        for output in outputs:
            logger.opt(colors=True).info(output)

    return wrapper