DEV Community: jorzel

What It Means to Be a 10x Engineer

jorzel — Sat, 09 Nov 2024 11:33:30 +0000

The concept of the "10x engineer" often gets a lot of attention in the tech industry. However, it is frequently misunderstood as a person who can simply write 10 times more code or work 10 times faster than their peers. This narrow interpretation makes many believe that a "10x engineer" is just a myth — a unicorn that doesn't exist in reality. However, I believe this concept is a metaphor for someone who can bring transformational value to a team, well beyond what their raw output might suggest.

In reality, being a 10x engineer is less about volume and speed and more about making smarter decisions, driving meaningful outcomes, and leveraging skills and knowledge to maximize the impact on the business. Let's dive into what truly makes an engineer "10x."

Strategic Decision-Making

10x engineers don't just solve problems — they solve the right problems. They have a knack for eradicating tasks that do not bring value and do not contribute to real business goals.

They take time to understand the business domain and the real problems users or customers face, enabling them to make informed and strategic decisions.

A true 10x engineer knows that context is crucial. They have a deep understanding of the company's goals, the current stage of the product, and the evolving priorities. This allows them to adapt their focus based on what matters most at any given time — whether it's security, reliability, performance, or speed of delivery.

They also resist the temptation to implement complex, trendy solutions unless those solutions are truly the best fit for the problem. Technical tools are a means to achieve business goals — not an end in themselves.

Effective Communication

Among the Networkers, on the other hand, 99.9 per cent will have the innovation. Only 0.1 per cent will have solved it on their own, but the rest will have learned it from friends. And each of these will now have an opportunity to improve the innovation, transmitting insights back into their networks. The result is clear — and is corroborated by field data, lab experiments and dozens of historical examples. As Henrich puts it: If you want to have cool technology, it is better to be social than smart.

— Matthew Syed, Rebel Ideas

A 10x engineer does not need to be the smartest person in the room — they are a connector (networker). They understand the value of diverse perspectives and know that the fastest way to solve problems often lies in leveraging external insights (or experiences). Instead of trying to solve every problem from scratch, they tap into the collective knowledge available outside their immediate environment.

They often have a deep understanding of the company's internal landscape. They know which teams have tackled similar problems in the past, who the domain experts are for specific technologies, and what historical decisions have already been made. This awareness allows them to avoid redundancy, save time, and quickly find the best resources or people to solve a given issue.

A 10x engineer is a person who delivers knowledge to the team. It can be mentorship but also good relationships with other experts.

Data-Driven Orientation

Success isn't defined by the amount of code written or the speed of delivery. A 10x engineer focuses on whether the changes lead to meaningful improvements. By using data and metrics, they can confidently answer whether the released solution genuinely works, delivering not just output but valuable outcomes.

A 10x engineer doesn't just build and ship a solution — they validate its effectiveness. They take responsibility for ensuring that what they've delivered genuinely solves the problem and meets the desired outcomes. This means they go beyond manual testing and rely on comprehensive metrics and observability to confirm the solution works as intended in real-world scenarios.

They are ready to answer questions like:

Does the new solution genuinely work?
Did this feature improve the user's experience?
Did this refactoring break the functionality?

And they have data to prove it.

Craftsmanship

A 10x engineer is a lifelong learner who proactively seeks out opportunities to enhance their skills. They focus on improving their technical expertise, architecture skills, and decision-making capabilities, staying ahead of the curve.

They don't wait for formal training but take the initiative in identifying and learning the skills needed to solve complex problems effectively. This dedication ensures they are always evolving and bringing fresh insights to the team.

Their focus on software craftsmanship goes beyond writing functional code. They aim to create scalable, maintainable solutions with a keen eye for quality, building systems that are robust and well-designed.

They embrace rapid experimentation, knowing that small, quick failures pave the way for innovative solutions and minimize costs, creating space for new ideas.

Deep and Shallow Work Balance

They proactively protect time for deep, uninterrupted work where they can tackle complex problems. This structured approach helps them deliver high-quality output while staying engaged with the team and gathering necessary feedback.

