Package pyquaca

PyQuaca: a Python package to query and cache APIs or web services.

Sub-modules

pyquaca.api_requester

APIRequester class to handle requests to a given API endpoint.

pyquaca.cache

Cache class for caching query results.

pyquaca.json_cache

This module provides a JSONCache class to work with JSON data.

pyquaca.parser

Parser for parsing query responses before storing in cache.

pyquaca.query

Query Class to be extended for use with specific sites

pyquaca.requester

Requester class to handle requests to a given URL.

Classes

class APIRequester (url: str, api_key: str | None = None)

Expand source code

class APIRequester(Requester):
    """APIRequester class to handle requests to a given API endpoint."""

    def __init__(self, url: str, api_key: str | None = None):
        super().__init__(url)
        self.logger = self.logger.getChild("API")
        self.api_key = api_key

    def format_url(self, query_string: str) -> str:
        """Format the URL with the given query_string and API key if provided."""
        return self.base_url.format(search_string=query_string, api_key=self.api_key)

APIRequester class to handle requests to a given API endpoint.

Ancestors

• Requester

Methods

def format_url(self, query_string: str) ‑> str

Expand source code

def format_url(self, query_string: str) -> str:
    """Format the URL with the given query_string and API key if provided."""
    return self.base_url.format(search_string=query_string, api_key=self.api_key)

Format the URL with the given query_string and API key if provided.

Inherited members

• Requester:
  • request

class Cache (cache_dir: str, file_extension: str = '')

Expand source code

class Cache:
    """Cache class for caching query results."""

    def __init__(self, cache_dir: str, file_extension: str = "") -> None:
        self.cache_dir = cache_dir
        self.file_extension = file_extension
        # Create the cache directory if it does not exist
        Path(self.cache_dir).mkdir(parents=True, exist_ok=True)
        self.logger = logging.getLogger("Cache")

    def retrieve(self, key: str) -> None | str:
        """
        Retrieve data from the cache.

        Args:
            key (str): The cache key to lookup.

        Returns:
            the data string if it is found in the cache, None otherwise.
        """
        full_path = os.path.join(self.cache_dir, key + self.file_extension)
        try:
            with open(full_path, "r", encoding="utf-8") as cache_file:
                return cache_file.read()
        except (FileNotFoundError, PermissionError, IOError) as e:
            self.logger.error("Could not read from cache %s", e)
        return None

    def store(self, key: str, value: Any) -> bool:
        """
        Store data in the cache.

        Args:
            key (str): The key string.
            value (str): The data to store.

        Returns:
            True if the data is successfully stored, False otherwise.
        """
        if not key or len(key) < 1:
            self.logger.error("Invalid key")
            return False
        full_path = os.path.join(self.cache_dir, key + self.file_extension)
        try:
            with open(full_path, "w", encoding="utf-8") as cache_file:
                cache_file.write(value)
            return True
        except (PermissionError, IOError) as e:
            self.logger.error("Could not write to cache %s", e)
        return False

Cache class for caching query results.

Subclasses

• JSONCache

Methods

def retrieve(self, key: str) ‑> str | None

Expand source code

def retrieve(self, key: str) -> None | str:
    """
    Retrieve data from the cache.

    Args:
        key (str): The cache key to lookup.

    Returns:
        the data string if it is found in the cache, None otherwise.
    """
    full_path = os.path.join(self.cache_dir, key + self.file_extension)
    try:
        with open(full_path, "r", encoding="utf-8") as cache_file:
            return cache_file.read()
    except (FileNotFoundError, PermissionError, IOError) as e:
        self.logger.error("Could not read from cache %s", e)
    return None

Retrieve data from the cache.

Args

key : str

The cache key to lookup.

Returns

the data string if it is found in the cache, None otherwise.

def store(self, key: str, value: Any) ‑> bool

Expand source code

