DEV Community: Antonis Markoulis

How I Learned to Stop Worrying and Love Raw Events, Event Sourcing & CQRS with FastAPI and Celery

Antonis Markoulis — Wed, 03 Sep 2025 05:07:50 +0000

Have you ever found yourself staring at a production issue, wondering "What the hell happened here?" You know something went wrong, but the system has no memory of what actually occurred. If this sounds familiar, you're not alone.

In my years of building distributed systems, I've encountered this frustration countless times. Traditional architectures, while familiar and comfortable, often leave us blind when things go wrong. We lose context, we lose history, and we lose the ability to understand our own systems.

But what if I told you there's a different way? A way that gives your systems perfect memory, complete audit trails, and debugging superpowers that would make traditional approaches seem primitive?

This is the story of how I learned to stop worrying and love raw events. It's about event sourcing, CQRS, and how Python's ecosystem provides everything you need to build systems that can explain themselves.

The Problem: When Systems Forget

Let me start with a story that probably sounds familiar to many of you. It's Monday afternoon, and you're getting ready to wrap up your day when the Slack message arrives:

"Sarah's account is missing!"

You check the database. Nothing. The user record is gone. Completely vanished.

Tuesday morning, you're still trying to figure out what happened. When was it deleted? Who did it? Why? Your manager is asking questions, and you have no answers.

This is the nightmare scenario that every developer is afraid of. Your system has no memory of what happened, and you're left scrambling to piece together clues from logs, monitoring systems, and whatever else you can find.

Why Traditional Systems Fail Us

Here's the thing about traditional architectures: they're designed around the concept of current state. You have one row per entity in your database, and when something changes, you update that row. When something gets deleted, it's gone.

At best, if you're smart about it, you might implement soft deletes. You keep a deleted_at timestamp and maybe an updated_by field. But even then, you're only capturing the bare minimum. You lose the why - the reason for the deletion. You lose the context - what led up to this decision. You lose the sequence - the chain of events that brought you to this point.

And if you don't have soft deletes? Well, then you've got nothing. The record is gone forever, along with any trace that it ever existed.

This is the fundamental limitation of traditional systems. They're optimized for storing current state, not for remembering what happened. They're designed for efficiency, not for understanding.

But what if we flipped this on its head? What if, instead of storing current state, we stored every change that ever happened? What if our system had perfect memory?

The Solution: Events as the Source of Truth

This is where event sourcing comes in. Instead of storing current state, we store events - immutable facts that represent every change that ever happened in our system.

Think of events as the DNA of your application. Every change, every decision, every action is recorded as an event that can never be altered or deleted. Events are stored as facts without interpretation, and this is actually a good thing - it means different parts of your system can interpret the same event differently.

Here's what a UserDeleted event looks like:

id: 550e8400-e29b-41d4-a716-446655440000
aggregate_id: 123e4567-e89b-12d3-a456-426614174000
event_type: USER_DELETED
version: 1
timestamp: 2024-03-15T16:47:23Z
revision: 5
data:
  deleted_by: admin@company.com
  reason: User requested account deletion

Notice a few key things here:

Events are immutable facts - they never change once created
The version field lets us evolve our event structure over time while maintaining backward compatibility
The revision field ensures we can replay events in the correct order
The data field contains all the context we need - who deleted the user and why

This is the power of events. Instead of losing information, we're capturing everything. We're building a complete history of our system.

The Event Store: Your System's Memory

Now, where do we store all these events? In an Event Store - an append-only storage system where events are organized in streams per aggregate.

Think of it like this: each user has their own story, and that story is told through a sequence of events. Here's what Sarah's complete story looks like:

UserCreated - Sarah joins the platform
UserNameChanged - She updates her display name
UserEmailChanged - She changes her email address
UserStatusChanged - Her account gets activated
UserDeleted - She requests account deletion

The beautiful thing about this approach is that the stream becomes the source of truth. We can rebuild any point in time by replaying these events in order. Want to know what Sarah's account looked like last Tuesday? Just replay the events up to that point. This is how we get time travel capabilities.

CQRS: Separating What We Do from What We Read

Now, here's where things get interesting. In traditional systems, we use the same models for both writing and reading data. But what if we separated these concerns? What if we optimized our write path for consistency and our read path for performance?

This is the core idea behind CQRS (Command Query Responsibility Segregation).

Commands: Expressing Intent

Commands represent the intent to change the system state. They're like saying "I want to create a new user account" or "I want to change this user's password."

Commands are validated before execution, they're idempotent for safety, and they serve as the entry point for all changes. When a command is executed, it creates events that get stored in the event store.

Queries: Optimized for Reading

Queries, on the other hand, represent the intent to read data from the system. They're completely separate from commands and are optimized for specific access patterns.

Here's the key insight: your read models don't have to follow database normalization rules. They can be denormalized for performance. You can have multiple read models for the same data, each optimized for different use cases.

This separation gives you the best of both worlds: strong consistency on the write side, and high performance on the read side.

Aggregates: Where Business Logic Lives

So far, we've talked about events and commands, but where does the actual business logic live? That's where aggregates come in.

Aggregates are the heart of your domain logic. They contain all the business rules and apply them to create events. Here's how it works:

First, we rebuild the aggregate's current state by applying all previous events from the event store
Once the aggregate is up-to-date, we call a business logic method that validates the command against business rules
When rules pass, the method creates new events
When rules fail, it returns errors

This ensures business invariants are maintained while keeping the aggregate state current.

Projections: Building Read Models from Events

Now, here's the final piece of the puzzle. We have events being created by aggregates, but these events are just stored in the event store. How do we actually use them to build the read models that our queries will use?

That's where projections come in. Projections listen to events and build optimized read models for fast querying.

When a UserCreated event happens, we create a user record. When EmailChanged happens, we update the email field. This gives us event-driven read model updates that are optimized for queries and eventually consistent.

The beautiful thing about projections is that one event can trigger multiple actions. The same UserCreated event might:

Create a user record in the read model
Send a welcome email
Create an audit log entry
Trigger analytics tracking

This decoupling allows us to add new behaviors without changing existing code.

From Theory to Practice: Building It with Python

Alright, so we've covered the theory. Now let's talk about how to actually build this stuff. The good news? The Python ecosystem has everything you need to implement event sourcing with CQRS effectively.

We'll use FastAPI for the web layer, Celery for asynchronous event processing, and see how all the pieces fit together to create a robust, scalable system.

But first, let me show you how everything connects in practice.

The Complete Picture

Let me walk you through how this all works together in practice. This diagram shows the two distinct paths: commands (write) and queries (read).

The Command Path (Write): When a user wants to change their password, they make an API call to our FastAPI endpoint. This creates a ChangePasswordCommand and passes it to the command handler. The command handler then:

