Hexagonal Architecture

The idea of Hexagonal Architecture is to put inputs and outputs at the edges of our design. Business logic should not depend on whether we expose a REST or a GraphQL API, and it should not depend on where we get data from — a database, a microservice API exposed via gRPC or REST, or just a simple CSV file.
– Ready for changes with hexagonal architecture | netflix blog

Figure 3: Architecture of some imaginary application which uploads some documents to a storage system

• everything related to the business case
  
  • uploader wants to upload some document
    
  • PM wants to have some metrics
    
• contains
  
  • Entities
    
  • Services
    
  • Repositories/Interfaces
    

Figure 4: The business domain contains the application login and uses abstractions (interfaces) for defining interactions.

class Document ():                        ❶
	"""A document is an entity"""
	def __init__(self, path: FilePath):
		self._file_path = path            ❷

	def meta(self):
	"""Display meta information about the file"""
		print("Some information")

• We only have Document ❶ as an entity
  
• The constructor will set an instance variable ❷ for storing the file path
  

class UploadDocumentService:           ❶
	"""Upload a document to storage repository"""

	def __init__(
		self,
		storage_repository: DocumentStorageRepository,
		metrics_repository: DocumentMetricsRepository,
	):
		self._storage_repo: DocumentStorageRepository
		self._metrics_repo: DocumentMetricsRepository

	def upload_document(self, document: Document):   ❷
		self._storage_repository(document)
		self._metrics_repo.send_metrics()

• ❶ We have an UploadDocumentService
  
• ❷ this service implements upload_document(document: Document)
  

The repositories are basically interfaces for the secondary (driven) adapters. In our case we have:

• a repository for dealing with document storage
  
  • define how to save documents
    
  • define how to search for documents
    
  • define how to delete a document
    
• a repository for dealing with metrics
  

Figure 5: The interfaces are implemented by adapters which use concrete 3rd-party libraries for “external” calls.

from import abc import ABC, abstractmethod

class DocumentStorageRepository(ABC):      ❶
	""" Driven port defining an interface for storing documents """
	@abstractmethod
	def save(self, path: FilePath):
		pass

	@abstractmethod
	def search(self, uuid: uuid):
		pass

	@abstractmethod
	def delete(self, uuid: uuid):
		pass

from import abc import ABC, abstractmethod

class DocumentMetricsRepository(ABC):    ❶
""" Driven port defining an interface for sending metrics about documents"""
@abstractmethod
	def send_metrics(self):
		pass

# HTTP controller
from flask import Flask, request
from domain.services import UploadDocumentService
from domain.entities import Document

app = Flask(__name__)

class HTTPController:
    def __init__(self, upload_service: UploadDocumentService):  ❶
	    self.upload_service = upload_service

	@app.route(self, '/upload', methods=['POST'])
	def upload_document():
		"""Uploads a document using DocumentUploadService """
		# Create a document object
		doc = entities.Document(request.form.get("document_path"))
        self.upload_service.upload_document(doc)                ❷

• ❶ The HTTPController expects a UploadDocumentService
  
  • we use an abstraction rather than a concrete implementation
    
• ❷ We use the upload_service to upload a document (of type entities.Document)
  

# Simpl CLI controller
import sys, getopt
from domain.services import UploadDocumentService
from domain.entities import Document

class CLIController:
    def __init_(self, arguments, upload_service: UploadDocumentService):  ❶
	    self.args = arguments
		self.upload_service = upload_service

	def upload(self):
	   inputfile = ''
	   try:
		  opts, args = getopt.getopt(self.args,"hi:",["ifile="])
	   except getopt.GetoptError:
		  print 'cli.py -i <inputfile>'
		  sys.exit(2)

	   for opt, arg in opts:
		  if opt == '-h':
			 print 'cli.py -i <inputfile>'
			 sys.exit()

		  elif opt in ("-i", "--ifile"):
			 inputfile = arg


	   # Create document object
	   doc = Document(inputfile)
	   service.upload_service.upload_document(doc)                   ❷

• ❶ CLIController expects an UploadDocumentService
  
• uses its upload_document method to upload ❷ a document (of type entities.Document)
  

from domain.entities import Document

class S3StorageRepository:
    """Implements DocumentStorageRepository """
	def __init__(self):
	    # Initiate here connection to AWS S3
		self.s3_conn = ...

	def save(self, doc: Document):                   ❶
	    # Read file contents
		content = get_file_content(doc)
		self.s3_conn.create_new_object(content, ...) ❷

	def search(): List[Document]                     ❸
	    # Search in S3 buckets and return list of documents
		doc_list = []
		for f in results:
		    # Create a document
			doc = Document(f.path())
			doc_list.append(doc)
		return doc_list
	...

• S3StorageRepository implements DocumentStorageRepository (interface)
  
• save takes as an argument a Document and ❷ defines how and where to save the document
  
  • these are implementation specific details
    
• ❸ search will return a list of Document
  
  • instead of a S3 object
    

from domain.entities import Document

class ELKMetricsRepository:
    """Implements DocumentMetricsRepository """
	def __init__(self):
	    # Initiate here connection to ELK stack
		self.elk_conn = ...             ❶

    def send_metrics():
	    self.elk_conn.send_data(...)    ❷

• ELKMetricsRepository implements DocumentMetricsRepository (interface)
  
• ❶ and ❷ are ELK (Elasticsearch, Logstash, Kibana) specific implementation details