They know how to design their work time to maximize focus, minimizing distractions to engage in deep work and high-quality output.

A 10x engineer excels at balancing deep, focused work with necessary shallow tasks such as Slack conversations, meetings, feedback sessions, or quick problem-solving. They understand that too much context-switching leads to busyness without meaningful progress, while too little collaboration can result in isolation and a lack of feedback.

Conclusion

The 10x engineer isn't a mythical figure producing 10 times more code — it's a metaphor for a type of engineer who consistently amplifies the team's impact through a combination of strategic thinking, domain awareness, crafted solutions, and effective communication. This concept represents an archetype, an ideal that underscores the importance of multidimensional growth.

The 10x engineer is not a fixed destination but a horizon to strive for, inspiring engineers to focus on what truly matters: outcomes over output, and balanced growth across all areas of their craft.

Originally published on Medium.

Which of these dimensions feels hardest to grow in your current role? For me it's protecting deep work — the shallow tasks always find a way back in.

If you enjoyed this, I write more about software engineering, design, and what makes developers grow. You can find more of my work here.

Cover photo by Malcolm Lightbody on Unsplash.

9 Laws That Every Software Developer Should Know

jorzel — Tue, 24 Sep 2024 21:13:36 +0000

In software development, there are numerous guidelines and observations referred to as laws or principles. While these are not strict formulas that hold universally in all situations, they provide important frameworks that influence the development process. These principles can significantly impact the productivity of organizations, teams, and individuals. It's valuable for anyone involved in software to be familiar with them.

Brooks's Law

Adding [human resources] to a late software project makes it later.

— Fred Brooks

Due to coordination costs, more developers added to a project don't always increase productivity. This law highlights the risks of "throwing more people" at a delayed project without proper planning.

Besides, it is important to find a balance between a too-large team and a small team consisting of members wearing multiple hats.

Source: Nathan S. Collier

Goodhart's Law

When a measure becomes a target, it ceases to be a good measure.

— Charles Goodhart

This means that once a specific metric is turned into a target or goal, people will start optimizing their behavior to meet that metric, often at the expense of the underlying goal it was meant to represent. We often aim for outcomes that are hard to quantify, and as a result, we rely on metrics that end up steering our efforts in directions we didn't intend. It could be:

Number of lines of code (to measure productivity)
Story points completed per sprint (to measure team velocity)
Code coverage (to measure testing quality)

To avoid this trap we don't have to abandon the data-driven approach. However, we need at least several metrics that drive our efforts. We have to constantly refine and reassess metrics, and in some cases don't avoid qualitative approaches.

Source: Gaurav Jain

Hyrum's Law

With a sufficient number of users of an API, it does not matter what you promise in the contract: all observable behaviors of your system will be depended on by somebody.

— Hyrum Wright

As your API gains more users, people will start to rely on behaviors that weren't intended or documented by the designers. Over time, even small, undocumented quirks or edge cases become "features" that users depend on, making changes or improvements to the API more complex without breaking something for someone.

This law highlights the challenges of maintaining backward compatibility and managing expectations as systems grow and evolve.

Source: Phil Sturgeon on Twitter

Conway's Law

Any organization that designs a system (defined broadly) will produce a design whose structure is a copy of the organization's communication structure.

— Melvin Conway

The structure of the software often mirrors the organizational structure that built it. If you blindly follow the existing team or department boundaries, you may end up with subdomains that are not well-aligned with the desired architecture or business capabilities.

The Inverse Conway Maneuver is the proactive application of Conway's Law. It suggests that if we want our software architecture to take on a specific shape or structure, we should first organize our teams and communication patterns to reflect that desired architecture.

Source: Luca Rossi, Refactoring

Linus's Law

Given enough eyeballs, all bugs are shallow.

— Linus Torvalds

It captures the essence of open-source collaboration, where broad community involvement helps identify and fix bugs more effectively than in closed systems. The idea is that the more people who inspect the code, the higher the likelihood that someone will notice and address bugs that others might miss.