def store(self, key: str, value: Any) -> bool:
    """
    Store data in the cache.

    Args:
        key (str): The key string.
        value (str): The data to store.

    Returns:
        True if the data is successfully stored, False otherwise.
    """
    if not key or len(key) < 1:
        self.logger.error("Invalid key")
        return False
    full_path = os.path.join(self.cache_dir, key + self.file_extension)
    try:
        with open(full_path, "w", encoding="utf-8") as cache_file:
            cache_file.write(value)
        return True
    except (PermissionError, IOError) as e:
        self.logger.error("Could not write to cache %s", e)
    return False

Store data in the cache.

Args

key : str

The key string.

value : str

The data to store.

Returns

True if the data is successfully stored, False otherwise.

class JSONCache (cache_dir: str, file_extension: str = '.json')

Expand source code

class JSONCache(Cache):
    """Cache class for storing and retrieving objects as JSON data."""

    def __init__(self, cache_dir: str, file_extension: str = ".json") -> None:
        super().__init__(cache_dir, file_extension)
        self.logger = self.logger.getChild("JSONCache")

    def retrieve(self, key: str) -> None | Any:
        data = super().retrieve(key)
        if data:
            return json.loads(data)
        return None

    def store(self, key: str, value: Any) -> bool:
        if not key or len(key) < 1:
            self.logger.error("Invalid key")
            return False
        if not value:
            self.logger.error("Invalid value")
            return False
        try:
            serialized_data = json.dumps(value)
        except TypeError as e:
            self.logger.error("Could not serialize data: %s", e)
            return False
        return super().store(key, serialized_data)

Cache class for storing and retrieving objects as JSON data.

Ancestors

• Cache

Inherited members

• Cache:
  • retrieve
  • store

class Parser

Expand source code

class Parser:
    """
    Parser for parsing query responses before storing in cache.

    - chain(parser) method is used to chain parsers together.
    - parse(raw) method is used to parse the raw data and return the parsed data.
      This method should be overridden by subclasses to implement parsing logic.
    """

    def __init__(self) -> None:
        self.logger = logging.getLogger("Parser")
        self.next_parser: SupportsParseChain | None = None

    def chain(self, parser: SupportsParseChain) -> None:
        """
        Chain a parser to this parser.

        Args:
            parser: The parser to chain.

        Returns:
            The chained parser.
        """
        self.next_parser = parser

    def parse(self, raw: Any) -> Any:
        """
        Parse the raw data and return the parsed data.

        Args:
            raw: The raw data to parse.

        Returns:
            The parsed data.
        """
        if self.next_parser:
            return self.next_parser.parse(raw)
        return raw

Parser for parsing query responses before storing in cache.

• chain(parser) method is used to chain parsers together.
• parse(raw) method is used to parse the raw data and return the parsed data. This method should be overridden by subclasses to implement parsing logic.

Methods

def chain(self,
parser: SupportsParseChain) ‑> None

Expand source code

def chain(self, parser: SupportsParseChain) -> None:
    """
    Chain a parser to this parser.

    Args:
        parser: The parser to chain.

    Returns:
        The chained parser.
    """
    self.next_parser = parser

Chain a parser to this parser.

Args

parser

The parser to chain.

Returns

The chained parser.

def parse(self, raw: Any) ‑> Any

Expand source code

def parse(self, raw: Any) -> Any:
    """
    Parse the raw data and return the parsed data.

    Args:
        raw: The raw data to parse.

    Returns:
        The parsed data.
    """
    if self.next_parser:
        return self.next_parser.parse(raw)
    return raw

Parse the raw data and return the parsed data.

Args

raw

The raw data to parse.

Returns

The parsed data.

class Query (url: str,
config: QueryConfig | None = None)

Expand source code

