8.13. Gateway

• EN: Gateway

• PL: Bramka

• Type: class

8.13.1. Pattern

8.13.2. Problem

import os
import requests


class HttpClient:
    def get(self, url):
        return requests.get(url).json()

    def post(self, url, data):
        return requests.post(url, data).json()


class HttpStub:
    def get(self, url):
        return {'status': 'ok'}

    def post(self, url, data):
        return {'status': 'created'}


if __name__ == '__main__':
    STAGE = os.getenv('STAGE', default='production')

    if STAGE == 'test':
        http = HttpStub()
    else:
        http = HttpClient()

    http.get('https://python3.info')

8.13.3. Solution

import os
import requests


class HttpClient:
    def get(self, url):
        return requests.get(url).json()

    def post(self, url, data):
        return requests.post(url, data).json()


class HttpStub:
    def get(self, url):
        return {'status': 'ok'}

    def post(self, url, data):
        return {'status': 'created'}


def gateway():
    STAGE = os.getenv('STAGE', default='production')
    if STAGE == 'test':
        return HttpStub()
    else:
        return HttpClient()


if __name__ == '__main__':
    http = gateway()
    http.get('https://python3.info')

8.13.4. Use Case - 1

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 Cache:
    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(Cache):
    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: Cache

    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 = MemoryCache(expiration=timedelta(minutes=2))

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

    print(html)

8.13.5. Assignments