Hofstadter's Law

It always takes longer than you expect, even when you take into account Hofstadter's Law.

— Douglas Hofstadter

Hofstadter's Law serves as a reminder of how consistently inaccurate we tend to be when estimating the time required for tasks, especially in software development.

It reinforces the importance of buffer time and managing expectations.

Source: PJ Milani

Kernighan's Law

Everyone knows that debugging is twice as hard as writing a program in the first place. So if you're as clever as you can be when you write it, how will you ever debug it?

— Brian Kernighan

Writing too complex code could be dangerous for system maintenance. If the code is written with excessive complexity, debugging becomes even more difficult, since you have to first decipher the logic before fixing any issues. Simplicity in coding is key — writing clear, maintainable code makes it easier to debug and improve in the long run.

Source: r/ProgrammerHumor

Peter Principle

People in a hierarchy tend to rise to a level of respective incompetence.

— Laurence Peter

Success often comes with a price. In many cases, individuals are promoted to higher and higher positions based on their achievements. However, there comes a point where the demands of the role may exceed their abilities.

In software, this is frequently seen when successful developers are promoted to managerial positions. The assumption is often made that leadership and soft skills naturally grow alongside technical expertise, but this isn't always true. A skilled coder may not necessarily have the same aptitude for managing people, leading teams, or handling the strategic demands of leadership, which can lead to challenges in their new role.

Source: Lotfi Alsouki on LinkedIn

Pareto Principle

For many outcomes, roughly 80% of consequences come from 20% of causes.

— Vilfredo Pareto

The Pareto Principle is widely applicable, and one of its key insights is that effort should be selective. The takeaway is that focusing on the most impactful areas — typically the 20% that yields 80% of the results — leads to greater success than spreading effort too thin. This also emphasizes that quality outweighs quantity and that true outcomes are more important than just the volume of generated output. Prioritizing what truly moves the needle helps achieve more meaningful and sustainable results.

Source: Growth Method

Originally published on Level Up Coding.

Which of these laws have bitten you the hardest in your career? For me it's Hyrum's Law and Hofstadter's Law — they show up on almost every project.

If you enjoyed this, I write more about software engineering, design, and what makes developers grow. You can find more of my work here.

Flask MVT. Refactor to service layer

jorzel — Sun, 05 Dec 2021 22:08:41 +0000

In this short post I would like to show how we can improve separation of concerns using Service Layer pattern within Model View Template approach.

Model View Template

MVT is an application building pattern, commonly used in frameworks like Flask (but also Django).
It consists of three separated components:

Model layer that is resposible for domain logic and persistence.
View that concerns about handling http request. It is a place where application API is defined. The view orchestrate application logic by interaction with model and executing requests to external services.
Template is a presentation layer (e.g. html templates) that handles User Interface part.

MVT approach often results in following code smells:

mixing resposibilites of handling http request (e.g. operating on query string params, cookies, headers, etc.) with application logic orchestration (strong coupling between a framework and business requirements).
complicated test setup (always need a http request to test application logic).
mixing resposibilities of domain logic and persistence (infrastructure concern)

Push your framework outside application logic

A solution that can ease first two of three mentioned smells is an introduction of a service layer (it can be class but also standalone function, depending on the specific case). It can decouple application logic from a framework we use (application features / use cases become agnostic of framework code). View is now resposible for handling requests and executing service layer methods / functions, while application logic is now only a concern of a service layer. It allows us to focus on application core at first, and implement API that enables entry to it afterwards. Thanks to the separation, API does not necessary need to be API supporting rendered templates but it can be CLI (Command Line Interface), REST API, GraphQL API, RPC, etc. The decision about API (and framework) can be deferred, but we still can devop our application features.
API and application logic separation could also improve our tests. We could now write tests of application core, covering logical paths extensively, while covering only happy path for API part.

Conclusion

This solution is not a silver bullet of course. We should consider it when our view endpoints are large and not trivial, and writing clean, readable code is challanging.

Persistence and domain model separation using SQLAlchemy ORM

