2022-10 Victor Dorneanu
Why should you chose 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 2: Source: https://threedots.tech/post/introducing-clean-architecture/
servers
handling requests from outside to your business applicationclients
As an uploader I’d like to upload documents to some infrastructure. After successful upload I’d like to get a link I can share with my friends.
– Uploader
Easy! Some observations:
As a product manager I’d like to see how many documents each uploader uploads and how many times he/she shares the link.
– Product Manager
Also easy! Again some observations:
Figure 3: Architecture of some imaginary application which uploads some documents to a storage system
Document
)DocumentUploadService
)DocumentStorageRepository
and DocumentMetricsRepository
)DocumentUploadService
for the terminalDocumentUploadService
via HTTP
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")
Document
❶ as an entityclass 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()
UploadDocumentService
upload_document(document: Document)
The repositories are basically interfaces for the secondary (driven) adapters. In our case we have:
save
documentssearch
for documentsdelete
a document
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
❶ DocumentStorageRepository
is an interface (a port) for describing which methods a document storage implementation should have
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
❶ DocumentMetricsRepository
is an interface (a port) for describing which methods a metrics system implementation should have
The handler only depends on entities and services.
# 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) ❷
HTTPController
expects a UploadDocumentService
upload_service
to upload a document (of type entities.Document
)
The handler only depends on entities and services.
# 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
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 documentsearch
will return a list of Document
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)
Here is where everything comes together.
package main
def main():
# Create new S3 storage repository
s3_repo = repositories.S3StorageRepository() ❶
# Create new ELK metrics repository
elk_repo = repositories.ELKMetricsRepository() ❷
# Create new upload service ❸
upload_service = services.UploadDocumentService(s3_repo, elk_repo)
# Create new HTTP controller/handler ❹
http_server = controller.HTTPController(upload_service)
http_server.Start()
if __name__ == "__main__":
main()
main
is where everything is glued togetherDocumentStorageRepository
and DocumentMetricsRepository
s3_repo
and elk_repo
satisfy this signatureUploadDocumentService
(service)upload_service
can be used as an argument to create the HTTP handler