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"
    }
}

@justanr I love the concept of Clean Architecture. I've been studying it like crazy over the past year. The obvious problems I've encountered are the lack of solid examples and how the principles apply to a dynamic typed languages, like Python. So I appreciate your ideas and also your honesty :)