Clean Architecture In Python

_core.py

from abc import ABC, ABCMeta, abstractmethod
from collections import namedtuple
from itertools import count


PayloadFactory = namedtuple('PayloadFactory', [
        'good', 'created', 'queued', 'unchanged', 'requires_auth',
        'permission_denied', 'not_found', 'invalid', 'error'
    ])
"""
This factory produces DTOs for the next level up so we can communicate
the *intent* of our return value without needlessly re-examining what
the domain has already discovered.

In this example, I use namedtuples as the DTOs so instead of defining
a class to produce them, I've simply used a namedtuple as the class. 
In a real application you could use a real class that runs 
serialization on the outputted data before spitting a DTO.

These could also be named ViewModels as they should only contain 
basic data structures ready to be inserted into some view.
"""


# I'm not a big fan of returning None and then having my
# callers try to guess at *why* the None was returned
# sometimes it's a useful return value to have, but as
# far as communicating failure I prefer exceptions
class FrobNotFound(Exception):
    pass


class CannotCreateFrob(Exception):
    pass


class InvalidFrobInput(Exception):
    pass


class FrobRepository(object):
    """
    Emulate a database. In a real application, you could wrap a database connection
    or even a SQLAlchemy/Django ORM repository with a similar sort of interface so you're not
    dependent on SQLAlchemy/Django ORM directly. Joins get a little weird in that case though.
    Consider that an exercise left for the reader.
    
    Or define this as an abstract class that you fulfill in multiple ways:
    
    ..code-block:: python
        class FrobRepository(ABC):
            ...
            
        class SQLAFrobRepository(FrobRepository):
            def __init__(self, session, model):
                self.session = session
                self.model = model
            
            def find_it(self, id):
                frob = self.session.query(self.model).get(id)
                if frob is None:
                    raise FrobNotFound(...)
                return frob
        
        class CachingFrobRepository(FrobRepository):
            def __init__(self, cache, repository):
                self._repository = repository
                self._cache = cache
            
            def find_it(id):
                frob = self._cache.get('frob::{!s}'.format(id), None)
                
                if frob is None:
                    frob = self._repository.find_it(id)
                    # set timeout for three minutes
                    self._cache.set('frob::{!s}'.format(id), frob, timeout=180)
                
                return frob
    """
    def __init__(self):
        self._frobs = {}
        self._next_id = count(1)
    
    def find_it(self, id):
        try:
            return self._frobs[id]
        except KeyError:
            raise FrobNotFound("Couldn't find frob: " + str(id)) from None
    
    def create_frob(self, **kwargs):
        if 'name' not in kwargs: 
            raise InvalidFrobInput("Must have name to be valid frob")
        
        id = next(self._next_id)
        r = randrange(2, 7)
        
        if not id % r:
            raise CannotCreateFrob("Constraint violated: {} % {} == 0".format(id, r))
        
        kwargs['id'] = id
        self._frobs[id] = kwargs
        return kwargs
    
    @property
    def available_frobs(self):
        return list(self._frobs.keys())


class FrobService(object):
    """
    This is basically an abstraction layer between data access and the whatever wants it.
    We can stick more specific, less broad rules here. In a multi tenant application,
    we might enforce things like a restriction on what frobs can be named by passing
    in a validator that the tenant defines:
    
        def __init__(self, frob_repository, payload_factory, validator=None):
            ...
            # validators should raise InvalidFrobInput
            self._validator = validator or lambda data: None
        
        def create_frob(self, data):
            try:
                self._validator(data)
                frob = self._frob_repository.create_frob(**data)
            except InvalidFrobInput as e:
                return self._payload_factory.invalid({'error': e.args[0]}, {})
            ...
    
    This layer also serves to translate information from the domain (the repository, etc)
    into datastructures that the thing calling it can use via the payload_factory
    dependency. In this example, it simply returns namedtuples but it could do things
    like run a serializer.
    """
    def __init__(self, frob_repository, payload_factory):
        self._frob_repository = frob_repository
        self._payload_factory = payload_factory
        
    def find_it(self, id):
        try:
            frob = self._frob_repository.find_it(id)
            return self._payload_factory.good(frob, {})
        except FrobNotFound:
            return self._payload_factory.not_found(
                {'error': 'Could not find frob with id: ' + str(id)},
                {'available_frobs': self._frob_repository.available_frobs}
            )
    
    def create_frob(self, data):
        try:
            frob = self._frob_repository.create_frob(**data)
        except CannotCreateFrob as e:
            return self._payload_factory.error({'error': e.args[0]}, {})
        except InvalidFrobInput as e:
            return self._payload_factory.invalid({'error': e.args[0]}, {'action': 'adjust input and try again'})
        else:
            return self._payload_factory.created(
                {'frob': frob},
                {'frob_id': frob['id']}
            )

        
