6.11. Gateway

• EN: Gateway

• PL: Bramka

• Type: class

6.11.1. Pattern

6.11.2. Problem

6.11.3. Solution

6.11.4. Use Case - 0x01

import logging
import os
from dataclasses import dataclass
from datetime import timedelta, datetime
import requests

logging.basicConfig(
        level=logging.INFO,
        format='"%(asctime).19s", "%(levelname)s", "%(message)s"'
)
log = logging.getLogger(__name__)


class ICache:
    def __init__(self, expiration: timedelta = timedelta(days=1), location: str = '') -> None:
        self.location = location
        self.expiration = expiration

    def get(self, key: str) -> str:
        raise NotImplementedError

    def set(self, key: str, value: str) -> None:
        raise NotImplementedError

    def is_valid(self, key: str) -> bool:
        raise NotImplementedError


class FilesystemCache(ICache):
    def __init__(self, location: str = 'tmp', *args, **kwargs) -> None:
        self.location = location
        super().__init__(*args, **kwargs)
        if not os.path.isdir(self.location):
            if os.path.isfile(self.location):
                os.remove(self.location)
            os.mkdir(self.location)

    def _get_cache_path(self, key: str) -> str:
        filename = key.replace('/', '-').replace(':', '').replace('--', '-')
        return os.path.join(self.location, filename)

    def get(self, key: str) -> str:
        filename = self._get_cache_path(key)
        if not os.path.isfile(filename):
            raise FileNotFoundError
        with open(filename, mode='r', encoding='utf-8') as file:
            return file.read()

    def set(self, key: str, value: str) -> None:
        filename = self._get_cache_path(key)
        if value is None:
            raise ValueError('Value cannot be None')
        with open(filename, mode='w', encoding='utf-8') as file:
            file.write(value)

    def is_valid(self, key):
        filename = self._get_cache_path(key)
        if not os.path.isfile(filename):
            return False
        last_modification = os.path.getmtime(filename)
        last_modification = datetime.fromtimestamp(last_modification)
        now = datetime.now()
        if (now - last_modification) > self.expiration:
            return False
        else:
            return True


@dataclass
class HTTPGateway:
    cache: ICache

    def get(self, url):
        if not self.cache.is_valid(url):
            log.info('Downloading...')
            html = requests.get(url).text
            self.cache.set(url, html)
            log.info('Done.')
        return self.cache.get(url)


if __name__ == '__main__':
    cache = FilesystemCache(expiration=timedelta(seconds=2), location='tmp')
    # cache = DatabaseCache(expiration=timedelta(minutes=2), location='database.sqlite')
    # cache = LocmemCache(expiration=timedelta(minutes=2))

    http = HTTPGateway(cache=cache)
    html = http.get('https://python3.info')

    print(html)

6.11.5. Assignments