xpost / util / util.py

import logging
import sys
import os
from typing import Any, Callable

import env

shutdown_hook: list[Callable[[], None]] = []

logging.basicConfig(stream=sys.stderr, level=logging.DEBUG if env.DEV else logging.INFO)
LOGGER = logging.getLogger("XPost")

def normalize_service_url(url: str) -> str:
    if not url.startswith("https://") and not url.startswith("http://"):
        raise ValueError(f"Invalid service url {url}! Only http/https are supported.")

    return url[:-1] if url.endswith('/') else url

def read_env(data: dict[str, Any]) -> None:
    keys = list(data.keys())
    for key in keys:
        val = data[key]
        match val:
            case str():
                if val.startswith('env:'):
                    envval = os.environ.get(val[4:])
                    if envval is None:
                        del data[key]
                    else:
                        data[key] = envval
            case dict():
                read_env(val)
            case _:
                pass