from os import environ.. meta::

author:

Matt Harasymczuk <matt@python3.info>

published:

2025-01-01

updated:

2025-12-09

readtime:

5 minutes

tags:

python, design patterns, behavioral, gateway

5.8. Gateway

• EN: Gateway

• PL: Bramka

• Type: class

5.8.1. Pattern

5.8.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')

5.8.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')

5.8.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)

5.8.5. Use Case - 2

>>> from urllib.request import urlopen
>>> from os import getenv, environ
>>> import json
>>>
>>>
>>>
>>> class LiveHttpGateway:
...     def get(self, url):
...         print('Download from live server...')
...         resp = urlopen(url)
...         data = resp.read()
...         return data.decode('utf-8')
>>>
>>>
>>> class MockHttpGateway:
...     def get(self, url):
...         print('Return mock data...')
...         return """[
...             {"firstname": "Alice", "lastname": "Apricot", "email": "alice@example.com"},
...             {"firstname": "Bob", "lastname": "Blackthorn", "email": "bob@example.com"},
...             {"firstname": "Carol", "lastname": "Corn", "email": "carol@example.com"},
...             {"firstname": "Dave", "lastname": "Durian", "email": "dave@example.org"},
...             {"firstname": "Eve", "lastname": "Elderberry", "email": "eve@example.org"},
...             {"firstname": "Mallory", "lastname": "Melon", "email": "mallory@example.net"}
...         ]"""
>>>
>>>
>>> class HttpGateway:
...     def __init__(self):
...         STAGE = getenv('STAGE', default='production')
...         if STAGE == 'test':
...             self.strategy = MockHttpGateway()
...         else:
...             self.strategy = LiveHttpGateway()
...
...     def get(self, url):
...         return self.strategy.get(url)

>>> environ['STAGE'] = 'production'
>>>
>>> http = HttpGateway()
>>> resp = http.get('https://python3.info/_static/myusers.json')
Download from live server...
>>> data = json.loads(resp)
>>>
>>> for user in data:
...     print(user['email'])
...
alice@example.com
bob@example.com
carol@example.com
dave@example.org
eve@example.org
mallory@example.net

>>> environ['STAGE'] = 'test'
>>>
>>> http = HttpGateway()
>>> resp = http.get('https://python3.info/_static/myusers.json')
Return mock data...
>>> data = json.loads(resp)
>>>
>>> for user in data:
...     print(user['email'])
...
alice@example.com
bob@example.com
carol@example.com
dave@example.org
eve@example.org
mallory@example.net

5.8.6. Assignments