Retrieves all previous events for this user from the event store
Creates a UserAggregate and rebuilds its current state by applying all those events
Calls the aggregate's business method change_password, which validates business rules and generates new events
Stores the new events in the event store and dispatches them to the event bus

The Query Path (Read): When someone wants to get user data, they make a query API call. This goes directly to our FastAPI query endpoint, which uses the query handler to fetch data from the optimized read models.

This is CQRS in action - commands go through the full event sourcing pipeline, queries read directly from optimized read models. Same API entry point, completely different paths.

FastAPI: Command & Query Interface

Here's how we implement the CQRS separation in FastAPI:

@router.post("/users/{user_id}/change-password/")
async def change_password(
    user_id: UUID,
    password_data: ChangePasswordDTO,
    handler: ChangePasswordCommandHandler = Depends(InfraFactory.create_change_password_command_handler)
) -> ChangePasswordResponseDTO:
    command = ChangePasswordCommand(user_id=user_id, password_data=password_data)
    await handler.handle(command)
    return ChangePasswordResponseDTO(success=True, message="Password updated successfully")

@users_router.get("/{user_id}/")
async def get_user(
    user_id: UUID,
    query_handler: GetUserQueryHandler = Depends(InfraFactory.create_get_user_query_handler)
) -> UserReadDTO:
    return await query_handler.handle(GetUserQuery(user_id=user_id))

Our APIs are very simple - they use Pydantic models for both the request payloads and the internal commands/queries. When a request comes in, we create the appropriate command or query object using Pydantic, get the handler from our infrastructure factory through dependency injection, and simply call its handle method.

Command Handlers: Orchestration Logic

Behind the API, command handlers are the orchestrators - they don't contain business logic, they delegate and coordinate. Here's our ChangePasswordCommandHandler:

class ChangePasswordCommandHandler(CommandHandler[ChangePasswordCommand]):
    async def handle(self, command: ChangePasswordCommand) -> None:
        # Get events and rebuild aggregate state
        events = await self.event_store.get_stream(command.user_id)
        user = UserAggregate(command.user_id)
        for event in events:
            user.apply(event)

        # Execute business logic and persist
        new_events = user.change_password(
            command.password_data.current_password,
            command.password_data.new_password
        )

        # Persist and dispatch events using unit of work
        async with self.uow:
            await self.event_store.append_to_stream(command.user_id, new_events)
            await self.event_handler.dispatch(new_events)

The Unit of Work pattern (async with self.uow:) is crucial here - it ensures that both storing events to the event store AND dispatching them to the event bus happen atomically. If either operation fails, the entire transaction is rolled back.

Aggregates: Pure Python Business Logic

Aggregates are pure Python classes that contain all the business rules and domain logic. They don't have any dependencies on infrastructure, databases, or external services. They just take commands (like change_password) and produce events (like PASSWORD_CHANGED).

class UserAggregate(Aggregate):
    """User domain aggregate - encapsulates user business logic"""

    def change_password(self, current_password: str, new_password: str) -> List[EventDTO]:
        if not self._verify_password(current_password):
            raise InvalidPasswordError("Current password is incorrect")
        if len(new_password) < 8:
            raise ValidationError("Password must be at least 8 characters")
        return [PasswordChangedEvent(
            aggregate_id=self.id,
            changed_at=datetime.utcnow(),
            password_hash=hash_password(new_password)
        )]

    def apply(self, event: EventDTO) -> None:
        if event.event_type == EventType.PASSWORD_CHANGED:
            self._apply_password_changed_event(event)
        # ... other event types

The apply method is crucial for event sourcing - it's how we reconstruct the aggregate's current state by replaying all previous events.

Event Handler: Celery Integration

Our CeleryEventHandler acts as the bridge between our domain events and Celery. It maps each event type to one or more Celery tasks:

class CeleryEventHandler(EventHandler):
    def __init__(self):
        # Map event types to Celery tasks
        self.event_handlers: Dict[EventType, List[str]] = {
            EventType.USER_CREATED: [
                "process_user_created_task",
                "send_welcome_email_task"
            ],
            # ... other event types
        }

    async def dispatch(self, events: List[Event]) -> None:
        for event in events:
            for task_name in self.event_handlers[event.event_type]:
                # All tasks receive the same event payload structure
                celery_app.send_task(task_name, kwargs={"event": event.model_dump()})

This mapping is crucial - it's how we define what should happen when each type of event occurs. We can easily add new side effects by just adding new tasks to the mapping, without touching the aggregate or command handler code.

Celery Tasks: Event Processing

On the receiving end, Celery tasks are wrappers that call the appropriate projection handlers:

@app.task(
    name="process_user_created_task", bind=True, acks_late=True,
)
def process_user_created_task(self, event: Dict[str, Any]) -> None:
    """Celery task for processing USER_CREATED events"""
    # Get infrastructure factory
    factory = get_infrastructure_factory()

    # Get projection
    projection = factory.create_user_created_projection()

    # Process the event using async_to_sync
    async_to_sync(projection.handle)(EventDTO(**event))
    logger.info(f"Successfully processed USER_CREATED event for user {EventDTO(**event).aggregate_id}")

Reliability is crucial here. Notice the acks_late=True parameter - this means Celery won't acknowledge the message until the task completes successfully. If the task fails, the message goes back to the queue for retry.

Projections: Event-Driven Read Models

Projections are the final piece of the event sourcing puzzle - they take events and transform them into optimized read models for fast querying:

class UserCreatedProjection(Projection[UserCreatedEvent]):
    async def handle(self, event: UserCreatedEvent) -> None:
        # Build read model from event
        user_data = {
            "aggregate_id": event.aggregate_id,
            "name": event.data.get("name"),
            "email": event.data.get("email"),
            "status": event.data.get("status"),
            "created_at": event.timestamp,
        }

        # Save to read model
        await self.db.save(user_data)

Projections focus solely on one thing: ensuring the read model is properly updated from events. They don't handle business logic (that's the aggregate's job), they don't orchestrate the process (that's the command handler's job), and they don't manage storage details (that's the repository's job).

The Reality Check: What Actually Happens in Production

So far, this all sounds great in theory. But what happens when you actually try to build this stuff and put it in production? Well, that's where things get interesting.

The transition from theory to practice often reveals challenges that you don't see in the documentation. Eventual consistency isn't just a technical concept - it's a UX challenge. Performance issues emerge as your event streams grow longer. And debugging becomes both easier and more complex at the same time.

Let me share some of the real-world lessons I've learned from implementing event sourcing and CQRS in production environments. These insights will help you avoid common pitfalls and make informed decisions about your own implementations.

The Eventual Consistency UX Problem

Here's a scenario that will sound familiar to anyone who's worked with event sourcing systems. Sarah updates her name from "Sarah" to "Sara" in your application. The API returns success immediately, but when the frontend calls the backend to get the updated data, it still shows "Sarah".