class CreateFrobAction(object):
    """
    Represents an actual action an application would want to preform. This layer
    isn't strictly needed, but it does help to solidify a single business rule
    in the frame of the application. There's tons of ways that actions like this could
    be used:
    
    .. code-block:: python
        from flask import request
        cfa = CreateFrobAction(...)
        frob_view = View(...)
    
        @frob.request('/create', methods=['POST'])
        def create_frob():
            return view.render(cfa(request.form), request)
        
        # or...
        
        class FrobView(View):
            view = None
            
            def dispatch_request(self, *args, **kwargs):
                rv = super().dispatch_request(*args, **kwargs)
                return self.view.render(rv, request)
                
        
        class CreateFrobController(FrobView, MethodView):
            view = View(...)
        
            def __init__(self, frob_creator, form):
                self._frob_creator = frob_creator
                self._form = form
            
            def get(self):
                return HTTPPayload({'form': form}, {}, 200)
            
            def post(self):
                if self._form.validate_on_submit():
                    return self._frob_creator(**self._form.data)
                else:
                    return HTTPPayload({'form': form}, {}, 400)
    
    But in any case the top level caller serves as a translation 
    device between the outside world (web, terminal, files, etc) 
    and our actual application. The only real validation that should
    happen here are things like: 
    
    * Ensure the required arguments are present
    * Ensure the arguments are the correct type
    
    But not things like "Does the domain consider this a valid input?"
    because only the domain can ensure that.
    """
    def __init__(self, frob_service):
        self._frob_service = frob_service
    
    def __call__(self, request):
        return self._frob_service.create_frob(request.data)
        

class View(ABC):
    """
    NOT strictly an HTTP view (template, json, etc). This could handle things like
    console coloring if this is a terminal application or converting to a specific format
    for transport, or it could be a HTTP thing.
    
    The main method -- View::render -- receives both the output and the input so it can
    properly represent the output if the requester had specific instructions
    (content-type, signing key, no colors, etc).
    """
    @abstractmethod
    def render(self, payload, request):
        return NotImplemented


class Renderer(ABC):
    """
    Would be helper class for a View if needed. In the case of HTTP Views this
    could carry a media type such as `application/json` and the renderer method
    would call `json.dumps`. Or could carry platform specific encoding knowledge.
    """
    @abstractmethod
    def render(self, payload):
        return NotImplemented

_web.py

from app.core import View, Renderer
from collections import namedtuple
from functools import partial
import json
    
class UnsupportedMediaType(Exception):
    pass

Request = namedtuple('Request', ['data', 'headers'])

HTTP_PAYLOADS = {
    'good': 200, 'created': 201, 'queued': 202, 'unchanged': 304,
    'invalid': 400, 'requires_auth': 401, 'permission_denied': 403,
    'not_found': 404, 'error': 500
}

HTTPPayload = namedtuple('HTTPPayload', ['data', 'meta', 'status'])

HTTPPayloadFactory = PayloadFactory(**{
        name: partial(HTTPPayload, status=code) for name, code in HTTP_PAYLOADS.items()
    })