jorzel — Thu, 02 Dec 2021 08:15:26 +0000

Introduction

You probably have heard about a test pyramid. It is the idea that the application should have proper balance of automated tests on different layers. There should be a lot of unit tests, significantly less integration tests and a few UI tests (End2End, functional). The reasons for this are maintenence cost and speed of particular test type. Unit tests are usually fast and isolated from the rest of the code (so are easy to setup and maintain).

That was the theory, but the reality in Python world is rather different. Most of applications exploit ORM (Object Relational Mapper) to setup domain models. So how this applications can be unit tested when domain classes are structurally binded to a database?
The first answer is: Tweak a little bit definition of unit test, to satisfy the need. If domain model tests need a database but are still relatively fast and easy to maintain, we can treat them as unit tests rather than integration tests. But, is it a real solution?
The second answer is: Separate your domain model from persistance model. So, here we go!

How to separate model?

To show commonly practiced domain and persistance model integration, we take SQLAlchemy library. It is probably the most popular ORM in python community (along with Django ORM). We define Restaurant and Table models using SQLAlchemy declarative mapping style.

# models.py
from sqlalchemy import Column, MetaData, String, Integer, create_engine
from sqlalchemy.orm import declarative_base, relationship

SQLALCHEMY_DATABASE_URI = "sqlite:///" # your db uri

engine = create_engine(SQLALCHEMY_DATABASE_URI)
metadata = MetaData(bind=engine)
Base = declarative_base(metadata=metadata)

class BookedTableException(Exception):
    pass

class Table(Base):
    id = Column(Integer, primary_key=True, autoincrement=True)
    restaurant_id = Column(Integer, ForeignKey("restaurant.id"))
    max_persons = Column(Integer, default=0)
    is_open = Column(Boolean, default=True)

    def can_book(self, persons: int) -> bool:
        if not self.is_open:
            return False
        if persons > self.max_persons:
            return False
        return True

    def book(self, persons: int) -> None:
        if self.can_book(persons):
            self.is_open = False
        else:
            raise BookedTableException(
                "{self} cannot be booked, becuase is not open now or is too small"
            )

class Restaurant(Base):
    id = Column(Integer, primary_key=True, autoincrement=True)
    tables = relationship("Table", lazy="dynamic")

    def _get_open_table(self, persons: int) -> Optional[Table]:
        return self.tables.filter(
            Table.max_persons >= persons, 
            Table.is_open.is_(True)
        ).first()

    def has_open_table(self, persons: int) -> bool:
        if self._get_open_table(persons):
            return True
        return False

    def book_table(self, persons: int) -> Optional[Table]:
        table = self._get_open_table(persons)
        if table:
            table.book(persons)
            return table
        raise BookedTableException("No open tables in restaurant")

We can see that our integrated model need Base object that can be instantiated only if we pass proper databse URI. So to test it we must provide database setup. But, is there any way to avoid it?

Separated model

# entities.py
from typing import List, Optional

class BookedTableException(Exception):
    pass

class Table:
    def __init__(self, table_id: int, max_persons: int, is_open: bool = True):
        self.id = table_id
        self.max_persons = max_persons
        self.is_open = is_open

    def can_book(self, persons: int) -> bool:
        if not self.is_open:
            return False
        if persons > self.max_persons:
            return False
        return True

    def book(self, persons: int) -> None:
        if self.can_book(persons):
            self.is_open = False
        else:
            raise BookedTableException(
                "{self} cannot be booked, becuase is not open now or is too small"
            )

class Restaurant:
    def __init__(self, restaurant_id: int, tables: List[Table]):
        self.id = restaurant_id
        self.tables = sorted(tables, key=lambda table: table.max_persons)

    def _get_open_table(self, persons: int) -> Optional[Table]:
        for table in self.tables:
            if table.can_book(persons):
                return table
        return None

    def has_open_table(self, persons: int) -> bool:
        if self._get_open_table(persons):
            return True
        return False

    def book_table(self, persons: int) -> Optional[Table]:
        table = self._get_open_table(persons)
        if table:
            table.book(persons)
            return table
        raise BookedTableException("No open tables in restaurant")