This is the reality of eventual consistency - your API says "success" but the read model might not be updated yet. The command has been processed, the event has been stored, but the projection that updates the read model is still running in the background.

I've seen two approaches to handle this:

Optimistic Updates (Naive): The frontend updates the UI immediately, but if the user refreshes or calls the backend, they might see old data. This is actually the approach I've used most often in production, and for many use cases, it's proven more than sufficient.
Outbox Pattern (Advanced): We store events in an outbox table with job status, track processing status like pending, processing, completed, or failed, and create views of unprocessed events.

The key insight is that eventual consistency isn't just a technical challenge - it's a UX challenge. Users expect immediate feedback, but they also expect consistency.

Performance with Snapshots

Another real-world challenge in event sourcing is performance degradation as event streams grow longer. Consider a user who has been active for 2 years and accumulated 10,000+ events. When they try to change their password, the system has to replay all those events to reconstruct their current state, which can take 5 seconds or more.

The solution is snapshots - we save state every 1,000 events:

async def handle(self, command: ChangePasswordCommand) -> None:
    snapshot = await self.snapshot_store.get_latest_snapshot(command.user_id)
    events = await self.event_store.get_stream(command.user_id, start_revision=snapshot.revision)
    user = UserAggregate.from_snapshot(snapshot)
    for event in events:  # Only 50 events instead of 10,000!
        user.apply(event)

Instead of replaying 10,000 events, we only replay the recent 50 events since the last snapshot. This reduces the password change time from 5 seconds to 0.5 seconds.

Debugging Superpowers: Testing Business Logic

Instead of trying to recreate scenarios in test environments, we can rebuild the exact state at any moment in history. This diagram shows how we can trace through the event stream to understand exactly what happened and when. The debugging superpowers come from the ability to replay events and see the system state evolve over time.

Here's how we can test the exact scenario that caused Sarah's account deletion:

class TestSarahAccountDeletion(AsyncIOIsolatedTestCase):
    async def test_sarah_account_deletion_scenario(self):
        # Arrange - Replay Sarah's exact production events
        sarah = UserAggregate()
        sarah.apply(UserCreatedEvent("sarah_456", name="Sarah", email="sarah@company.com"))
        sarah.apply(UserLoginEvent("sarah_456", ip="192.168.1.100"))
        sarah.apply(UserProfileUpdatedEvent("sarah_456", field="role", value="admin"))

        # Act - Test the deletion event that caused the issue
        result = sarah.apply(UserDeletedEvent("sarah_456", deleted_by="admin@company.com", reason="User requested account deletion"))

        # Assert - Verify business logic behavior
        self.assertTrue(result.is_success)
        self.assertEqual(sarah.deleted_by, "admin@company.com")
        self.assertEqual(sarah.deletion_reason, "User requested account deletion")

This gives us the ability to debug issues that happened hours or days ago, and test business logic against real production data at any point in time. This is debugging and testing superpowers combined - we can answer "What was Sarah's state when her account was deleted?" with complete certainty.

Summary: Key Takeaways and Strategic Insights

After implementing event sourcing in production environments, I've learned that success comes down to making the right decisions at the right time. Let me share the strategic insights that will help you navigate this architectural choice.

🚀 Start Simple: The Power of Incremental Adoption

One of the biggest mistakes I see teams make is trying to implement event sourcing with all the bells and whistles from day one. You don't need Kafka, you don't need specialized event stores, and you definitely don't need to rewrite your entire system.

Start with what you know: PostgreSQL as your event store works perfectly fine for most applications. Redis or even a simple message queue can handle your event bus needs initially. The Python ecosystem gives you everything you need - FastAPI, Celery, SQLAlchemy, and Pydantic are all production-ready and battle-tested.

Why this matters: By starting simple, you can prove the value of event sourcing to your team and stakeholders before investing in more complex infrastructure. You can learn the patterns, understand the trade-offs, and build confidence in the approach. Once you've validated the benefits, you can always evolve to more sophisticated tooling.

⚠️ When NOT to use Event Sourcing: The Honest Assessment

Event sourcing isn't a silver bullet, and I've seen teams struggle when they try to apply it everywhere. Here's my honest assessment of when to avoid it:

Simple CRUD with basic audit needs: If you're building a straightforward CRUD application where you only need to know "who changed what and when," traditional audit logging is probably sufficient. Event sourcing adds complexity that you don't need.

High-frequency systems requiring immediate consistency: Trading systems, real-time gaming, or any system where milliseconds matter and you need immediate consistency across all reads - event sourcing's eventual consistency model will cause more problems than it solves.

Teams without distributed systems experience: Event sourcing introduces concepts like eventual consistency, event replay, and projection management that can be overwhelming for teams new to distributed systems. Start with simpler patterns first, build that experience, then consider event sourcing.

The key insight: Event sourcing is a powerful tool, but it's not the right tool for every job. The complexity it introduces needs to be justified by the benefits it provides.

🎯 What you gain: The Strategic Advantages

When event sourcing is the right fit, the benefits are transformative:

Complete audit trail & time travel: This isn't just about compliance or debugging - it's about building systems that can explain themselves. When something goes wrong, you can see exactly what happened, when it happened, and why it happened. This level of observability is invaluable for building reliable systems.

Debugging superpowers with real production data: Instead of trying to recreate issues in test environments, you can replay the exact sequence of events that caused problems in production. This isn't just faster debugging - it's more accurate debugging. You're working with real data, real timing, and real conditions.

Scalability with eventual consistency: By separating your write and read concerns, you can scale them independently. Your write path can be optimized for consistency and correctness, while your read path can be optimized for performance and user experience. This separation of concerns is crucial for building systems that can grow.

The strategic advantage: Event sourcing gives you systems that are not just functional, but understandable. They're not just scalable, but debuggable. They're not just fast, but reliable.

Conclusion

The Python ecosystem is incredibly powerful for distributed systems. FastAPI's dependency injection system makes it trivial to wire up command and query handlers. Celery's battle-tested task queue handles event processing reliably. Pydantic's validation ensures your events are always well-formed. You don't need to look outside Python for distributed systems capabilities.

This combination gives you everything you need to build systems that can explain themselves, scale gracefully, and handle real-world complexity. Most importantly, we've solved the exact problem we started with - we can now tell Sarah exactly when, why, and by whom her account was deleted. The nightmare becomes a solvable mystery with a complete audit trail.

Resources

Full Implementation: github.com/anmarkoulis/event-sourcing

This post is based on my presentation at PyCon Greece 2025. Feel free to explore the complete implementation on GitHub and adapt these patterns to your own projects.

Layered Architecture & Dependency Injection: A Recipe for Clean and Testable FastAPI Code

Antonis Markoulis — Thu, 29 May 2025 05:50:52 +0000

Introduction