class Query:  # pylint: disable=too-few-public-methods
    """Query Class to be extended for use with specific sites"""

    def __init__(self, url: str, config: Optional[QueryConfig] = None):
        if config is None:
            config = {}
        self.auth = config.get("auth", None)
        self.check_cache = config.get("check_cache", True)
        self.api_key = config.get("api_key", None)
        cache_path = config.get("cache_path", "cache")
        self.parser = config.get("parser", None)
        self.cache = config.get("cache", Cache(cache_path))
        self.requester = config.get("requester", Requester(url))
        service_name = self.__class__.__name__.lstrip("Query").lower()
        if len(service_name) == 0:
            service_name = "query"
        self.logger = logging.getLogger(f"{service_name}")
        self.logger.setLevel(logging.DEBUG)

    def query(self, query_string: str) -> Any:
        """Query the site with query_string"""
        query_string = query_string.lower()
        if self.check_cache and self.cache:
            cached: Any | None = self.cache.retrieve(query_string)
            if cached is not None:
                self.logger.info("Search string (%s) found in cache", query_string)
                return cached
            self.logger.info("Search string (%s) not found in cache", query_string)
        else:
            self.logger.info("Skipping Cache as requested")
        response = self.requester.request(query_string)
        if response is None:
            return None
        if self.parser:
            self.logger.debug("Parsing data")
            response = self.parser.parse(response)
        else:
            self.logger.debug("No parser found, returning raw data")
        if self.check_cache and self.cache:
            self.cache.store(query_string, response)
        return response

Query Class to be extended for use with specific sites

Methods

def query(self, query_string: str) ‑> Any

Expand source code

def query(self, query_string: str) -> Any:
    """Query the site with query_string"""
    query_string = query_string.lower()
    if self.check_cache and self.cache:
        cached: Any | None = self.cache.retrieve(query_string)
        if cached is not None:
            self.logger.info("Search string (%s) found in cache", query_string)
            return cached
        self.logger.info("Search string (%s) not found in cache", query_string)
    else:
        self.logger.info("Skipping Cache as requested")
    response = self.requester.request(query_string)
    if response is None:
        return None
    if self.parser:
        self.logger.debug("Parsing data")
        response = self.parser.parse(response)
    else:
        self.logger.debug("No parser found, returning raw data")
    if self.check_cache and self.cache:
        self.cache.store(query_string, response)
    return response

Query the site with query_string

class QueryConfig (*args, **kwargs)

Expand source code

class QueryConfig(TypedDict):
    """Configuration for Query Class"""

    auth: NotRequired[str]
    check_cache: NotRequired[bool]
    api_key: NotRequired[str]
    cache_path: NotRequired[str]
    parser: NotRequired[Parser]
    cache: NotRequired[Cache]
    requester: NotRequired[Requester]

Configuration for Query Class

Ancestors

• builtins.dict

Class variables

var api_key : str

The type of the None singleton.

var auth : str

The type of the None singleton.

var cache : Cache

The type of the None singleton.

var cache_path : str

The type of the None singleton.

var check_cache : bool

The type of the None singleton.

var parser : Parser

The type of the None singleton.

var requester : Requester

The type of the None singleton.

class Requester (url: str)

Expand source code

class Requester:
    """Requester class to handle requests to a given URL."""

    def __init__(self, url: str):
        self.base_url = url
        self.logger = logging.getLogger("Requester")

    def request(self, query_string: str) -> str | requests.Response | None:
        """Make a request with the given query_string"""
        self.logger.info("Requesting with query_string: %s", query_string)
        url = self.format_url(query_string)
        response = requests.get(url, timeout=5)
        if response.status_code != 200:
            self.logger.error("Request to %s failed (%s)", url, response.status_code)
            return None
        self.logger.debug("Received response from %s", url)
        return response

    def format_url(self, query_string: str) -> str:
        """Format the URL with the given query_string"""
        return self.base_url.format(search_string=query_string)

Requester class to handle requests to a given URL.

Subclasses

• APIRequester

Methods

def format_url(self, query_string: str) ‑> str

Expand source code

def format_url(self, query_string: str) -> str:
    """Format the URL with the given query_string"""
    return self.base_url.format(search_string=query_string)

Format the URL with the given query_string

def request(self, query_string: str) ‑> str | requests.models.Response | None

Expand source code

def request(self, query_string: str) -> str | requests.Response | None:
    """Make a request with the given query_string"""
    self.logger.info("Requesting with query_string: %s", query_string)
    url = self.format_url(query_string)
    response = requests.get(url, timeout=5)
    if response.status_code != 200:
        self.logger.error("Request to %s failed (%s)", url, response.status_code)
        return None
    self.logger.debug("Received response from %s", url)
    return response

Make a request with the given query_string