Source code for atomiq.arguments.json

from atomiq.arguments import ArgumentProvider
from atomiq.helper import file_from_url
import json

import logging

logging.basicConfig()
logger = logging.getLogger(__name__)




class JSONArgumentProvider(ArgumentProvider):
    def __init__(self, url: str, use_class_defaults: bool = True):
        """
        An argument provider that grabs arguments from a JSON stored at a URL.

        This can either be a local file by starting the URL with (file://...) or a web resource
        that serves a JSON like for example a CouchDB URL

        Args:
            url: URL of the CouchDB document
        """
        ArgumentProvider.__init__(self)

        self.url = url
        self.use_class_defaults = use_class_defaults

    def _build_arguments(self, target_class: type, arguments: dict = {}) -> dict:

        # if the class defines arguments, take them as defaults
        if self.use_class_defaults and hasattr(target_class, "arguments"):
            arguments.update(target_class.arguments)

        try:
            d = json.loads(file_from_url(self.url).read())
        except Exception as e:
            logger.warning(f"Could not load arguments from JSON. Error: {e}")
            d = {}

        if target_class.__name__ in d.keys():
            arguments.update(d[target_class.__name__])

        return arguments