Building a FastAPI application that scales? It's not just about async endpoints and Pydantic models. In this post, I'll share how to structure FastAPI applications for long-term maintainability using layered architecture and dependency injection. You'll learn how to keep your code organized, write testable code, and build a flexible system that can evolve with your needs.

Understanding Layered Architecture

Layered architecture (also known as n-tier architecture) is a pattern that organizes code into distinct layers, each with a specific responsibility. This separation of concerns makes the codebase more maintainable, testable, and flexible. Think of it like a well-organized kitchen in a restaurant: each station has its specific role, and the flow of work moves in one direction, from preparation to plating.

Layer 1: Presentation Layer

The presentation layer is your application's front door. It's where FastAPI lives, along with other entry points like Celery tasks and Typer commands. This layer is responsible for handling all external communication, whether it's HTTP requests, background jobs, or CLI commands.

Here's a typical example of how this layer looks in practice:

@router.post("/foos")
async def create_foo(
    foo: FooCreateDTO,
    foo_service: FooService = Depends(get_foo_service)
) -> FooResponseDTO:
    return await foo_service.create_foo(foo)

Key characteristics:

Minimal code focusing on input/output mapping: The layer acts as a translator between external formats (HTTP, Celery, CLI) and your service layer's standardized format, handling both request transformation and response formatting.
Handles service injection through FastAPI's dependency injection system: The layer leverages FastAPI's built-in dependency injection to provide services with their required dependencies, making the code more testable and maintainable.
Consistent pattern across all entry points: Whether you're handling an HTTP request, a background job, or a CLI command, the same pattern is followed, making the code predictable and easier to maintain.

Layer 2: Service Layer (Business)

This is where the magic happens. The service layer is the heart of your application, containing all business logic and transaction management. It's like the kitchen in our restaurant analogy - where the actual cooking (business logic) takes place.

Let's look at how we structure this layer:

class FooServiceInterface(ABC):
    @abstractmethod
    async def create_foo(self, foo: FooCreateDTO) -> FooResponseDTO:
        ...

class StandardFooService(FooServiceInterface):
    def __init__(self, foo_dao: FooDAOInterface, uow: UnitOfWorkInterface):
        self.foo_dao = foo_dao
        self.uow = uow

    async def create_foo(self, foo: FooCreateDTO) -> FooResponseDTO:
        async with self.uow:
            return await self.foo_dao.create(foo)

Key features:

Interface-based design that allows for multiple implementations: By defining clear interfaces, we can easily swap implementations without affecting the rest of the system.
Transaction management through the Unit of Work pattern: The Unit of Work pattern ensures that all database operations within a transaction either succeed together or fail together.
Centralized business logic that's easy to find and modify: All business rules and operations are contained within the service layer, making them easy to locate and update.
Self-documenting code through clear method names and responsibilities: The service layer's methods and classes are named to clearly indicate their purpose and responsibilities.

The service layer is where you'll spend most of your time implementing business rules. It's also where you'll handle complex operations that might involve multiple data access operations, ensuring they all succeed or fail together.

Layer 3: Persistence Layer (DAOs)

The persistence layer is your application's memory. It handles all database interactions through Data Access Objects (DAOs), which are like specialized waiters in our restaurant analogy - they know exactly how to interact with the storage system.

Here's how we implement this layer:

class FooDAOInterface(ABC):
    @abstractmethod
    async def create(self, foo: FooDTO) -> FooDTO:
        ...

class SQLAlchemyFooDAO(FooDAOInterface):
    def __init__(self, session: AsyncSession):
        self.session = session

    async def create(self, foo: FooDTO) -> FooDTO:
        db_foo = Foo(**foo.model_dump())
        self.session.add(db_foo)
        await self.session.flush()
        return FooDTO.model_validate(db_foo)

Key aspects:

Interface-based DAO design: Each DAO implements a clear interface that defines the contract for data access operations.
Transaction management at service level: The DAO layer never handles commits, leaving transaction management to the service layer for better control.
Flexible implementation swapping: The interface-based design allows easy switching between different ORMs or data storage solutions without affecting the service layer.

DTOs: The Glue Between Layers

DTOs (Data Transfer Objects) are the messengers between your layers. They're like the standardized order forms in our restaurant - they ensure everyone speaks the same language.

Here's how we define our DTOs:

from pydantic import BaseModel
from decimal import Decimal
from typing import List

class FooItemDTO(BaseModel):
    name: str
    quantity: int

class FooDTO(BaseModel):
    id: str
    name: str
    items: List[FooItemDTO]
    total: Decimal

Characteristics:

Immutable data transfer: DTOs are implemented as frozen Pydantic models, ensuring data consistency as they flow between layers.
Validation-free by design: DTOs typically don't include validation logic, except when used as FastAPI input models where we can add API-specific validation methods.
Type-safe layer communication: DTOs ensure type safety as data moves between layers, making the code more maintainable and catching errors early.
JSON serialization ready: DTOs are designed to be easily serializable to and from JSON, making them perfect for API responses and external communication.

Dependency Injection with Dependency Service

Dependency injection is like having a well-organized kitchen where each chef knows exactly what tools they need. Our DependencyService is the kitchen manager that ensures everyone has what they need.

Here's how we implement it:

class DependencyService:
    @staticmethod
    def get_foo_service(
        db: AsyncSession = Depends(get_db),
    ) -> FooServiceInterface:
        uow = SQLAUnitOfWork(db)
        return FooService(
            foo_dao=FooDAO(uow.db),
            foo_client=FooClient(),
            uow=uow,
        )

Benefits:

Transparent service assembly: The dependency service clearly shows what each service needs, making it easy to understand and modify service configurations.
Explicit dependency management: All dependencies are clearly declared and injected, eliminating hidden requirements and making the code more predictable.
Extensible architecture: The centralized dependency service makes it easy to add cross-cutting concerns like caching, feature flags, or logging without modifying business logic.
Centralized configuration: All dependency management happens in one place, making it easier to maintain and modify the application's structure.

Testing Strategy

Testing is crucial in any architecture, and our layered approach makes it easier to test each component in isolation. We follow the classic testing pyramid:

System Tests

System tests verify that your entire application works together. They're like having a food critic visit your restaurant:

class TestFooAPI:
    @pytest.mark.anyio
    async def test_get_foo(
        self,
        async_client: AsyncClient,
    ) -> None:
        # given
        payload = { bar: "bar" }
        response = await async_client.post("v1/foo/", json=payload)
        foo_id = response.json()["id"]

        # when
        response = await async_client.get(f"v1/foo/{foo_id}")

        # then
        assert response.status_code == 200
        assert response.json() == {
            "id": str(job.id),
            "bar": "bar",
        }

Integration Tests

Integration tests verify that your layers work together correctly. They're like testing each station in the kitchen:

class TestSQLAFooDAO:
    @pytest.mark.anyio
    async def test_create(
        self,
        test_session: AsyncSession,
    ) -> None:
        # Given
        dao = SQLAlchemyFooDAO(test_session)
        foo_dto = FooDTO(bar="bar")

        # When
        result = await dao.create(foo_dto)

        # Then
        assert result.bar == "bar"
        assert result.created_at is not null

Unit Tests

Unit tests verify individual components in isolation. They're like testing each chef's skills:

@pytest.mark.anyio
class TestFooService:
    async def test_foo_happy(self) -> None:
        # given
        foo_dao = AsyncMock(spec=FooDAOInterface)
        foo_client = AsyncMock(spec=FooClientInterface)
        foo_service = FooService(
            foo_dao=foo_dao,
            foo_client=foo_client
        )
        foo_id = "foo_id"
        foo_dao.get_one.return_value = FooDTO(bar="bar", id="foo_id")
        foo_client.get.return_value = FooDTO(bar="bar", id="foo_id")

        # when
        response = await foo_service.get_one(foo_id=foo_id)

        # then
        assert response == FooDTO(bar="bar", id="foo_id")
        foo_dao.get_one.assert_called_once_with(foo_id)
        foo_client.get.assert_called_once_with(foo_id)

Trade-offs and Challenges

While layered architecture offers many benefits, it's important to understand the trade-offs and challenges you might face when implementing this pattern. These challenges shouldn't deter you, but rather help you make an informed decision about whether this architecture is right for your project.

Challenges

Initial development overhead: The layered architecture requires more boilerplate code upfront, but this investment pays off in maintainability. However, with proper Cursor or GitHub Copilot rules, you can significantly reduce this overhead as these AI tools can generate code that follows your established patterns.
Learning curve for new developers: Team members need time to understand the architecture patterns and dependency injection concepts. However, if your company uses this pattern consistently across projects, developers can quickly become productive as they move between projects, as the patterns remain familiar.
Potential over-engineering: For very small applications, this architecture might introduce unnecessary complexity and overhead. For example, if you're just exposing an ML model through a simple API endpoint, the full layered architecture would be overkill as there's minimal business logic to manage.

Benefits

Despite these challenges, the benefits of layered architecture often outweigh the initial investment. Let's explore the key advantages that make this architecture pattern valuable for many projects.

Enhanced maintainability: The clear separation of concerns and consistent patterns make the codebase easier to understand and modify.
Comprehensive testability: Each layer can be tested in isolation, making it easier to write and maintain tests.
Implementation flexibility: The interface-based design allows easy swapping of implementations without affecting other parts of the system. This is particularly useful for early validation, as you can implement null DAOs to have a fully functional end-to-end application without persistence, allowing you to validate business logic before committing to a specific database solution.
Consistent development patterns: The architecture enforces consistent patterns across the codebase, making it easier for teams to collaborate.

When to Use This Architecture

Choosing the right architecture for your project is crucial. While layered architecture is powerful, it's not a one-size-fits-all solution. Here are the scenarios where this architecture pattern truly shines and can provide the most value to your project.

Domain-focused applications: When your application has a clear, well-defined domain with complex business rules that need to be managed effectively. For multi-domain applications or large monoliths, other architectures that provide better inter-domain interfaces (like hexagonal architecture) should be preferred.
Growing projects: When you anticipate the application will grow over time and need a structure that can scale with it.
Complex business logic: When your application requires clear separation of concerns to manage intricate business rules and processes.
Multiple entry points: When your application needs to handle different types of requests (API, CLI, background jobs) while maintaining consistent business logic.
Long-term maintenance: When you need a codebase that will be maintained and extended over a long period.
Quality-focused development: When you need a structure that supports comprehensive testing and quality assurance.

Conclusion

While this architecture might seem like overkill for small projects, it pays dividends as your application grows. The initial investment in structure and patterns leads to a more maintainable, testable, and flexible codebase. The consistent patterns make it easier for new team members to onboard, and the clear separation of concerns makes it easier to modify and extend the system.

You can find all the code examples in our GitHub repository.

From Dependency Inversion to Dependency Injection in Python

Antonis Markoulis — Sun, 22 Sep 2024 12:16:20 +0000

Introduction

The Dependency Inversion Principle (DIP) and Dependency Injection (DI) are powerful concepts that can significantly improve the design of your Python code. In this post, we will explore these principles in detail, starting with a high-level overview and progressing to practical examples. By the end, you’ll have a deeper understanding of how to write modular, maintainable, and flexible Python code that adheres to these best practices.

Let’s start with a famous quote from Robert C. Martin:

“High-level modules should not depend on low-level modules; both should depend on abstractions. Abstractions should not depend on details; details should depend upon abstractions.”

This quote introduces us to the Dependency Inversion Principle (DIP). Let’s explore this principle further.

Dependency Inversion

What is Dependency Inversion?

The Dependency Inversion Principle (DIP) states that high-level modules should not depend on low-level modules but should instead rely on abstractions. This principle encourages us to decouple high-level business logic from the specific implementation details that support it. By doing so, we can swap out the underlying implementations without affecting the higher-level logic, making our code more flexible and maintainable.

Now, let’s visualize this with two examples:

Visualizing Dependency Inversion

In the following images, you can see the progression from a tightly coupled design to a design where components rely on interfaces and abstractions:

Tightly Coupled Design: In this scenario, components directly depend on one another, making the system rigid and hard to extend or modify without changing all the related components.
Decoupled Design with Dependency Inversion: By introducing abstractions (interfaces), we decouple the components, making the system more flexible. The high-level modules now rely on abstractions, making it easier to replace or update components without breaking the system.

Initial Design (No Abstraction)

Now that we understand the concept, let’s look at an initial example that does not follow the Dependency Inversion Principle. In the following code, the PizzaFanclass directly handles the process of ordering a pizza, and the main() function is tightly coupled to this specific implementation:

class PizzaFan:
    def order_margherita(self):
        print("Order margherita")

def main():
    pizza_fan = PizzaFan()
    pizza_fan.order_margherita()

This design lacks flexibility because if we want to use a different pizza provider, we would need to modify the main() function. This violates the Dependency Inversion Principle because the high-level module (main) depends on a low-level module (PizzaFan).

Improvement 1: Introducing Abstraction

To reduce the dependency on the low-level module, we can introduce an abstraction. Instead of depending directly on PizzaFan, we define an interface (AbstractPizza) and make both FancyPizza and GourmetPizza implement this interface. Now, the main() function depends on an abstraction (interface) rather than a concrete class.

from abc import ABC, abstractmethod

class AbstractPizza(ABC):
    @abstractmethod
    def order_margherita(self):
        pass

class FancyPizza(AbstractPizza):
    def order_margherita(self):
        print("Order margherita from FancyPizza")