# orm.py
from sqlalchemy import Boolean, Column, ForeignKey, Integer, MetaData, create_engine
from sqlalchemy import Table as sa_Table
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import mapper, relationship
from entities import Restaurant, Table

SQLALCHEMY_DATABASE_URI = "sqlite:///" # your db uri

engine = create_engine(SQLALCHEMY_DATABASE_URI)
metadata = MetaData(bind=engine)
Base = declarative_base(metadata=metadata)

restaurant = sa_Table(
    "restaurant",
    metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
)

table = sa_Table(
    "table",
    metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("restaurant_id", Integer, ForeignKey("restaurant.id")),
    Column("max_persons", Integer),
    Column("is_open", Boolean),
)

def run_mappers():
    """
    Provides mapping between db tables and domain models.
    """
    mapper(
        Restaurant,
        restaurant,
        properties={"tables": relationship(Table, backref="restaurant")},
    )
    mapper(Table, table)

run_mappers() # it should be executed in the app runtime

We can see that models.py file was divided into entities.py file, consisting domain models that are plain python classes and don't know anything about ORM or a database, and orm.py file that defines database tables and mapping from a python class to a database table.

In the end, in test_entities.py we test Restaurant class method without need of the database setup. Cool? But it's not over yet.

# test_entities.py
import pytest
from entities.restaurant import Restaurant, Table

def test_restaurant_has_open_table_should_pass_if_any_table_in_restaurant_is_open_in_desired_capacity():
    open_table = Table(table_id=1, max_persons=5, is_open=True)
    restaurant = Restaurant(restaurant_id=1, tables=[open_table])

    assert restaurant.has_open_table(3)

What about unit tests at a service layer?

Thanks to the separation we can also unit test the application at a service layer (I have recently written something about service layer abstraction here) imitating usage of a database, so the whole application logic can be tested without any integration tests. To make it happen, we need:

Repository pattern, an abstraction serving access to data storage (e.g. database),
Unit of Work pattern, an abstraction that groups set of operations into transaction (e.g. database transaction).

We define these abstractions as interfaces and inject them into the service.

# uow.py
from abc import ABC, abstractmethod

class UnitOfWork(ABC):
    is_commited: bool
    is_rollbacked: bool

    @abstractmethod
    def __enter__(self):
        pass

    @abstractmethod
    def __exit__(self, *args):
        pass

    @abstractmethod
    def commit(self):
        pass

    @abstractmethod
    def rollback(self):
        pass

# repositories.py
from abc import ABC, abstractmethod

class RestaurantRepository(ABC):
    @abstractmethod
    def get(self, restaurant_id):
        pass

    @abstractmethod
    def add(self, restaurant):
        pass

# service.py
from uow import UnitOfWork
from repositories import RestaurantRepository

class RestaurantNotExist(Exception):
    pass

class BookingTableService:
    def __init__(
        self,
        restaurant_repository: RestaurantRepository,
        unit_of_work: UnitOfWork,
    ):
        self._restaurant_repository = restaurant_repository
        self._uow = unit_of_work

    def book_table(self, restuarant_id: int, persons: int) -> Table:
        restaurant = self._restaurant_repository.get(restaurant_id)
        if not restaurant:
            raise RestaurantNotExist

        with self._uow:
            restaurant.book_table(persons)

For testing needs we can use in-memory imlementation of Repository (MemoryRestaurantRepository) and Unit of Work (MemoryUnitOfWork), while in production we have Repository (SQLAlchemyRestaurantRepository) and Unit of Work (SQLAlchemyUnitOfWork) with a connection to the database.

# uow.py
from sqlalchemy.orm import Session

class MemoryUnitOfWork(UnitOfWork):
    def __init__(self):
        self.is_commited = False
        self.is_rollbacked = False

    def __enter__(self):
        self.is_commited = False
        self.is_rollbacked = False
        return self

    def __exit__(self, *args):
        pass

    def commit(self):
        is_commited = True

    def rollback(self):
        is_rollbacked = True

