fba/helpers/json.py

   1 # Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
   2 # Copyright (C) 2023 Free Software Foundation
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published
   6 # by the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
  16
  17 import logging
  18
  19 import json
  20 import requests
  21
  22 logging.basicConfig(level=logging.INFO)
  23 logger = logging.getLogger(__name__)
  24
  25 def from_response(response: requests.models.Response) -> any:
  26     logger.debug("response[]='%s' - CALLED!", type(response))
  27
  28     if not isinstance(response, requests.models.Response):
  29         raise ValueError(f"Parameter response[]='{type(response)}' is not type of 'Response'")
  30     elif not response.ok or response.status_code > 200:
  31         raise ValueError(f"response.ok='{response.ok}',response.status_code={response.status_code},response.reason='{response.reason}' but function was invoked")
  32     elif response.text.strip() != "" and response.headers.get("content-type") is not None and response.headers.get("content-type").split(";")[0] not in ["*/*", "application/json", "application/jrd+json", "application/activity+json"]:
  33         logger.warning("response.headers[content-type]='%s' is not a JSON type, below json() invocation may raise an exception", response.headers.get("content-type"))
  34
  35     data = list()
  36     if response.text.strip() != "":
  37         logger.debug("response.text()=%d is not empty, invoking response.json() ...", len(response.text))
  38         try:
  39             data = response.json()
  40
  41             logger.debug("data[]='%s' - EXIT!", type(data))
  42             if not isinstance(data, list) and not isinstance(data, dict):
  43                 logger.warning("data[]='%s' is not wanted, wrapping into 'dict'", type(data))
  44                 data = {data}
  45
  46         except json.decoder.JSONDecodeError as exception:
  47             logger.warning("Exception '%s' during decoding JSON from response.url='%s'", type(exception), response.url)
  48
  49     logger.debug("data[]='%s' - EXIT!", type(data))
  50     return data