class HTTPView(View):
    """
    Example of a potential web view. Accepts multiple renderers and
    chooses the most appropriate one based on the request's Accept header.
    And finally displays a curl-ish output of the full response.
    """
    # stole this from Werkzeug because seriously
    # no one wants to type all this out
    HTTP_STATUS_CODES = {
        100:    'Continue',
        101:    'Switching Protocols',
        102:    'Processing',
        200:    'OK',
        201:    'Created',
        202:    'Accepted',
        203:    'Non Authoritative Information',
        204:    'No Content',
        205:    'Reset Content',
        206:    'Partial Content',
        207:    'Multi Status',
        226:    'IM Used',              # see RFC 3229
        300:    'Multiple Choices',
        301:    'Moved Permanently',
        302:    'Found',
        303:    'See Other',
        304:    'Not Modified',
        305:    'Use Proxy',
        307:    'Temporary Redirect',
        400:    'Bad Request',
        401:    'Unauthorized',
        402:    'Payment Required',     # unused
        403:    'Forbidden',
        404:    'Not Found',
        405:    'Method Not Allowed',
        406:    'Not Acceptable',
        407:    'Proxy Authentication Required',
        408:    'Request Timeout',
        409:    'Conflict',
        410:    'Gone',
        411:    'Length Required',
        412:    'Precondition Failed',
        413:    'Request Entity Too Large',
        414:    'Request URI Too Long',
        415:    'Unsupported Media Type',
        416:    'Requested Range Not Satisfiable',
        417:    'Expectation Failed',
        418:    'I\'m a teapot',  # see RFC 2324
        422:    'Unprocessable Entity',
        423:    'Locked',
        424:    'Failed Dependency',
        426:    'Upgrade Required',
        428:    'Precondition Required',  # see RFC 6585
        429:    'Too Many Requests',
        431:    'Request Header Fields Too Large',
        449:    'Retry With',  # proprietary MS extension
        451:    'Unavailable For Legal Reasons',
        500:    'Internal Server Error',
        501:    'Not Implemented',
        502:    'Bad Gateway',
        503:    'Service Unavailable',
        504:    'Gateway Timeout',
        505:    'HTTP Version Not Supported',
        507:    'Insufficient Storage',
        510:    'Not Extended'
    }
    
    def __init__(self, *renderers):
        self._renderers = renderers
    
    def render(self, payload, request):
        renderer = self._get_renderer(request)
        payload.meta.update({'Content-Type': renderer.media_type})
        print(self._get_full_status_code(payload))
        print(*("{}: {}".format(name, value) for name, value in payload.meta.items()), sep='\n')
        print()
        print(renderer.render(payload))
        print()
    
    def _get_full_status_code(self, payload):
        status = getattr(payload, 'status', 200)
        description = self.HTTP_STATUS_CODES.get(status, None)
        
        if description is None:
            raise ValueError("Got non-standard status code")
        
        return "HTTP/1.1 {!s} {}".format(status, description)
    
    def _get_renderer(self, request):
        accept = request.headers.get('Accept', 'text/plain')
        
        for renderer in self._renderers:
            if renderer.match(accept):
                return renderer
        else:
            raise UnsupportedMediaType("Cannot support media type: {}".format(accept))

            
class HTTPRenderer(Renderer):
    """
    Exposes a way to match the self identified media type against
    the one the client requested. Might seem overkill, but consider
    that application/yaml, application/x-yaml and text/yaml are all
    valid mimetypes for yaml. A yaml renderer could override the
    match method to be:
    
    .. code-block:: python
        def match(self, wanted):
            return wanted in self.media_types
    """
    media_type = None
    
    def match(self, wanted):
        return wanted == self.media_type
            
            
class PlainTextRenderer(HTTPRenderer):
    "Simple plain text renderer, simply spits out the resulting data"
    media_type = 'text/plain'
    
    def render(self, payload):
        return payload.data

    
class JSONRenderer(HTTPRenderer):
    "Simple json renderer, runs the resulting data through json.dumps"
    media_type = 'application/json'
    
    def render(self, payload):
        return json.dumps(payload.data, indent=4)

app.py

from app.core import FrobRepository, FrobService, CreateFrobAction
from app.web import HTTPPayloadFactory, HTTPView, JSONRenderer, HTTPRenderer, Request
from random import randrange

fr = FrobRepository()
fs = FrobService(fr, HTTPPayloadFactory)

CFA = CreateFrobAction(fs)
view = HTTPView(JSONRenderer(), PlainTextRenderer())

def use_cfa():
    names = ['jeff', 'fred', 'michael', 'thomas']

    for idx, name in enumerate(names, 1):
        accept = 'application/json' if (idx % 2 == 0) else 'text/plain'
        data = {'name': name} if name.lower() != 'fred' else {}
        req = Request(data=data, headers={'Accept': accept})
        view.render(CFA(req), req)

sample_output

HTTP/1.1 201 Created
frob_id: 1
Content-Type: text/plain

{'frob': {'id': 1, 'name': 'jeff'}}

HTTP/1.1 400 Bad Request
Content-Type: application/json
action: adjust input and try again

{
    "error": "Must have name to be valid frob"
}

HTTP/1.1 500 Internal Server Error
Content-Type: text/plain

{'error': 'Constraint violated: 2 % 2 == 0'}

HTTP/1.1 201 Created
frob_id: 3
Content-Type: application/json

{
    "frob": {
        "id": 3,
        "name": "thomas"
    }
}