application_json

Strategy class for application/json.

JSONConfig

Bases: AttrDict

JSON parse-specific Configuration Data Model.

Source code in oteapi/strategies/parse/application_json.py

class JSONConfig(AttrDict):
    """JSON parse-specific Configuration Data Model."""

    datacache_config: Optional[DataCacheConfig] = Field(
        None,
        description="Configurations for the data cache for storing the downloaded file content.",
    )

JSONDataParseStrategy

Parse strategy for JSON.

Registers strategies:

• ("mediaType", "application/json")

Source code in oteapi/strategies/parse/application_json.py

@dataclass
class JSONDataParseStrategy:
    """Parse strategy for JSON.

    **Registers strategies**:

    - `("mediaType", "application/json")`

    """

    parse_config: JSONResourceConfig

    def initialize(self, session: "Optional[Dict[str, Any]]" = None) -> SessionUpdate:
        """Initialize."""
        return SessionUpdate()

    def get(self, session: "Optional[Dict[str, Any]]" = None) -> SessionUpdateJSONParse:
        """Parse json."""
        downloader = create_strategy("download", self.parse_config)
        output = downloader.get()
        cache = DataCache(self.parse_config.configuration.datacache_config)
        content = cache.get(output["key"])

        if isinstance(content, dict):
            return SessionUpdateJSONParse(content=content)
        return SessionUpdateJSONParse(content=json.loads(content))

get(self, session=None)

Parse json.

Source code in oteapi/strategies/parse/application_json.py

def get(self, session: "Optional[Dict[str, Any]]" = None) -> SessionUpdateJSONParse:
    """Parse json."""
    downloader = create_strategy("download", self.parse_config)
    output = downloader.get()
    cache = DataCache(self.parse_config.configuration.datacache_config)
    content = cache.get(output["key"])

    if isinstance(content, dict):
        return SessionUpdateJSONParse(content=content)
    return SessionUpdateJSONParse(content=json.loads(content))

initialize(self, session=None)

Initialize.

Source code in oteapi/strategies/parse/application_json.py

def initialize(self, session: "Optional[Dict[str, Any]]" = None) -> SessionUpdate:
    """Initialize."""
    return SessionUpdate()

JSONResourceConfig

Bases: ResourceConfig

JSON parse strategy filter config.

Source code in oteapi/strategies/parse/application_json.py

class JSONResourceConfig(ResourceConfig):
    """JSON parse strategy filter config."""

    configuration: JSONConfig = Field(
        JSONConfig(), description="JSON parse strategy-specific configuration."
    )

SessionUpdateJSONParse

Bases: SessionUpdate

Class for returning values from JSON Parse.

Source code in oteapi/strategies/parse/application_json.py

class SessionUpdateJSONParse(SessionUpdate):
    """Class for returning values from JSON Parse."""

    content: dict = Field(..., description="Content of the JSON document.")