class GourmetPizza(AbstractPizza):
    def order_margherita(self):
        print("Order margherita from GourmetPizza")

def main():
    fancy_pizza = FancyPizza()
    fancy_pizza.order_margherita()

    gourmet_pizza = GourmetPizza()
    gourmet_pizza.order_margherita()

By introducing the AbstractPizza interface, the main() function can now work with any pizza provider that implements this interface. This demonstrates how the Dependency Inversion Principle makes our code more flexible and adaptable to changes.

Improvement 2: Adding Flexibility with Parameters

Now that our main() function is decoupled from the specific pizza providers, let’s take it one step further by adding flexibility to handle different ingredients. In this example, we allow the user to specify ingredients, making the ordering process more dynamic.

from abc import ABC, abstractmethod
from typing import List

class AbstractPizza(ABC):
    @abstractmethod
    def order(self, ingredients: List[str]):
        pass

class FancyPizza(AbstractPizza):
    def order(self, ingredients: List[str]):
        if set(ingredients) == {"cheese", "tomatoes"}:
            print("Order margherita from FancyPizza")
        else:
            print(f"Order generic pizza from FancyPizza with {ingredients}")

class GourmetPizza(AbstractPizza):
    def order(self, ingredients: List[str]):
        print(f"Order pizza from GourmetPizza with {ingredients}")

def main():
    ingredients = ["cheese", "tomatoes"]
    fancy_pizza = FancyPizza()
    fancy_pizza.order(ingredients)

    gourmet_pizza = GourmetPizza()
    gourmet_pizza.order(ingredients)

Here, FancyPizza checks if the ingredients match a Margherita pizza, while GourmetPizza simply prints the order with the provided ingredients. The key benefit is that the main() function remains decoupled from specific implementations, allowing us to easily swap or extend the pizza services without modifying the client code.

Dependency injection

What is Dependency Injection?

Now that we’ve decoupled the high-level module (main()) from the specific implementation (PizzaFan) using Dependency Inversion, let’s see how we can inject those dependencies. Dependency Injection (DI) is a technique that provides an object with its dependencies rather than letting the object create those dependencies itself. This makes the system more flexible, as dependencies can be swapped out easily, and it promotes better separation of concerns.

Step 1: Dependency Injection with Manual Assembly

In this step, we demonstrate manual dependency injection, where dependencies are passed to the client at runtime.

We’ll model this with two levels of services: Electricity Providers and Banks. The Bank class depends on an ElectricityProvider, but the specific provider (e.g., Solar or Wind) is injected at runtime.

from abc import ABC, abstractmethod

# Abstract classes for electricity providers and banks
class ElectricityProvider(ABC):
    @abstractmethod
    def pay(self, amount: float):
        pass

class Bank(ABC):
    def __init__(self, provider: ElectricityProvider):
        self.provider = provider

    @abstractmethod
    def pay_electricity(self, amount: float):
        pass

# Concrete implementations
class SolarPower(ElectricityProvider):
    def pay(self, amount: float):
        print(f"Paid {amount} to Solar Power")

class WindEnergy(ElectricityProvider):
    def pay(self, amount: float):
        print(f"Paid {amount} to Wind Energy")

class GlobalBank(Bank):
    def pay_electricity(self, amount: float):
        print("Global Bank processing payment...")
        self.provider.pay(amount)

class NationalBank(Bank):
    def pay_electricity(self, amount: float):
        print("National Bank processing payment...")
        self.provider.pay(amount)

def main():
    solar_provider = SolarPower()  # Service
    global_bank = GlobalBank(solar_provider)  # Client with injected service
    global_bank.pay_electricity(100)

    wind_provider = WindEnergy()  # Service
    national_bank = NationalBank(wind_provider)  # Client with injected service
    national_bank.pay_electricity(200)

if __name__ == '__main__':
    main()

In this example, the Bank class is decoupled from the specific electricity provider by using constructor injection. The client code (in main()) determines which provider to inject at runtime.

Step 2: Dependency Injection with a Dependency Service

In this step, we introduce a Dependency Service that centralizes the logic of creating and providing dependencies. This removes the responsibility of assembling dependencies from the client code, making the system easier to maintain and extend.

from abc import ABC, abstractmethod

class DependencyService:
    @staticmethod
    def get_bank() -> 'Bank':
        return GlobalBank(WindEnergy())

# Abstract classes for electricity providers and banks
class ElectricityProvider(ABC):
    @abstractmethod
    def pay(self, amount: float):
        pass

class Bank(ABC):
    def __init__(self, provider: ElectricityProvider):
        self.provider = provider

    @abstractmethod
    def pay_electricity(self, amount: float):
        pass

# Concrete implementations
class SolarPower(ElectricityProvider):
    def pay(self, amount: float):
        print(f"Paid {amount} to Solar Power")

class WindEnergy(ElectricityProvider):
    def pay(self, amount: float):
        print(f"Paid {amount} to Wind Energy")

class GlobalBank(Bank):
    def pay_electricity(self, amount: float):
        print("Global Bank processing payment...")
        self.provider.pay(amount)

class NationalBank(Bank):
    def pay_electricity(self, amount: float):
        print("National Bank processing payment...")
        self.provider.pay(amount)

def main():
    global_bank = DependencyService.get_bank()
    global_bank.pay_electricity(100)

if __name__ == '__main__':
    main()

Here, the DependencyService abstracts the logic for creating the Bank and its dependencies. The client (main() function) no longer needs to know how the dependencies are created, simplifying the code.

Step 3: Using a Dependency Injection Library

Finally, we use the dependency-injector library to automate dependency injection. This formalizes the process by centralizing the configuration and management of dependencies in a container.

from abc import ABC, abstractmethod
from dependency_injector import containers, providers
from dependency_injector.wiring import Provide, inject

# Abstract classes
class ElectricityProvider(ABC):
    @abstractmethod
    def pay(self, amount: float):
        pass

class Bank(ABC):
    def __init__(self, provider: ElectricityProvider):
        self.provider = provider

    @abstractmethod
    def pay_electricity(self, amount: float):
        pass

# Concrete implementations
class SolarPower(ElectricityProvider):
    def pay(self, amount: float):
        print(f"Paid {amount} to Solar Power")

class WindEnergy(ElectricityProvider):
    def pay(self, amount: float):
        print(f"Paid {amount} to Wind Energy")

class GlobalBank(Bank):
    def pay_electricity(self, amount: float):
        print("Global bank processing payment...")
        self.provider.pay(amount)

class NationalBank(Bank):
    def pay_electricity(self, amount: float):
        print("National Bank processing payment...")
        self.provider.pay(amount)

# Container to manage dependencies
class Container(containers.DeclarativeContainer):
    electricity_provider = providers.Singleton(SolarPower)  # Default to SolarPower
    bank = providers.Factory(GlobalBank, provider=electricity_provider)

@inject
def main(bank: Bank = Provide[Container.bank]):
    bank.pay_electricity(150)

if __name__ == '__main__':
    container = Container()
    container.wire(modules=[__name__])
    main()

Explanation:

Container: This class is a central configuration for managing dependencies. It wires together services (SolarPower), and clients (GlobalBank) through providers.
Providers: Providers define how dependencies are created or managed. In this example:
- providers.Singleton: This provider ensures that a single instance of SolarPower is shared across the application.
- providers.Factory: This provider creates new instances of Eurobank, injecting the SolarPower provider each time.
@inject and Provide: These decorators and classes simplify injecting dependencies into functions, in this case, injecting the Bank into the main() function.

By using dependency-injector, you streamline dependency management and make your system easier to scale.

Conclusion

By understanding and applying both the Dependency Inversion Principle and Dependency Injection, we decouple our code, making it more flexible, adaptable, and maintainable. We’ve explored three different approaches to implementing DI: manual assembly, using a centralized Dependency Service, and leveraging a DI library. Each method enhances modularity and ease of maintenance, allowing us to focus on writing scalable, testable, and extensible Python code.

Try these patterns in your own projects, and feel free to share your feedback or questions in the comments!

Automating Python Library Releases Using GitHub Actions and Commitizen

Antonis Markoulis — Tue, 27 Aug 2024 15:55:34 +0000

Introduction

Maintaining a Python library can be challenging, especially when it comes to releasing new versions. The process can be time-consuming and error-prone if done manually. In this post, I’ll walk you through automating the release process using GitHub Actions and Commitizen. This approach ensures that your releases are consistent, adhere to semantic versioning (semver), and keep your changelog up to date—all while reducing manual intervention.

What is Semantic Versioning?

Semantic Versioning (semver) is a versioning scheme that uses three numbers in the format MAJOR.MINOR.PATCH. This scheme provides a clear and predictable way to communicate what has changed in each release:

MAJOR: Breaking changes—anything that is not backward-compatible.
MINOR: New features, but backward-compatible.
PATCH: Bug fixes—fully backward-compatible.

Semantic Versioning is crucial because it helps developers manage dependencies effectively. When you know that a new version of a library doesn’t introduce breaking changes (e.g., a minor or patch update), you can confidently update your dependencies without fear of your application breaking.

For more details on semver, you can check out semver.org.

Introduction to commitizen

Commitizen is a tool that standardizes commit messages and automates versioning and changelog creation. By enforcing a specific commit message format, Commitizen can determine the type of version bump required (major, minor, or patch) and generate a changelog automatically.

The commit message format follows this structure:

<commit-type>(<topic>): the commit message

Commit Types:
- feat: Indicates a new feature. This can lead to a minor version bump. If a commit includes a BREAKING CHANGE note, it results in a major version bump.
- fix: Indicates a bug fix and leads to a patch version bump.
- chore, ci, and others: These do not trigger a version bump.

For example:

feat(parser): add support for parsing new file formats

fix(api): handle null values in the response

feat(api): change response of me endpoint

BREAKING CHANGE: 

changes the API signature of the parser function

In this example, the BREAKING CHANGE note within a feat commit would trigger a major version bump. This consistency ensures that your version numbers communicate the right level of change, which is critical for users relying on your library.

Configuring Commitizen

To integrate Commitizen with your Python project, you need to configure it in your pyproject.toml file. Below is the configuration you'll need to add:

[tool.commitizen]
name = "cz_conventional_commits"
version = "0.1.0"
tag_format = "v$version"
version_files = [
    "pyproject.toml:version",
]
update_changelog_on_bump = true

Explanation:

name: Specifies the commit message convention to use. We’re using the conventional commits format.
version: The current version of your project. You should start with "0.1.0" or whatever your initial version is.
tag_format: Defines how tags are formatted, with v$version being the typical format (v1.0.0, v1.1.0, etc.).
version_files: Lists the files where version numbers are tracked. This setup ensures that the version number in pyproject.toml is updated automatically.
update_changelog_on_bump: Automatically updates the CHANGELOG.md file whenever a version bump occurs.

Why Automate the Release Process?

Manually managing releases can be tedious and prone to errors, especially as your project grows. Automation brings several key benefits:

Consistency: Ensures that version bumps and changelogs are handled the same way every time.
Efficiency: Saves time by reducing the manual steps involved in releasing a new version.
Accuracy: Minimizes human error, such as forgetting to update the changelog or incorrectly bumping the version.

Overview: The Automated Release Process

To give you a clear picture of how the automation works, here’s a high-level overview:

On Merge to Main: When a pull request (PR) is merged into the main branch, the workflow checks the commit messages, decides whether a version bump is needed, updates the changelog, and tags the release if necessary.
On Tag Creation: When a tag is pushed (indicating a new release), the workflow publishes the new version to PyPI and creates a GitHub release with the corresponding changelog.

Splitting the Workflow: Merge vs Tag Events

For simplicity and clarity, we’ll split the automation into two workflows:

Merge to Main Workflow
On Tag Creation Workflow

Workflow 1: On Merge to Main

This workflow handles the logic of detecting changes and bumping the version:

name: Merge to Main

on:
  push:
    branches:
      - "main"

concurrency:
  group: main
  cancel-in-progress: true

jobs:
  bump:
    if: "!startsWith(github.event.head_commit.message, 'bump:')"
    runs-on: ubuntu-latest
    steps:
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.10"
      - name: Checkout Repository
        uses: actions/checkout@v4
        with:
          token: ${{ secrets.PERSONAL_ACCESS_TOKEN }}
          fetch-depth: 0
      - name: Create bump and changelog
        uses: commitizen-tools/commitizen-action@0.21.0
        with:
          github_token: ${{ secrets.PERSONAL_ACCESS_TOKEN }}
          branch: main

Explanation:

Trigger: This workflow is triggered when a new commit is pushed to the main branch.
Concurrency: The concurrency setting ensures that only one instance of the workflow runs at a time for the main branch, preventing race conditions and duplicate version bumps.
bump job: It checks whether the commit message starts with ‘bump:’, which indicates an automated commit from the previous release. If not, Commitizen determines the necessary version bump based on the commit messages, updates the CHANGELOG.md, and creates a new tag.

Workflow 2: On Tag Creation

This workflow is triggered when a tag is pushed, and it handles the release process:

name: On Tag Creation

on:
  push:
    tags:
      - 'v*'

concurrency:
  group: tag-release-${{ github.ref }}
  cancel-in-progress: true

