Create a gist now

Instantly share code, notes, and snippets.

@justanr /
Last active Jul 12, 2017

What would you like to do?
Clean Architecture In Python
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):
class CannotCreateFrob(Exception):
class InvalidFrobInput(Exception):
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):
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
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):
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):
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):
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'})
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(**
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(
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).
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.
def render(self, payload):
return NotImplemented
from app.core import View, Renderer
from collections import namedtuple
from functools import partial
import json
class UnsupportedMediaType(Exception):
Request = namedtuple('Request', ['data', 'headers'])
'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
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(*("{}: {}".format(name, value) for name, value in payload.meta.items()), sep='\n')
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
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):
class JSONRenderer(HTTPRenderer):
"Simple json renderer, runs the resulting data through json.dumps"
media_type = 'application/json'
def render(self, payload):
return json.dumps(, indent=4)
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)
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 thanks for this example! But what is Entity in your example? I cannot see it


justanr commented Jun 24, 2017

@Infernion You're right, the crucial piece is missing - Frob itself, which should be front and center. There's a lot wrong with this example. It was thrown together as a quick example of taking a traditional three tier application - Presentation, Logic and Storage - and blowing the logic level of into a full blown application itself that relied on external bits to provide storage and presentation.

Looking over this again, some of the errors are obvious.

  • What the hell is frob and why is everything named after it? The critical piece is missing!

  • The repository is responsible for much more than it should be. Ideally it's goal is the translation between the domain model and the persistence model. Here it's also responsible for stuff like creating Frobs, validating frobs, etc.

  • FrobService is horribly named and will likely attract all sorts of cruft because of that. And CreateFrobAction is just a dumb wrapper around it. Seems like they could be merged into one thing.

  • Validation will be applied equally to all Frobs because of where it lives. However, validation is contextual. What is a valid Frob? Well that depends on where and how it was created. Instead this says there is one and only one way to create a valid Frob.

For a quick example, this isn't terrible but it doesn't really live up to the standard of either clean architecture or Domain Driven Design (the two seem to intertwined at the moment).

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment