checktheroads-hyperglass/hyperglass/api/events.py

"""API Events."""
# Third Party Imports
from starlette.exceptions import HTTPException

# Project Imports
from hyperglass.configuration import REDIS_CONFIG
from hyperglass.configuration import URL_DEV
from hyperglass.configuration import URL_PROD
from hyperglass.configuration import frontend_params
from hyperglass.configuration import params
from hyperglass.exceptions import HyperglassError
from hyperglass.util import build_frontend
from hyperglass.util import check_python
from hyperglass.util import check_redis
from hyperglass.util import clear_redis_cache
from hyperglass.util import log


async def check_python_version():
    """Ensure Python version meets minimum requirement.

    Raises:
        HyperglassError: Raised if Python version is invalid.
    """
    try:
        python_version = check_python()
        log.info(f"Python {python_version} detected")
    except RuntimeError as r:
        raise HyperglassError(str(r), level="danger") from None


async def check_redis_instance():
    """Ensure Redis is running before starting server.

    Raises:
        HyperglassError: Raised if Redis is not running.

    Returns:
        {bool} -- True if Redis is running.
    """
    try:
        await check_redis(db=params.cache.database, config=REDIS_CONFIG)
    except RuntimeError as e:
        raise HyperglassError(str(e), level="danger") from None

    log.debug(f"Redis is running at: {REDIS_CONFIG['host']}:{REDIS_CONFIG['port']}")
    return True


async def build_ui():
    """Perform a UI build prior to starting the application.

    Raises:
        HTTPException: Raised if any build errors occur.

    Returns:
        {bool} -- True if successful.
    """
    try:
        await build_frontend(
            dev_mode=params.developer_mode,
            dev_url=URL_DEV,
            prod_url=URL_PROD,
            params=frontend_params,
        )
    except RuntimeError as e:
        raise HTTPException(detail=str(e), status_code=500)
    return True


async def clear_cache():
    """Clear the Redis cache on shutdown."""
    try:
        await clear_redis_cache(db=params.cache.database, config=REDIS_CONFIG)
    except RuntimeError as e:
        log.error(str(e))
        pass


on_startup = [check_python_version, check_redis_instance, build_ui]
on_shutdown = [clear_cache]