jobs:
  detect-release-parameters:
    runs-on: ubuntu-latest
    outputs:
      notes: ${{ steps.generate_notes.outputs.notes }}
    steps:
      - name: Setup Python
        uses: actions/setup-python@v5
      - name: Checkout Repository
        uses: actions/checkout@v4
        with:
          fetch-depth: 0
      - name: Get release notes
        id: generate_notes
        uses: anmarkoulis/commitizen-changelog-reader@v1.2.0
        with:
          tag_name: ${{ github.ref }}
          changelog: CHANGELOG.md

  release:
    runs-on: ubuntu-20.04
    needs: detect-release-parameters
    steps:
      - name: Checkout repo
        uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.10"
      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install poetry
      - name: Configure pypi token
        run: |
          poetry config pypi-token.pypi ${{ secrets.PYPI_TOKEN }}
      - name: Build and publish package
        run: |
          poetry publish --build

  release-github:
    runs-on: ubuntu-latest
    needs: [release, detect-release-parameters]
    steps:
      - name: Checkout Repository
        uses: actions/checkout@v4
      - name: Create Release Notes File
        run: |
          echo "${{ join(fromJson(needs.detect-release-parameters.outputs.notes).notes, '') }}" > release_notes.txt
      - name: Create GitHub Release
        env:
          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
          VERSION: ${{ github.ref_name }}
        run: |
          gh release create ${{ github.ref }} \
          --title "Release $VERSION" \
          --notes-file "release_notes.txt"

Explanation:

Trigger: This workflow is triggered by a tag push (e.g., v1.2.0).
Concurrency: The concurrency setting ensures that only one instance of the workflow runs for each tag, preventing issues like multiple release attempts for the same version.
detect-release-parameters job: Extracts the changelog notes for the release.
release job: Builds the package and publishes it to PyPI using Poetry.
release-github job: Creates a new GitHub release with the generated release notes.

Setting Up the Personal Access Token

To ensure the workflows can perform actions like creating commits and tagging releases, you’ll need to set up a Personal Access Token (PAT) in your GitHub repository:

Go to your repository on GitHub.
Navigate to Settings > Secrets and variables > Actions.
Click on New repository secret.
Add a secret with the name PERSONAL_ACCESS_TOKEN and paste your PAT in the value field.

This token is crucial because it allows the workflow to push changes (like the updated changelog and version bump) back to the repository.

Generated CHANGELOG.md Example

After running the workflows, a CHANGELOG.md file will be generated and updated automatically. Here’s an example of what it might look like:

## v2.0.0 (2021-03-31)

### Feat

- **api**: change response of me endpoint

## v1.0.1 (2021-03-30)

### Fix

- **api**: handle null values in the response

## v1.0.0 (2021-03-30)

### Feat

- **parser**: add support for parsing new file formats

This CHANGELOG.md is automatically updated by Commitizen each time a new version is released. It categorizes changes into different sections (e.g., Feat, Fix), making it easy for users and developers to see what's new in each version.

Common Issues and Troubleshooting

Finally, here’s what a GitHub release looks like after being created by the workflow:

Incorrect Token Permissions: If the workflow fails due to permission errors, ensure that the PAT has the necessary scopes (e.g., repo, workflow).
Commitizen Parsing Issues: If Commitizen fails to parse commit messages, double-check the commit format and ensure it's consistent with the expected format.
Bump Commit Conflicts: If conflicts arise when the bump commit tries to merge into the main branch, you might need to manually resolve the conflicts or adjust your workflow to handle them.
Concurrent Executions: Without proper concurrency control, multiple commits or tags being processed simultaneously can lead to issues like duplicate version bumps or race conditions. This can result in multiple commits with the same version or incomplete releases. To avoid this, we’ve added concurrency settings to both workflows to ensure only one instance runs at a time for each branch or tag.

Conclusion and Next Steps

Automating the release process of your Python library with GitHub Actions and Commitizen not only saves time but also ensures consistency and reduces human errors. With this setup, you can focus more on developing new features and less on the repetitive tasks of managing releases.

As a next step, consider extending your CI/CD pipeline to include automated testing, code quality checks, or even security scans. This would further enhance the robustness of your release process.

Call to Action

If you found this post helpful, please feel free to share it with others who might benefit. I’d love to hear your thoughts or any questions you might have in the comments below. Have you implemented similar workflows in your projects? Share your experiences!

DEV Community: Antonis Markoulis

How I Learned to Stop Worrying and Love Raw Events, Event Sourcing & CQRS with FastAPI and Celery

The Problem: When Systems Forget

Why Traditional Systems Fail Us

The Solution: Events as the Source of Truth

The Event Store: Your System's Memory

CQRS: Separating What We Do from What We Read

Commands: Expressing Intent

Queries: Optimized for Reading

Aggregates: Where Business Logic Lives

Projections: Building Read Models from Events

From Theory to Practice: Building It with Python

The Complete Picture

FastAPI: Command & Query Interface

Command Handlers: Orchestration Logic

Aggregates: Pure Python Business Logic

Event Handler: Celery Integration

Celery Tasks: Event Processing

Projections: Event-Driven Read Models

The Reality Check: What Actually Happens in Production

The Eventual Consistency UX Problem

Performance with Snapshots

Debugging Superpowers: Testing Business Logic

Summary: Key Takeaways and Strategic Insights

🚀 Start Simple: The Power of Incremental Adoption

⚠️ When NOT to use Event Sourcing: The Honest Assessment

🎯 What you gain: The Strategic Advantages

Conclusion

Resources

Layered Architecture & Dependency Injection: A Recipe for Clean and Testable FastAPI Code

Introduction

Understanding Layered Architecture

Layer 1: Presentation Layer

Layer 2: Service Layer (Business)

Layer 3: Persistence Layer (DAOs)

DTOs: The Glue Between Layers

Dependency Injection with Dependency Service

Testing Strategy

System Tests

Integration Tests

Unit Tests

Trade-offs and Challenges

Challenges

Benefits

When to Use This Architecture

Conclusion

From Dependency Inversion to Dependency Injection in Python

Introduction

Dependency Inversion

What is Dependency Inversion?

Visualizing Dependency Inversion

Initial Design (No Abstraction)

Improvement 1: Introducing Abstraction

Improvement 2: Adding Flexibility with Parameters

Dependency injection

What is Dependency Injection?

Step 1: Dependency Injection with Manual Assembly

Step 2: Dependency Injection with a Dependency Service

Step 3: Using a Dependency Injection Library

Explanation:

Conclusion

Automating Python Library Releases Using GitHub Actions and Commitizen

Introduction

What is Semantic Versioning?

Introduction to commitizen

Configuring Commitizen

Why Automate the Release Process?

Overview: The Automated Release Process

Splitting the Workflow: Merge vs Tag Events

Workflow 1: On Merge to Main

Workflow 2: On Tag Creation

Setting Up the Personal Access Token

Generated CHANGELOG.md Example

Common Issues and Troubleshooting

Conclusion and Next Steps

Call to Action

References and Further Reading