blitz_api/app/api/warmup.py

import asyncio
from typing import List

from fastapi import HTTPException, status
from loguru import logger

from app.apps.service import get_app_status
from app.bitcoind.service import get_btc_info
from app.lightning.service import (
    get_fee_revenue,
    get_ln_info,
    get_ln_info_lite,
    get_wallet_balance,
)
from app.system.service import get_hardware_info, get_system_info


@logger.catch(exclude=(HTTPException,))
async def get_bitcoin_client_warmup_data() -> List:
    """Get the reduced data set needed when the lightning client is not yet ready."""
    res = await asyncio.gather(
        *[
            get_btc_info(),
            get_hardware_info(),
        ]
    )
    return [*res]


@logger.catch(exclude=(HTTPException,))
async def get_full_client_warmup_data() -> List:
    """Get the full data set needed when the lightning client is not yet ready."""

    res = await asyncio.gather(
        *[
            get_system_info(),
            get_btc_info(),
            get_ln_info(),
            get_ln_info_lite(),
            get_fee_revenue(),
            get_wallet_balance(),
            get_app_status(),
            get_hardware_info(),
        ],
        return_exceptions=True,
    )

    for i, r in enumerate(res):
        if isinstance(r, HTTPException):
            if r.status_code == status.HTTP_501_NOT_IMPLEMENTED:
                logger.trace(f"Not implemented Error in warmup data {i}: {r.detail}")
            # TODO: find a better way to handle this, client receives an error but
            # disguised as a valid response. For example:
            # event: installed_app_status
            # data: {
            #   "status_code": 501,
            #   "detail": "Not available in native python mode.",
            #   "headers": null
            # }
            res[i] = r
        elif isinstance(r, Exception):
            logger.error(f"Error in warmup data {i}: {r}")
            res[i] = HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)

    return [*res]


@logger.catch(exclude=(HTTPException,))
async def get_full_client_warmup_data_bitcoinonly() -> List:
    """Get the full data set needed without Lightning available"""

    res = await asyncio.gather(
        *[
            get_system_info(),
            get_btc_info(),
            get_app_status(),
            get_hardware_info(),
        ]
    )
    return [*res]