class SQLAlchemyUnitOfWork(UnitOfWork):
    def __init__(self, session: Session):
        self.session = session
        self.is_commited = False
        self.is_rollbacked = False

    def __enter__(self):
        self.is_commited = False
        self.is_rollbacked = False
        return self

    def __exit__(self, *args):
        try:
            self.commit()
        except Exception:
            self.rollback()

    def commit(self):
        self.is_commited = False
        self.session.commit()

    def rollback(self):
        self.is_rollbacked = False
        self.session.rollback()

# repositories.py
from typing import List, Optional
from sqlalchemy.orm import Query, Session
from entities import Restaurant

class MemoryRestaurantRepository(RestaurantRepository):
    def __init__(self):
        self._restaurants = {}

    def get(self, restaurant_id: int) -> Optional[Restaurant]:
        return self._restaurants.get(restaurant_id)

    def add(self, restaurant: Restaurant) -> None:
        self._restaurants[restuarant.id] = restaurant

class SQLAlchemyRestaurantRepository(RestaurantRepository):
    def __init__(self, session: Session):
        self.session = session

    def get(self, restaurant_id: int) -> Optional[Restaurant]:
        return self.session.query(Restuarant).get(restaurant_id)

    def add(self, restaurant: Restaurant) -> None:
        self.session.add(restaurant)
        self.session.flush()

# test_service.py
from typing import List, Optional
import pytest
from service import BookingTableService
from repositories import MemoryRestaurantRepository
from uow import MemoryUnitOfWork

@pytest.fixture
def restaurant_factory():
    def _restaurant_factory(
        restaurant_id: int,
        tables: Optional[List[Table]] = None,
        repository: RestaurantRepository = MemoryRestaurantRepository(),
    ):
        if not tables:
            tables = []
        restaurant = Restaurant(restaurant_id, tables)
        repository.add(restaurant)
        return restaurant
    yield _restaurant_factory

def test_booking_service_book_table_should_pass_when_table_in_restaurant_is_available(
    restaurant_factory
):
    repository = MemoryRestaurantRepository()
    uow = MemoryUnitOfWork()
    booking_service = BookingTableService(repository, uow)
    table = Table(table_id=1, max_persons=5, is_open=True)
    restaurant = restaurant_factory(
        restaurant_id=1, tables=[table], repository=repository
    )

    booking_service.book_table(restaurant_id=restaurant.id, persons=3)

    assert table.is_open is False
    assert uow.is_committed is True

Is the separation always a good idea?

We have seen that there is a need of some additional patterns (Dependency Injection, Repository, Unit of Work) and boilerplate code in this solution. So, it must be carefully considered whether given project need it. Here are some tips when we should go for it, becuase it brings some value and when we should avoid it, because it would be overkill or unnecessary complication.

For arguments:

unit testing without database setup.
infrastructure resposibility separated from domain. You can focus on business rules when your domain model is rich,
your model does not have direct mapping between persistance and domain, e.g. eventsourcing (you persist stream of events that are projected into stateful aggregate).

Against arguments:

connascene - code that is changed for the same reason should be in the same place. Assuming, we have direct mapping between domain model classes and tables in database, the change in domain model, trigger the change in persistance model and database table (more about connascene: https://connascence.io/),
easier and more popular SQLAlchemy declarative mapping API,
your project is a CRUD application. It doesn't follow classical test pyramid. You probably can test whole application through API (functional / integration tests), because there is no business logic,
performance issues, e.g. for filtering tables in integrated model we can execute indexed database query, while in domain model we have to iterate through the whole collection (for large collections it could be a problem).

That's all.

If you find this subject interesting, here is full implementation: https://github.com/jorzel/opentable/.
I also recommend great book and code repository that was inspiration for writing this post: https://github.com/cosmicpython/book.

Port and adapters architecture. Python + Nameko microexample.

jorzel — Wed, 01 Dec 2021 17:48:17 +0000

Introduction

Port and adapters (or hexagonal) architecture is a software design concept introduced by Alistair Cockburn in 2005. The main goal of it is to provide a clear seperation between application logic and external dependencies like database, user interface, framework providing HTTP requests, etc.

Application core, that is agonostic about external services and dependencies, should provide orchestration of whole business process exploiting existing ports. There are two types of ports:

primary port, that is an entry exposing application core to outside world. It is usually a fascade called by a primary adapter (e.g. REST API, CLI, etc.),
secondary port, enables application core to communicate with external world (e.g. database, mail sender, etc). It is an interface that is imlemented by a secondary adapter.

Implementation

Nameko is a python framework for building microservices providing simple HTTP requests, RPC and Messaging over AMQP protocol.
In our simple example we implement an application that register entries to the system and emit event that should be handled in other part of the system (or other microservice).

We define a secondary port as an interface that enables publishing domain events from application core. Depending on implementation it could publish messages to local event bus or genuine event broker like RabbitMQ (but implementation is a resposibility of an adapter).

from abc import ABC, abstractmethod

class EventPublisher(ABC):
    @abstractmethod
    def publish(self, events):
        pass

To implement secondary adapter for EventPublisher we make a wrapper for Nameko EventDispatcher class.

# domain/events.py
from datetime import datetime
from typing import Dict
from dataclasses import dataclass

@dataclass(frozen=True)
class DomainEvent:
    @property
    def as_dict(self) -> Dict:
        serialized = asdict(self)
        serialized["name"] = self.name
        return serialized

@dataclass(frozen=True)
class RegisteredEntryEvent(DomainEvent):
      email: str 
      name: str = 'RegisteredEntryEvent'
      timestamp: datetime = datetime.now()

# infrastructure/event_publisher.py
from typing import List
from nameko.events import EventDispatcher
from application.event_publisher import EventPublisher
from domain.event import DomainEvent

class NamekoEventPublisher(EventPublisher):
    def __init__(self, dispatcher: EventDispatcher):
        self._dispatcher = dispatcher

    def publish(self, events: List[DomainEvent]) -> None:
        for event in events:
            self._dispatcher(event.name, event.as_dict)

Primary port in our example is a simple application service defining registration use case. EventPublisher interface (not implementation) is injected into the service.

# application/service.py
from application.event_publisher import EventPublisher
from domain.event import RegisteredEntryEvent

class RegistrationService:
    def __init__(self, event_publisher: EventPublisher):
        self._event_publisher = event_publisher

    def register_entry(self, email: str) -> None:
        event = RegisteredEntryEvent(email=email)
        self._event_publisher.publish(event)

Nameko HTTP endpoint is a primary adapter for our RegistrationEntryService port. It handles HTTP request data and execute the service method.

import json
from nameko.events import EventDispatcher
from nameko.web.handlers import http
from werkzeug.wrappers import Request, Response

class NamekoRegistrationService:
    name = "nameko_registration_service"
    dispatcher = EventDispatcher()

    @http("POST", "/register")
    def register_entry(self, request: Request) -> Response:
        request_params = json.loads(request.data)
        email = request_params["email"]
        service = RegistrationService(
       event_dispatcher=NamekoEventPublisher(self.dispatcher),
        )
        service.register_entry(email)
        return Response(f"Registered entry for {email=}")

The crucial thing here is that our application core does not have any knowledge about infrastructure and API, it operates only on interfaces. If we would like to have a database access, we defined a repository interface and injected it into our service. Or, if there is a need to send a notification, we probably make a notification sender interface that can be implemented by SMS or Email sender adapter in the infrastructure layer. But inside the application core we know nothing about infrastructure implementations (thanks to it the application logic can be easily unit tested). And this is the main gain of using this architecture.

Summary

It was a microexample of hexagonal architecture base concepts using Python and Nameko. If you find it interesting, I recommend you to visit my github repository for extended implementation (including also a domain layer that was omitted here, for the sake of simplicity) of similar project (also Python, Nameko and Port and Adapters): https://github.com/jorzel/opentable. For more theoretical background about hexagonal architecture, go here.