Forem: Szymon

Exploring Different Schools of Unit Testing in Python

Szymon — Fri, 26 Jan 2024 23:00:00 +0000

Intro

Unit testing is a crucial aspect of software development that ensures the reliability and correctness of individual components within a system.
Over the years, different schools of thought have emerged, each with its own principles and practices.
In this blog post, I will show you two schools of unit testing: the Classical and London schools.
Through Python code examples, I will illustrate the key concepts associated with each approach.

What is a unit?

When it comes to unit tests we want to test a certain unit, but what is this unit?
Is it one class? Multiple classes? Or maybe the whole module?

This is the key point of a discussion about Classical and London School of unit testing.

In the Classical School, a "unit" typically refers to a class or a set of classes working together to perform a specific functionality.
It is oriented toward behaviors.
The emphasis is on testing the state changes of these classes after invoking certain actions.
This approach assumes limited usage of mock or any other kind of test doubles.
They should be used only for shared dependencies or dependencies that change the state.
By "shared" dependency, I mean a dependency used by different components in our system.

In the London School, a "unit" is often defined as an individual class.
The focus shifts towards testing the interactions and collaborations between objects rather than the internal state or behaviors.
Mock objects play a significant role in isolating these interactions during testing.
This approach assumes that all dependencies internal/external/shared should be replaced by mock or any other kind of test doubles.

Note:
If you're interested in exploring types of test doubles further, such as mocks, stubs, fakes, and others please let me know in the comment section below.
I'm eager to create another article about it.

Historical Background

The Classical school (aka. Chicago or Detroit), associated with the "state-based" or "classical" approach, originated with Kent Beck, a key figure in the Agile and XP movements.
The focus is on testing based on the observed state of an object after an action, influenced by Beck's work on the "xUnit" testing framework.

The London school, known as the "interaction-based" or "mockist" approach, gained prominence with developers like Steve Freeman and Nat Pryce.
They introduced the concept of using mock objects to test interactions between components, emphasizing interactions over state.

Differences between Classical and London Schools

Classical School

Focus: Emphasizes testing based on the observed behaviors/state of objects.
Approach: Tests often involve arranging objects, performing actions, and asserting the expected state.
Pros:
- Intuitive for straightforward state-based testing.
- Simple setup and verification.
Cons:
- Test is more coupled to implementation details.
- Tests can be brittle to changes in the internal structure.

London School

Focus: Prioritizes testing the interactions between objects.
Approach: Utilizes mock objects to verify the expected behavior and interactions.
Pros:
- Encourages loose coupling and better separation of concerns.
- More resilient to changes in internal implementations.
Cons:
- Learning curve for those new to mocking concepts.
- Tests might be less readable due to increased setup.

Code Examples

Consider an example of transaction processing.
The code example is trivial purposely to illustrate the difference between these two schools.

from src.classical_vs_london.account import AccountType
from src.classical_vs_london.database import AccountLimit, InMemoryAccountDatabase
from src.classical_vs_london.event_bus import EventBus, TransactionProcessed


class TransactionError(Exception):
    pass


class TransactionValidator:
    def __init__(self, account_limit: AccountLimit) -> None:
        self._account_limit = account_limit

    def validator_for(self, account_id: int, amount: float) -> None:
        if amount < 0:
            raise TransactionError()

        limit = self._account_limit.get_for(account_id)

        if amount > limit:
            raise TransactionError()


class TransactionProcessor:
    def __init__(
        self,
        database: InMemoryAccountDatabase,
        validator: TransactionValidator,
        event_bus: EventBus
    ) -> None:
        self._database = database
        self._validator = validator
        self._event_bus = event_bus

    def process_transaction(self, account_id: int, amount: float) -> None:
        self._validator.validator_for(account_id, amount)

        account = self._database.get(account_id)

        account.balance -= amount
        account.transaction_counter += 1

        if account.transaction_counter > 10:
            account.type = AccountType.VIP

        self._database.save(account)
        self._event_bus.dispatch(TransactionProcessed(account.account_id))

All code available here

Tests:

def test_can_process_transaction_classical() -> None:
    # given
    account = Account(1, 100.0)
    database = InMemoryAccountDatabase()
    account_limit = InMemoryAccountLimit()
    event_bus = Mock(spec_set=EventBus)
    processor = TransactionProcessor(database, TransactionValidator(account_limit), event_bus)

    # and
    database.save(account)
    account_limit.add_for(account.account_id, 100.0)

    # when
    processor.process_transaction(account.account_id, 50.0)

    # then
    account = database.get(account.account_id)
    assert account.transaction_counter == 1
    assert account.balance == 50.0
    event_bus.dispatch.assert_called_once_with(TransactionProcessed(account.account_id))


def test_can_process_transaction_london() -> None:
    # given
    account = Account(1, 100.0)
    database = Mock(spec_set=AccountDatabase)
    database.get.return_value = account
    validator = Mock(spec_set=TransactionValidator)
    event_bus = Mock(spec_set=EventBus)
    processor = TransactionProcessor(database, validator, event_bus)

    # when
    processor.process_transaction(account.account_id, 50.0)

    # then
    assert account.balance == 50.0
    validator.validator_for.assert_called_once_with(account.account_id, 50.0)
    database.save.assert_called_once_with(account)
    event_bus.dispatch.assert_called_once_with(TransactionProcessed(account.account_id))

All tests available here

As you can see I used mock for EventBus inside test_can_process_transaction_classical.
EventBus is a shared dependency, it will be used by other components of my system.
From Classical School point of view it is fine to mock such dependencies.
The other two AccountDatabase and TransactionValidator are private dependencies of my TransactionProcessor class.
They will not be used by other components of my system. That's why I don't want to mock them.

On the other hand, London School tells us to mock every dependency of the class we are testing.
It does not matter if it is private or shared.

Which one to choose?

It's important to highlight that there is no "one and only" approach.
The choice depends on project requirements, and a balance may be necessary.

Each of the schools has its advantages.

The Classical School - tests better reflect the actual behavior of the system.

The London Schools - tests are better isolated, and writing mocks may be easier than creating a complicated object.

I'm not the purist in this case but in my opinion, I favor Classical School.
I've seen too many mocks overuse.
Mock that returns a mock that returns a mock.

On the other hand, I can imagine a situation where using London School will be more appropriate.
For example CRUD logic.

Summary

In the realm of Python unit testing, the Classical and London schools offer distinct approaches, each with its own set of advantages and drawbacks.
Whether you prefer the state-based simplicity of Classical testing or the interaction-based flexibility of the London school, the key is to choose an approach that aligns with your project's needs.
As you delve into the world of unit testing, consider the historical context and the practical examples provided to make informed decisions for robust and effective testing in your Python projects.

How to implement DDD Entities in Python

Szymon — Thu, 21 Dec 2023 23:00:00 +0000

Intro

Some time ago I created a blog post about "Value Objects".
I want to continue the series about DDD tactic building blocks and in
today's blog post I want to show you another building block - entity.

What is an entity?

Entity - an identifiable object that contains and encapsulates business logic

This is the simplest and shortest definition I can think of regarding entity.

If we would want to take a deeper look at what it means. It means that the entity is
a building block that is used to describe the structure of our model. By model, I mean the technical
solution/implementation for a problem that exists in the real world/business.

It means that an entity is something that:

changes over time (it is mutable)
has life cycle
has an identity that is different from the value of attributes that describe it

For example, if we want to model a citizen of some country, we could define an entity Citizen.
This citizen may have some attributes like: first name, last name, height, marital status, or eye color.
Even if our citizen will decide to change their first name, last name, or marital status.
It is still the same citizen.

To be able to identify the citizen we have to endow him identity.
This identity may be the personal identity number.

Our citizen obtains their personal identity number sometime after birth.
Years later, the first name, last name, or height may be different.
Even the color of the eyes may change, and he may have some diseases.
It is still the same citizen. His attributes have changed over time but his identity is the same.

For these types of things that we find during our domain modeling, we will want
to describe/define them using entities.

Implementation

Entity is a mutable object, we have to make sure that the change will be done safely.
What does safely in the context mean? It means that we want to perform the change
ensuring all business rules.

Our entity exposes methods (behaviors) named in a domain language that encapsulates all the implementation details like state checks, validation, etc.
No public getters and setters.
We don't want to allow anyone outside our module to manipulate any attribute of our entity.

In simple words, the procedure is as follows:

public method names using domain language
perform business logic, check all the rules, invariants, etc
change the state of the entity

Let me show you the code example from one of my projects. I will obfuscate the business domain to not break the NDA.
Let's imagine that you have some sort of analytical report.
A client can suppress some of the items from the analytical report.
Based on it, you want to show some metrics to the client.
By default, if there are no suppressed data items all metrics are equal to 0.
The client may suppress up to 10 data items.
Once he requests the metrics taking into account the suppressed data items the recalculation needs to happen.
The client may disable the suppression whenever he wants.

from __future__ import annotations
from dataclasses import dataclass
from uuid import UUID
from typing import List
from enum import Enum, unique
from decimal import Decimal
import random

@dataclass(frozen=True)
class SuppressionId:
    value: UUID


@dataclass(frozen=True)
class DataItem:
    name: str
    value: Decimal

    def __post_init__(self) -> None:
        if " " in self.name:
            raise ValueError(
                f"The provided data item name `{self.name}` is not correct."
            )


@unique
class SuppressionStatus(Enum):
    ACTIVATED = "activated"
    DISABLED = "disabled"


@dataclass(frozen=True)
class Metrics:
    metric_a: Decimal = Decimal(0)
    metric_b: Decimal = Decimal(0)
    metric_c: Decimal = Decimal(0)
    metric_d: Decimal = Decimal(0)


class SuppressionError(Exception):
    pass


class Suppression:
    DATA_ITEMS_SUPPRESSION_LIMIT = 10

    def __init__(self, id: SuppressionId, owner: str) -> None:
        self._id = id
        self._owner = owner
        self._data_items: List[DataItem] = []
        self._metrics = Metrics()
        self._status = SuppressionStatus.ACTIVATED

    @property
    def id(self) -> SuppressionId:
        return self._id

    @property
    def data_items(self) -> List[DataItem]:
        return self._data_items

    @property
    def metrics(self) -> Metrics:
        return self._metrics

    @property
    def is_active(self) -> bool:
        return self._status == SuppressionStatus.ACTIVATED

    def disable(self) -> None:
        self._status = SuppressionStatus.DISABLED

    def activate(self) -> None:
        self._status = SuppressionStatus.ACTIVATED

    def suppress_data_item(self, data_item: DataItem) -> None:
        if len(self._data_items) == self.DATA_ITEMS_SUPPRESSION_LIMIT:
            raise SuppressionError(
                f"`You can suppress only up to `{self.DATA_ITEMS_SUPPRESSION_LIMIT}` data items"
            )

        if self._status == SuppressionStatus.DISABLED:
            raise SuppressionError(
                "Can not perform any changes on a `Suppression` that is disabled"
            )

        if data_item in self._data_items:
            return

        self._data_items.append(data_item)

    def restore_data_item(self, data_item: DataItem) -> None:
        if self._status == SuppressionStatus.DISABLED:
            raise SuppressionError(
                "Can not perform any changes on a `Suppression` that is disabled"
            )

        try:
            self._data_items.remove(data_item)
        except ValueError:
            return

    def recalculate_metrics(self) -> Metrics:
        if self._status == SuppressionStatus.DISABLED:
            raise SuppressionError(
                "Can not recalculate metrics on a disabled `Suppression`"
            )

        if len(self._data_items) == 0:
            return Metrics()

        return self._recalculate_metrics()

    def _recalculate_metrics(self) -> Metrics:
        # math operations needed for metrics recalculation
        # skipping this part of the code as it is implementation details
        # and does not bring any value to the example I want to show you
        return Metrics(
            Decimal(random.randrange(0, 1000)),
            Decimal(random.randrange(0, 1000)),
            Decimal(random.randrange(0, 1000)),
            Decimal(random.randrange(0, 1000))
        )

God entity

One of the problems I consider most dangerous is the "God entity class".

God class, or god object is a class that does too much.

Have you ever seen things like class User or class Order in your project?

Where literally everything is inside it? User details, user permissions, user roles, user activities.
When it comes to Order - order details, order items, invoice details, payment details, statuses, and sub-statuses.

Let's be fair, modeling an entity is not about it. We should consider it as an anti-pattern.
Trust me from a long-time perspective maintaining such software is almost impossible. The complexity grows exponentially

Refactor it as soon as possible.
I realize that refactoring such a class is not easy.
Let me know if you would like to see another blog post that will be exactly about it.
I can show you some patterns/heuristics on how to deal with such problems.

Summary

In this exploration of Domain-Driven Design (DDD), we delved into the foundational concept of entities, crucial components that encapsulate the core business logic of an application.

Entities stand as the bedrock upon which rich domain models are built.
They serve as more than just data structures, evolving into active participants in the business processes, embodying the true essence of DDD principles.

That’s all Folks!
Thank you for staying till the end.
We lay the groundwork for future discussions.
I plan to write about other building blocks too like repositories, aggregated, and factories.

Stay tuned and happy coding!

Enhancing AWS Athena Efficiency - Building a Python Athena Client

Szymon — Thu, 02 Nov 2023 23:00:00 +0000

Intro

So, a while back, as part of my work-related escapades,
I was assigned to a ticket that was about running a few quite complex SQL queries in AWS Athena.

So, I moved the ticket to "in progress" and started working on it.
When I opened boto3 Athena documentation
my first thought was "Ohh, it will be a tough journey" 😄

What is AWS Athena?

For those unfamiliar, Athena is Amazon's serverless query service that lets you analyze data in S3 using standard SQL.
Athena offers the power of SQL queries on data stored in S3.

If you have not heard about AWS Athena, I encourage you to take a look at this service.
You can read more about it here.

The problem with Athena

Managing Athena queries is a formidable task especially when you're dealing with a heap of queries (that was my case).

Those who have ever worked with AWS Athena know what I'm talking about 😉.

In a nutshell, the whole process looks as follows:

you run the query
as a result of the above, you get the query execution id
if the query state value is "successful" you can fetch the query result
if it is succeeded you can fetch the query result

About the above, let me highlight the most problematic things, at least from my perspective:

you have to monitor query states, you have to ensure they are completed successfully
you have to monitor query execution time (it demands constant vigilance)
you have to handle result pagination (by default Athena returns only the first 1000 records)
you have to take care of error management
Athena returns describe <table> statement result as plain text

I'm highlighting it because when I saw the ticket at the beginning
I was almost sure that it would be as simple as running any SQL statement in some SQL database.

That's why I decided to write a custom Python Athena client, that will solve all of these problems.

The goal was simple: a nice and stable interface that would hide all the Athena’s meanders.

Implementation

The requirements are as follows:

I want to be able to execute a single query in AWS Athena
I want to be able to execute multiple queries in AWS Athena at once (as it is a serverless service, and it scales very well)
I want to have query results as a list of dicts
I don't want to cast the column values, everything is returned as a string even if a column type is, for example, timestamp, or integer

Check the code below 🚀

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from enum import Enum, unique
from logging import Logger
from typing import Any, Dict, List, Optional

from botocore.exceptions import ClientError
from mypy_boto3_athena.client import AthenaClient as AthenaSdkClient
from typing_extensions import TypeAlias

AthenaQueryResult: TypeAlias = List[Dict[str, Any]]


class QueryExecutionFailed(Exception):
    pass


@unique
class AthenaQueryStatus(Enum):
    QUEUED = "QUEUED"
    RUNNING = "RUNNING"
    SUCCEEDED = "SUCCEEDED"
    FAILED = "FAILED"
    CANCELLED = "CANCELLED"

    def __str__(self) -> str:
        return str(self.value)


@dataclass(frozen=True)
class AthenaClientConfig:
    s3_output_location: str
    query_waiting_delay: float = 0.25  # second
    timeout: int = 600  # seconds
    maximum_workers_number: Optional[int] = None


@dataclass
class AthenaQuery:
    database_name: str
    sql_statement: str
    result: AthenaQueryResult = field(default_factory=lambda: [])
    status: AthenaQueryStatus = AthenaQueryStatus.QUEUED

    def __post_init__(self) -> None:
        if not self.database_name:
            raise ValueError("`database_name` may not be empty!")

    @property
    def is_successful(self) -> bool:
        return bool(self)

    def __bool__(self) -> bool:
        return self.status == AthenaQueryStatus.SUCCEEDED


class AthenaClient:
    def __init__(
        self,
        sdk: AthenaSdkClient,
        config: AthenaClientConfig,
        logger: Logger,
    ) -> None:
        self._sdk = sdk
        self._config = config
        self._logger = logger

    def execute(self, query: AthenaQuery) -> None:
        self._logger.info(
            "Running query `%s` on `%s`",
            query.sql_statement,
            query.database_name,
        )
        try:
            query.result = self._execute(query.sql_statement, query.database_name)
            query.status = AthenaQueryStatus.SUCCEEDED
        except QueryExecutionFailed:
            query.status = AthenaQueryStatus.FAILED

    def execute_many(self, *queries: AthenaQuery) -> None:
        self._logger.info(
            "Running `%s` queries in parallel",
            len(queries),
        )

        max_workers = self._config.maximum_workers_number or len(queries)
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {}
            for query in queries:
                future = executor.submit(self._execute, query.sql_statement, query.database_name)
                futures[future] = query

            for future in as_completed(futures):
                query = futures[future]

                if future.exception():
                    query.status = AthenaQueryStatus.FAILED
                else:
                    query.result = future.result()
                    query.status = AthenaQueryStatus.SUCCEEDED

    def _execute(self, sql_statement: str, database_name: str) -> AthenaQueryResult:
        response = self._sdk.start_query_execution(
            QueryString=sql_statement,
            QueryExecutionContext={"Database": database_name},
            ResultConfiguration={"OutputLocation": self._config.s3_output_location},
        )
        query_execution_id = response["QueryExecutionId"]
        self._logger.info("query_execution_id = `%s`", query_execution_id)
        self._wait_for_query_results(query_execution_id)
        return self._get_query_results(query_execution_id, is_describe_query="DESCRIBE" in sql_statement)

    def _wait_for_query_results(self, query_execution_id: str) -> None:
        self._logger.info("Waiting for query_execution_id = `%s` to finish", query_execution_id)

        start_time = time.time()
        while True:
            try:
                response = self._sdk.get_query_execution(QueryExecutionId=query_execution_id)
                state = response["QueryExecution"]["Status"]["State"]
                valid_statuses = [
                    str(AthenaQueryStatus.SUCCEEDED),
                    str(AthenaQueryStatus.FAILED),
                    str(AthenaQueryStatus.CANCELLED),
                ]
                if state in valid_statuses:
                    return

                if time.time() - start_time > self._config.timeout:
                    self._logger.error(
                        "Query `%s` execution reached out the maximum timeout value (%s).",
                        query_execution_id,
                        f"{self._config.timeout}s",
                    )
                    raise QueryExecutionFailed(f"Query `{query_execution_id}` execution timeout has been reached")

                self._logger.info(
                    "Waiting %s before next check on query status.",
                    f"{self._config.query_waiting_delay}s",
                )
                time.sleep(self._config.query_waiting_delay)
            except ClientError as error:
                self._logger.error(
                    "An unexpected error occurred. Error = %s",
                    str(error),
                )
                raise QueryExecutionFailed("An unexpected error occurred during query execution") from error

    def _get_query_results(self, query_execution_id: str, is_describe_query: bool = False) -> AthenaQueryResult:
        try:
            self._logger.info("Getting query results for %s", query_execution_id)
            response = self._sdk.get_query_execution(QueryExecutionId=query_execution_id)
            state = response["QueryExecution"]["Status"]["State"]

            if state != str(AthenaQueryStatus.SUCCEEDED):
                self._logger.info(
                    "Query `%s` has finished with a state = %s",
                    query_execution_id,
                    state,
                )
                raise QueryExecutionFailed(f"Query `{query_execution_id}` has finished with a not-success state.")

            results_paginator = self._sdk.get_paginator("get_query_results")
            results_iterator = results_paginator.paginate(
                QueryExecutionId=query_execution_id,
                PaginationConfig={"PageSize": 1000},
            )
            results = []
            columns = None
            for results_page in results_iterator:
                if not columns:
                    columns = [col["Label"] for col in results_page["ResultSet"]["ResultSetMetadata"]["ColumnInfo"]]

                for row in results_page["ResultSet"]["Rows"]:
                    values = [item.get("VarCharValue") for item in row["Data"]]
                    results.append(dict(zip(columns, values)))

            # remove items that contains column name as the item value,
            # Athena always returns column names as the first
            # item inside ["ResultSet"]["Rows"], because we are using paginator, this will happen on every page.
            filtered_list = [
                item for item in results if not any(column in item.values() for column in columns)  # type: ignore
            ]

            if is_describe_query:
                describe_query_response = []
                for row in filtered_list:  # type: ignore
                    cleaned_column_name = row["col_name"].split("\t")[0].strip()  # type: ignore
                    cleaned_column_value = row["col_name"].split("\t")[1].strip()  # type: ignore

                    if cleaned_column_name and not cleaned_column_name.startswith("#"):
                        describe_query_response.append({cleaned_column_name: cleaned_column_value})

                return describe_query_response

            return filtered_list
        except ClientError as error:
            self._logger.error(
                "An unexpected error occurred. Error = %s",
                str(error),
            )
            raise QueryExecutionFailed("An unexpected error occurred during query execution") from error

execute function is responsible for executing a single query in Athena. The query is represented by the AthenaQuery object.
If the execution is successful, it sets the query.status to AthenaQueryStatus.SUCCEEDED and stores the query results in query.result.
If the query execution fails, it sets the query.status to AthenaQueryStatus.FAILED.

execute_many function is responsible for executing multiple queries in Athena concurrently.
It determines the number of threads to use based on the maximum_workers_number configuration or the total number of queries if no maximum_workers_number is specified.
It uses a ThreadPoolExecutor to manage the concurrent execution of queries.
For each query, it submits a task to the thread pool.
As tasks are completed, it updates the status and result of each query based on the task's outcome.
If a task fails, the query's status is set to AthenaQueryStatus.FAILED, and if it succeeds, the status is set to AthenaQueryStatus.SUCCEEDED, and the result is stored in query.result.

These functions execute queries in Athena either individually or in parallel,
providing you with the flexibility to process multiple queries concurrently while managing their statuses and results.
The use of a thread pool ensures that tasks are executed safely in separate threads without interfering with each other.

If you want to check the tests, they are available on my GitHub here 😉

Usage

I want to show you a real case from my project where I applied my Athena client.
I will anonymize the business domain to not break the NDA but the logic stays the same.

Let's imagine that you have a list of string values, and for these values, you have to generate all combinations.
Then for each item from the combination list, you want to get the number of records that contain this combination item value as a column value.
As a result, you want to return a single integer that is a total count of all records for each combination value.

Let me show you the code:

import itertools
from dataclasses import dataclass
from kink import inject
from typing import List


@dataclass
class GetTotalNumberOfRecords:
    values: List[str]

@inject
class GetNumberOfRecordsQueryHandler(
    QueryHandler[GetTotalNumberOfRecords, int]
):
    def __init__(self, athena_client: AthenaClient) -> None:
        self._athena_client = athena_client

    def __call__(self, query: GetTotalNumberOfRecords) -> int:
        total_number_of_records = 0

        all_combinations = []
        for i, _ in enumerate(query.values, start=1):
            for item in itertools.combinations(query.values, i):
                if len(item) == 1:
                    all_combinations.append(item[0])
                else:
                    all_combinations.append(
                        " | ".join(sorted(item))
                    )

        athena_queries: List[AthenaQuery] = []

        for item in sorted(all_combinations):
            athena_query = AthenaQuery(
                database_name="my_database",
                sql_statement=f"""SELECT DISTINCT
                COUNT(*) OVER () AS number_of_records
                FROM my_table
                WHERE "my_column"='{item}'
                GROUP BY "my_other_column"
                """,
            )
            athena_queries.append(athena_query)

        self._athena_client.execute_many(*athena_queries)

        for athena_query in athena_queries:
            if not athena_query.is_successful:
                continue

            to_add = int(athena_query.result[0]["number_of_records"])
            total_number_of_records += to_add

        return total_number_of_records

What do you think? Now let's imagine how hard it would be if we had to base on a pure boto3 Athena client.

In the above scenario, all we have to care about is our business/query logic. No knowledge of how Athena works under the hood.

Summary

That is it 😉 That was my journey.

I’ve presented you with a custom Python Athena client that streamlines and elevates the querying experience.

Feel free to use it!
I'd love to hear your thoughts on this implementation.
Do you find it beneficial?
What would you add or change to make it even more powerful?
After taking a peek at the code, how do you feel? Can you envision using it in your projects?

I'm eager to engage with you, our fellow tech enthusiasts, to make my AthenaClient even better.

Happy coding!

Don't put your business logic in the controllers

Szymon — Mon, 02 Oct 2023 22:00:00 +0000

Intro

In the world of WEB API development with Python, maintaining clean and maintainable code is paramount.
One common pitfall to avoid is mixing your business logic with your controllers/views/handlers.

I mentioned handlers for a reason.
Serverless hype means that no one uses controllers anymore right? 😄

In this blog post, we'll explore why it's crucial not to put your business logic in the controllers/views/handlers
of your application and discuss alternative approaches to keep your codebase well-organized.

What are controllers?

Depending on where you're coming from and what your project looks like,
I'm assuming you're using one of these three - controllers, views, handlers.

If your project is built using Django or Flask probably you are using views.

If your project uses a Serverless stack such as AWS Lambda + API Gateway, you are probably using handlers.

If your project is built using litestar then you have controllers.

I mention all of these to tell you that it is just a naming because the general intention is the same.

Controllers/Views/Handlers play a vital role in the request-response cycle of a web application.
They are responsible for handling incoming HTTP requests, routing them to the appropriate functions, and returning HTTP responses.
That's it, nothing more!

Let’s establish that for the purpose of this article, I will be using controllers as a name.
The reason is simple - this name is not specific to any environment or framework and is commonly used in our industry.

What is business logic?

Business logic represents the core functionality of your application.
It includes tasks like data validation, calculations, database interactions, and decision-making processes.
Business logic is what makes your application unique and valuable.

The problems with mixing controllers and business logic

When you put your business logic inside your controllers, several issues can arise:

Reduced code readability and maintainability.
Difficulty in testing individual components.
Violation of the Single Responsibility Principle (SRP).
Potential code duplication.

To illustrate this, we will implement a simple article creation endpoint.

The requirements are as follows, nothing really complicated, but I think they are good enough for the example purpose:

The number of tags of the article may not be bigger than 10
The title of the article may not be smaller than 5 words
The content of the article may not be smaller than 50 words!
The article has to have at least one category!
The article has to have at least 2 reviewers
The e-mail addresses of reviewers need to be valid e-mail addresses
The article should be created with DRAFT status by default

Take a look at the code below:

# blog/examples/src/business_logic_in_controllers/example1.py

@app.post("/articles", status_code=201)
async def create_article(
    request: CreateArticleRequest,
    repository: Annotated[ArticleRepository, Depends(get_article_repository)]
) -> JSONResponse:
    if len(request.tags) > 10:
        raise HTTPException(status_code=400, detail="Number of tags may not be bigger than 10!")

    if len(request.title.split(" ")) < 5:
        raise HTTPException(status_code=400, detail="Title may not be smaller than 5 words!")

    if len(re.findall(r"\w+", request.content)) < 50:
        raise HTTPException(status_code=400, detail="Content of the article may not be smaller than 50 words!")

    if len(request.categories) < 1:
        raise HTTPException(status_code=400, detail="Article has to have at least one category!")

    reviewers = []
    for email in reviewers:
        try:
            validated_email = validate_email(email, check_deliverability=False)
            reviewers.append(validated_email)
        except EmailNotValidError:
            raise HTTPException(status_code=400, detail=f"Provided reviewer email address `{email}` is not correct")

    if len(reviewers) < 2:
        raise HTTPException(status_code=400, detail="Article has to have at least 2 reviewers")

    article_id = EntityId.new_one()
    article = Article(
        id=article_id,
        title=request.title,
        content=request.content,
        created_at=datetime.utcnow(),
        status=Status.DRAFT
    )

    repository.save(article)

    for reviewer in request.reviewers:
        notify_reviewers(reviewer)

    return JSONResponse(content={}, headers={"Location": f"/articles/{article.id}"})


def notify_reviewers(reviewer: str) -> None:
    print(f"Notifying reviewer {reviewer} that a new article is awaiting his review!")

After reading this code snippet, what do you feel? There are a lot of things happening there right?

Reading this code is hard
Testing is hard
Maintaining it in a longer time perspective is also hard
Adding new features - the same
What if we would want to add the article via the CLI command? Not possible at all!

⚠️ REMINDER - controllers should be only responsible for handling incoming HTTP requests,
routing them to the appropriate functions, and returning HTTP responses

What are the alternative approaches?

If you've gotten to this point, you probably know that there are other alternative approaches.

Adding everything to the controllers is not the only option that we have 😄 Thanks to God!

One approach to separate business logic is to create a service layer.
The second is to use command handlers.

About both of them, you can read on my blog. I compared them.
You can read about it here (Services vs Command Handlers - choosing the right approach in your application)!
Code examples are included there too.

To fill the gap I also created an article about How to implement CommandBus in Python.

Benefits of separating business Logic

Enhanced code readability and maintainability
Easier testing of business logic in isolation
Better adherence to SRP
Reduced risk of code duplication

Implementation tips

When implementing the chosen approach, consider the following tips:

Organize your codebase into clear directories and modules.
Ensure that controllers remain lightweight and only handle HTTP-related tasks.
Write comprehensive unit tests for your business logic components.

Summary

In conclusion, keeping your business logic out of your controllers is a fundamental practice in WEB API development.
It leads to cleaner, more maintainable code and facilitates testing and scalability.
By adopting alternative approaches like service layers, command handlers, or query handlers you can achieve a well-structured and robust codebase for your applications.

Share your thoughts and experiences with separating business logic in the comments below.
Have you faced any challenges or discovered unique solutions?
Let’s discuss it!

Services vs Command Handlers

Szymon — Mon, 03 Jul 2023 22:00:00 +0000

Intro

Organizing code in an application is crucial for its maintainability and scalability.
Two common approaches for structuring code are services and command handlers.
In this blog post, we will explore these approaches and discuss when it is appropriate to use each one.

Services

Services are an effective way to encapsulate business logic and provide higher-level operations.
They abstract away implementation details and promote code modularity.
Services are particularly useful when handling complex operations that involve multiple components.

Here's an example of a payment service:

class PaymentService:
    ...

    def process_payment(self, amount: Money, payment_method: str) -> None:
        # Validate payment details
        # Process payment
        # Update transaction records
        # Send confirmation email

In the code snippet above, the PaymentService encapsulates the logic for processing a payment.
It performs validation, processes the payment, updates transaction records, and sends a confirmation email.
By encapsulating these operations in a service, we can keep the code modular and maintainable.

Command Handlers

Command handlers focus on executing specific commands or actions.
They follow the Single Responsibility Principle (SRP) and maintain separation of concerns.
Command handlers are beneficial when we want to handle one specific task or action per handler.

Consider the following example of a command handler for user registration:

class RegisterUserCommandHandler:
    def handle(self, command: RegisterUser) -> None:
        # Validate user input
        # Create user record in the database
        # Send welcome email

or in more Pythonic way, we can do this like that:

class RegisterUserCommandHandler:
    def __call__(self, command: RegisterUser) -> None:
        # Validate user input
        # Create user record in the database
        # Send welcome email

In this example, the RegisterUserCommandHandler is responsible for handling the registration of a user.
It validates the user input, creates a user record in the database, and sends a welcome email.
By using command handlers, we can keep our codebase focused and maintain a single responsibility per handler.

Pros and Cons

Both services and command handlers have their advantages and disadvantages. Let's explore them in more detail:

Services:

Pros:

Encapsulation: Services encapsulate business logic, making it easier to manage and reason about.
Modularity: Services abstract away implementation details, promoting code modularity and reusability.
Handling complex operations: Services can orchestrate multiple components and handle complex operations effectively.

Cons:

Potential for larger and more complex service classes: As the application grows, services may become larger and harder to maintain.
Difficulty in maintaining single responsibility: It can be challenging to maintain a single responsibility principle in large service classes.

Command Handlers:

Pros:

Simplicity: Command handlers focus on specific tasks, making the codebase simpler and more maintainable.
Single responsibility: Each command handler has a clear responsibility, promoting separation of concerns.
Ease of testing and maintenance: With command handlers, it's easier to write unit tests and maintain code due to their focused nature.

Cons:

Increased number of classes: As the number of commands grows, the number of command handlers can increase, potentially leading to a larger codebase.
Potential for duplication of code: Command handlers may share similar logic, leading to duplicated code if not properly managed.

Choosing the Right Approach

Choosing between services and command handlers depends on the specific requirements and complexity of your application.
Here are some guidelines to help you make an informed decision:

Consider the complexity and size of your application. If it involves complex operations and multiple components, services might be a suitable choice.
Evaluate the level of modularity and separation of concerns required. If your application requires a high degree of modularity and separation of responsibilities, command handlers can provide a more focused and maintainable approach.
Think about the potential for code reuse and maintenance. If there are common functionalities or operations that need to be reused across different parts of your application, services can help promote code reuse. On the other hand, if you anticipate frequent changes or updates to specific actions or tasks, command handlers can make maintenance and testing easier.
Consider the level of testing required for the application. If your application requires extensive testing at the command level or if you're following a CQRS (Command Query Responsibility Segregation) pattern, command handlers can facilitate unit testing and provide a clear separation between commands and queries.

By carefully considering these factors and understanding the trade-offs of each approach, you can choose the approach that best aligns with your application's needs, promoting code maintainability, and scalability.

Remember to document this decision! You can use ADRs to do this.

Outro

Remember, there is no one-size-fits-all solution.
The choice between services and command handlers depends on the unique characteristics of your project.
By carefully evaluating the complexity, modularity, code reuse, and testing requirements, you can select the approach that best aligns with your application's needs.

Code organization and maintainability are key factors in the long-term success of any software project.
So, take the time to plan and structure your codebase thoughtfully, keeping in mind the principles and patterns discussed in this blog post.

I hope this article has provided you with valuable insights into the services vs. command handlers debate and helps you make informed decisions in your projects.

Happy coding! 🚀

PS.

The plan for the next blog post is "How to implement CommandBus in Python" so stay tuned! 😉

How to test an API request to the external system

Szymon — Fri, 05 May 2023 22:00:00 +0000

Intro

Using external APIs is a common aspect of modern software development.
To be clear, by external API - I mean the API over which we have no control.
The simplest example I can think of is a payment gateway.
We have the 21st century, every business wants to have online payments.
To achieve this there are a lot of SaaS solutions like Braintree that allow us to do this by using their API.

In today’s blog post, I want to show you how you can quickly and easily write unit tests for it.

The problem

For this blog post purpose, let's imagine that we want to validate if the provided email address is a university email address.
We want to validate if the domain of provided email address belongs to some university.

To do this, we will use University Domains and Names Data List & API
this is an open-source project that provides the API that allows us to search for a university by domain.

We will use Python's requests library.
This is the implementation of our client:

# blog/examples/src/external_api_testing/university_email_address_validator.py

from abc import ABC, abstractmethod
from logging import Logger

import requests
from requests import RequestException

from src.external_api_testing.email_address import EmailAddress


class IUniversityEmailAddressValidator(ABC):
    @abstractmethod
    def validate(self, email: EmailAddress) -> bool:
        ...


class HipoLabsUniversityEmailAddressValidator(IUniversityEmailAddressValidator):
    """
    External API service for searching the university based on domain
    https://github.com/Hipo/university-domains-list/
    http://universities.hipolabs.com
    """

    def __init__(self, base_url: str, logger: Logger) -> None:
        self._base_url = base_url
        self._logger = logger

    def _is_university_domain(self, domain: str) -> bool:
        try:
            self._logger.info("Checking `%s` with HipooLabsClient", domain)

            url = f"{self._base_url}/search?domain={domain}"
            response = requests.get(url=url, timeout=(2, 3))
            response.raise_for_status()

            self._logger.info("Response [%s] json = %s", response.status_code, response.json())

            # If number of records in the response is greater than 0 it means that domain belongs to university
            return len(response.json()) > 0
        except RequestException as error:
            self._logger.error("An error occurred during domain verification!")
            self._logger.error("Error = %s", str(error))
            return False

    def validate(self, email: EmailAddress) -> bool:
        domain = email.domain

        if self._is_university_domain(domain):
            return True

        # skip suffix if email contains department name eg. @eti.pg.gda.pl -> the real domain is pg.gda.pl
        for part in domain.split(".")[:-2]:
            domain = domain[len(f"{part}."):]

            if self._is_university_domain(domain):
                return True

        return False

We want to test if the particular email address is a university one based on the response from HipooLabs API.

Let's see what possibilities we have 😉.

Solutions

Let me show you a few options using different libraries.

❌ ❌ unittest patch ❌ ❌

Before I show you the code example, I want to highlight this is an anti-pattern.
I’m showing this to you only for one reason - to make you aware of it and tell you to stop using it anymore.

# blog/examples/tests/test_external_api_testing/bad_example_dont_use_it/test_university_email_address_validator.py

from logging import Logger
from unittest.mock import Mock, patch

import pytest

from src.external_api_testing.email_address import EmailAddress
from src.external_api_testing.university_email_address_validator import HipoLabsUniversityEmailAddressValidator


@pytest.fixture
def hipo_base_url() -> str:
    return "http://universities.hipolabs.dev.com"


@pytest.fixture
def hipo_email_address_validator(hipo_base_url: str, test_logger: Logger) -> HipoLabsUniversityEmailAddressValidator:
    return HipoLabsUniversityEmailAddressValidator(hipo_base_url, test_logger)


@patch("src.external_api_testing.university_email_address_validator.requests")
def test_should_correctly_validate_university_email_address(
    mock_requests: Mock, hipo_email_address_validator: HipoLabsUniversityEmailAddressValidator
) -> None:
    # given
    mock_requests.get.return_value.status_code.return_value = 200
    mock_requests.get.return_value.json.return_value = [
        {
            "state-province": None,
            "domains": ["zut.edu.pl"],
            "country": "Poland",
            "web_pages": ["http://www.zut.edu.pl/"],
            "name": "Zachodniopomorska School of Science and Engineering",
            "alpha_two_code": "PL",
        }
    ]

    # when
    result = hipo_email_address_validator.validate(EmailAddress("john.deo@zut.edu.pl"))

    # then
    assert result is True


@patch("src.external_api_testing.university_email_address_validator.requests")
def test_should_correctly_validate_non_university_email_address(
    mock_requests: Mock, hipo_email_address_validator: HipoLabsUniversityEmailAddressValidator
) -> None:
    # given
    mock_requests.get.return_value.status_code.return_value = 200
    mock_requests.get.return_value.json.return_value = []

    # when
    result = hipo_email_address_validator.validate(EmailAddress("john.deo@gmail.com"))

    # then
    assert result is False

Let me explain why it is an anti-pattern, together with other articles that prove my opinion:

❌ it breaks the "don’t mock what you don’t own" rule
❌ if you change the requests.get() to requests.Session().get() the tests will fail
❌ if you change the file name or move it to another package the tests will fail, you will have to modify each decorator manually
❌ it makes your tests highly coupled with the implementation details

Additional materials:

✅ responses

responses is a library created by the Sentry team.
It is a utility library for mocking out the requests Python library.

# blog/examples/tests/test_external_api_testing/responses/test_university_email_address_validator.py

from logging import Logger

import pytest
import responses

from src.external_api_testing.email_address import EmailAddress
from src.external_api_testing.university_email_address_validator import HipoLabsUniversityEmailAddressValidator


@pytest.fixture
def hipo_base_url() -> str:
    return "http://universities.hipolabs.dev.com"


@pytest.fixture
def hipo_email_address_validator(hipo_base_url: str, test_logger: Logger) -> HipoLabsUniversityEmailAddressValidator:
    return HipoLabsUniversityEmailAddressValidator(hipo_base_url, test_logger)


@responses.activate
def test_should_correctly_validate_university_email_address(
    hipo_base_url: str, hipo_email_address_validator: HipoLabsUniversityEmailAddressValidator
) -> None:
    # given
    responses.add(
        responses.GET,
        f"{hipo_base_url}/search?domain=zut.edu.pl",
        json=[
            {
                "state-province": None,
                "domains": ["zut.edu.pl"],
                "country": "Poland",
                "web_pages": ["http://www.zut.edu.pl/"],
                "name": "Zachodniopomorska School of Science and Engineering",
                "alpha_two_code": "PL",
            }
        ],
        status=200,
    )

    # when
    result = hipo_email_address_validator.validate(EmailAddress("john.deo@zut.edu.pl"))

    # then
    assert result is True


@responses.activate
def test_should_correctly_validate_non_university_email_address(
    hipo_base_url: str,
    hipo_email_address_validator: HipoLabsUniversityEmailAddressValidator,
) -> None:
    # given
    responses.add(
        responses.GET,
        f"{hipo_base_url}/search?domain=gmail.com",
        json=[],
        status=200,
    )

    # when
    result = hipo_email_address_validator.validate(EmailAddress("john.deo@gmail.com"))

    # then
    assert result is False


@responses.activate
def test_should_fail_validation_if_network_error(
    hipo_base_url: str, hipo_email_address_validator: HipoLabsUniversityEmailAddressValidator
) -> None:
    # given
    responses.add(
        responses.GET,
        f"{hipo_base_url}/search?domain=zut.edu.pl",
        status=503,
    )
    responses.add(
        responses.GET,
        f"{hipo_base_url}/search?domain=edu.pl",
        status=503,
    )

    # when
    result = hipo_email_address_validator.validate(EmailAddress("john.deo@zut.edu.pl"))

    # then
    assert result is False

We have to add @responses.activate and then we can register our mock responses.

✅ wiremock

Another option is wiremock which is an open-sourced API mocking tool.
It also has a Python client.

When it comes to the Python environment, the disadvantage I can see is you have to run the wiremock docker image before running your tests.

wiremock setup:

# blog/examples/tests/test_external_api_testing/wiremock/conftest.py

import pytest
import requests
from requests import RequestException
from wiremock.constants import Config


@pytest.fixture()
def wiremock_base_url() -> str:
    return "http://localhost:8080"


@pytest.fixture
def wiremock(wiremock_base_url: str) -> None:
    wiremock_admin_url = f"{wiremock_base_url}/__admin"
    Config.base_url = wiremock_admin_url

    try:
        response = requests.get(wiremock_admin_url)
        response.raise_for_status()
    except RequestException:
        pytest.fail(
            "You forget to run wiremock docker instance! "
            "Use `docker-compose up -d` and run the container from `docker-compose.yml` file"
        )

tests:

# blog/examples/tests/test_external_api_testing/wiremock/test_university_email_address_validator.py

from logging import Logger

import pytest
from wiremock.resources.mappings import HttpMethods, Mapping, MappingRequest, MappingResponse
from wiremock.resources.mappings.resource import Mappings

from src.external_api_testing.email_address import EmailAddress
from src.external_api_testing.university_email_address_validator import HipoLabsUniversityEmailAddressValidator


@pytest.fixture
def hipo_email_address_validator(
    wiremock_base_url: str, test_logger: Logger
) -> HipoLabsUniversityEmailAddressValidator:
    return HipoLabsUniversityEmailAddressValidator(wiremock_base_url, test_logger)


@pytest.mark.usefixtures("wiremock")
def test_should_correctly_validate_university_email_address(
    hipo_email_address_validator: HipoLabsUniversityEmailAddressValidator,
) -> None:
    # given
    mapping = Mapping(
        request=MappingRequest(method=HttpMethods.GET, url="/search?domain=zut.edu.pl"),
        response=MappingResponse(
            status=200,
            json_body=[
                {
                    "state-province": None,
                    "domains": ["zut.edu.pl"],
                    "country": "Poland",
                    "web_pages": ["http://www.zut.edu.pl/"],
                    "name": "Zachodniopomorska School of Science and Engineering",
                    "alpha_two_code": "PL",
                }
            ],
        ),
    )
    Mappings.create_mapping(mapping=mapping)

    # when
    result = hipo_email_address_validator.validate(EmailAddress("john.deo@zut.edu.pl"))

    # then
    assert result is True


@pytest.mark.usefixtures("wiremock")
def test_should_correctly_validate_non_university_email_address(
    hipo_email_address_validator: HipoLabsUniversityEmailAddressValidator,
) -> None:
    # given
    mapping = Mapping(
        request=MappingRequest(method=HttpMethods.GET, url="/search?domain=gmail.com"),
        response=MappingResponse(status=200, json_body=[]),
    )
    Mappings.create_mapping(mapping=mapping)

    # when
    result = hipo_email_address_validator.validate(EmailAddress("john.deo@gmail.com"))

    # then
    assert result is False


@pytest.mark.usefixtures("wiremock")
def test_should_fail_validation_if_network_error(
    hipo_email_address_validator: HipoLabsUniversityEmailAddressValidator,
) -> None:
    # given
    Mappings.create_mapping(
        Mapping(
            request=MappingRequest(method=HttpMethods.GET, url="/search?domain=zut.edu.pl"),
            response=MappingResponse(fault="CONNECTION_RESET_BY_PEER"),
        )
    )
    Mappings.create_mapping(
        Mapping(
            request=MappingRequest(method=HttpMethods.GET, url="/search?domain=edu.pl"),
            response=MappingResponse(fault="CONNECTION_RESET_BY_PEER"),
        )
    )

    # when
    result = hipo_email_address_validator.validate(EmailAddress("john.deo@zut.edu.pl"))

    # then
    assert result is False

✅ vcrpy

vcrpy is an interesting tool -
it allows to run the tests using a real endpoint while simultaneously capturing and serializing the request and the response to the yaml files.

When you run the tests again, it mocks our HTTP request based on the previously saved yaml files.

You can run test run against the real API whenever you want, and it will update the yaml request/response files.

All we have to do is to add @vcr.use_cassette() decorator, where the param is the file path where the yaml files should be saved.

# blog/examples/tests/test_external_api_testing/vcrpy/test_university_email_address_validator.py

from logging import Logger
from pathlib import Path

import pytest
import vcr

from src.external_api_testing.email_address import EmailAddress
from src.external_api_testing.university_email_address_validator import HipoLabsUniversityEmailAddressValidator


@pytest.fixture
def hipo_base_url() -> str:
    return "http://universities.hipolabs.com"


@pytest.fixture
def hipo_email_address_validator(hipo_base_url: str, test_logger: Logger) -> HipoLabsUniversityEmailAddressValidator:
    return HipoLabsUniversityEmailAddressValidator(hipo_base_url, test_logger)


@vcr.use_cassette(str(Path(__file__).parent / "correct_university_email_address.yaml"))
def test_should_correctly_validate_university_email_address(
    hipo_base_url: str, hipo_email_address_validator: HipoLabsUniversityEmailAddressValidator
) -> None:
    # when
    result = hipo_email_address_validator.validate(EmailAddress("john.deo@zut.edu.pl"))

    # then
    assert result is True


@vcr.use_cassette(str(Path(__file__).parent / "incorrect_university_email_address.yaml"))
def test_should_correctly_validate_non_university_email_address(
    hipo_base_url: str,
    hipo_email_address_validator: HipoLabsUniversityEmailAddressValidator,
) -> None:
    # when
    result = hipo_email_address_validator.validate(EmailAddress("john.deo@gmail.com"))

    # then
    assert result is False

![vcrpy yaml files]

As you can see after the first tests suite execution the yaml files have been added.

✅ sandbox environment

The last option that comes to my mind is a sandbox environment.
Not every provider gives such a possibility but if the API provider has the sandbox environment then we can try using it in our tests.

But we have to remember that it may make our tests fragile.
If the sandbox environment is down our tests will ❌ fail ❌ - which does not mean that our code/business logic is not working as expected.
That's the downside of it.

Summary

![🎉 🎉 🎉]

✅ All tests passed ✅ 🎉.

All source code is available also on my GitHub here 🚀.

HipoLabsUniversityEmailAddressValidator implementation - https://github.com/szymon6927/szymonmiks.pl/blob/master/blog/examples/src/external_api_testing/university_email_address_validator.py

all tests - https://github.com/szymon6927/szymonmiks.pl/tree/master/blog/examples/tests/test_external_api_testing

I hope I helped you, and now testing your code that uses external API will be much easier for you.

When it comes to my personal preferences, I prefer responses I used this tool in many projects, and it works well for me.

Let me know what do you use, and which option is most preferable for you 😉 .

Happy testing!

How to make an API request to an AWS microservice that is protected by IAM auth

Szymon — Mon, 01 May 2023 22:00:00 +0000

Intro

One of the authorization methods that AWS supports for the API Gateway endpoints is
IAM authorization.

Two things are required to use IAM auth:

signed request using Signature Version 4
execute-api permission set up for the client for invoked endpoint

There are other authorization methods available like: Lambda authorizers or JWT authorizers you can read more about them
here.

In today's blog post, I will show you how to request a microservice that is protected by IAM auth.

The problem

For the blog post purpose, let's imagine we have two microservices: Microservice A and Microservice B.
Both of them were built using AWS lambda and API Gateway.
We own Microservice A, and some other team owns Microservice B.
We want to call Microservice B to get the response, it exposes the endpoint GET /items, and this endpoint is protected by IAM auth.

Solution

import os
from urllib.parse import urlparse
from logging import Logger

import requests
from aws_requests_auth.boto_utils import BotoAWSRequestsAuth
from requests.exceptions import RequestException

class MicroserviceBClientClientError(Exception):
    pass


class MicroserviceBClient:
    def __init__(self, base_url: str, logger: Logger) -> None:
        self._base_url = base_url
        self._logger = logger

    @property
    def auth(self) -> BotoAWSRequestsAuth:
        return BotoAWSRequestsAuth(
            aws_host=urlparse(self._base_url).hostname, aws_region=os.environ["AWS_REGION"], aws_service="execute-api"
        )

    def get_items(self) -> None:
        try:
            self._logger.info("Getting items from MicroserviceB!")
            response = requests.post(
                f"{self._base_url}/items", auth=self.auth
            )
            response.raise_for_status()
            self._logger.info(f"Successful request! Response = {response.json()}")
        except RequestException as error:
            self._logger.error(f"An error occurred during request to `MicroserviceB` service! Error = {error}")
            raise MicroserviceBClientClientError

The aws-requests-auth does most of the things for us.

We need to provide the hostname of the service we want to call, the AWS region,
and the service - in our case it is execute-api as we are working in a serverless lambda environment.

BotoAWSRequestsAuth generates the appropriate headers and adds them to the requests object.
All we need to do is to add it as a auth param to the requests method.

Summary

It is as simple as that 😉 I hope you enjoyed it.

There are other methods that you can use to make such a request.
The one I showed you is simple and easy.
I have tested it on production, it is working 😉.

How to make the right decision when it comes to software architecture?

Szymon — Tue, 18 Apr 2023 22:00:00 +0000

Intro

There are many definitions of software architecture.

The definition that appeals to me the most is a set of made decisions.
These decisions are driven by something right? That is exactly what architectural drivers are.

This is something that influences our decisions about architecture.
It shapes our software.
For example, we chose a relational database or some library for logging,
or we decided to follow hexagonal architecture principles - each of them has an influence on the architecture.

Architectural drivers is a set of things that shape, influence, and drive the software architecture.

Classes of architectural drivers

We identify 5 different classes of architectural drivers. Here is a short description of each of them.

Functional requirements

It is a list of functions that our software needs to provide.
We like to call it business logic.

For example, our system needs to send an email after successful account creation.
This potentially leads us to choose some email services providers like AWS SES or others.

Quality attributes

This includes all technical elements like:

scalability
extensibility
security
auditability
configurability
observability
testability

and many others 😉.

Project's constraints

Each project is different, the team size, the team's experience, budget, deadline, etc.
This is also something that influences our architecture.

Let me give you an example from Python's world.

If there is a short deadline and our team knows Django very well. We won't choose some new framework that no one knows.
But instead, most probably we will choose the well-known framework that we as a team feel comfortable with.

Another example: If the project's budget is low, we won't choose a very expensive CI/CD tool.

Conventions

These are all the rules that our organization/team follows to have cohesive solutions.

Example: If our organization uses pytest in all projects, and this is standard and common practice.
We should not use unittest library without having strong indications and reasoning behind it.

Project's goals

For example, our project might be some company internal tool or MVP.

If it is a company internal tool we might not be interested in a high-performance database there or an auto-scaling mechanism.
Or if it is an MVP we might want to do some shortcuts and use an in-memory database, just to prove that something is working as we expect.

An example from the DDD's world.
If it is a "generic" or "supporting" domain, we might want to use some already existing tool on the market or order the development from someone else - freelancers or software houses.

Documenting the decisions

Some time ago, I wrote an article about ADR.
This is one of the ways how you can document your architectural decisions.
In my opinion, the easiest and cheapest one. I recommend it.
If you have not heard anything about ADRs I encourage you to read my article 😉.

Discovering the architectural drivers

It is important to discover as many architectural drivers as we can.
Especially at the beginning of the project.
Correctly discovered might save us money and time.
We make sure that we build what is intended to be built.
It makes us closer to the solution and allows us to narrow down the area of the unknown.

The source of the information about them is business stakeholders.
They know what needs to be done, what features should be there, what will be the number of users on live, etc.

How we can get those pieces of information from them?

One of the techniques to discover the architectural drivers is stakeholder mapping.

The diagram above is called the stakeholders' matrix.
In this diagram, we can see who has the highest interest and the highest impact on the project, also we can see who has the lowest interest and impact on the project.

With that knowledge, we know that different people will have different perspectives on the project.
Most probably the project’s investor will be interested in features and delivery time, while CTO might be more interested in conventions and standards that are present in the company.

This is our role to discover the most important architectural drivers and reconcile the interests of the most important business stakeholders.
Be a partner for business stakeholders.

Summary

Everything is a trade-off.
Especially in the IT industry 😉.

Sometimes the architectural drivers may exclude each other.
High availability is against a low project budget.

We can not say that solution X is better than solution Y.
Everything has its own context.

By discovering the architectural drivers we may achieve a "good enough" situation within the current context.
It does not mean that is it the best option ever.
Instead, it means that it is the best option for the current situation, requirements, constraints, etc.

Making a decision is hard, making a good decision is even harder.
I hope I showed you how you can minimize the risk of making a wrong decision about your software architecture.

I hope you enjoyed it.
Let me know how you make decisions about software architecture, and what systems/tools/techniques do you use.

Additional Resources

How to assume an IAM role to access another AWS account

Szymon — Wed, 05 Apr 2023 22:00:00 +0000

Intro

AWS is one of the most popular cloud service providers nowadays.
During the writing of this blog post, AWS has ~33% of the market, so it's very probably you will work with AWS, or you're already working with them 😉.

In today's blog post, I will show you how to assume an IAM role using Python and boto3 library to access another AWS account.

Sometimes you need to access something from a different AWS account. It is called cross account access.
For example, you may want to get a secret from the secrets manager.
By default, it is not possible, but AWS comes with a solution.
You can assume the role that has permission to the actions you want to do.
Once you do this, you can act on the other account.

You can read more about it here.

This blog post will be different,
I'm not going to go into details about what boto3 is,
or what iam role is.
Instead, I will jump straight to the code.

Code

I will use the example from the intro section. We want to delete a secret from the secrets manager on a different AWS account.

The code together with the tests
is also available on my GitHub here 🚀.

# blog/examples/src/assume_iam_role/delete_secret.py

import os
from typing import Any

import structlog
from boto3 import Session
from botocore.exceptions import ClientError

from src.assume_iam_role import LambdaEvent, SecretsManagerSdkClient, STSSdkClient
from src.assume_iam_role.errors import SecretsDeletionError

LOGGER = structlog.get_logger()


def handle(event: LambdaEvent, context: Any, sts_sdk_client: STSSdkClient) -> LambdaEvent:
    if "ROLE_ARN_TO_ASSUME" not in os.environ:
        return event

    try:
        secret_id = event["secret_id"]

        response = sts_sdk_client.assume_role(
            RoleArn=os.environ["ROLE_ARN_TO_ASSUME"],
            RoleSessionName="blogpost-delete-secret",
        )

        session = Session(
            aws_access_key_id=response["Credentials"]["AccessKeyId"],
            aws_secret_access_key=response["Credentials"]["SecretAccessKey"],
            aws_session_token=response["Credentials"]["SessionToken"],
        )
        client: SecretsManagerSdkClient = session.client("secretsmanager")

        LOGGER.info(f"Deleting secret from secretsmanager! secret_id={secret_id}")
        response = client.delete_secret(SecretId=secret_id, ForceDeleteWithoutRecovery=True)  # type: ignore
        LOGGER.info(f"Successful request! Response = {response}")
        return event
    except ClientError as error:
        if error.response["Error"]["Code"] == "ResourceNotFoundException":
            return event

        LOGGER.error(f"An error occurred during request to SecretsManager service. Error = {error}")
        raise SecretsDeletionError
    except Exception as error:
        LOGGER.error(f"An unknown error occurred during `delete_secret.handle` function execution. Error = {error}")
        raise SecretsDeletionError

and the test:

# blog/examples/tests/test_assume_iam_role/conftest.py

import os
from typing import Generator

import boto3
import pytest
from _pytest.monkeypatch import MonkeyPatch
from moto import mock_secretsmanager, mock_sts

from src.assume_iam_role import SecretsManagerSdkClient, STSSdkClient


@pytest.fixture(autouse=True)
def aws_envs() -> None:
    os.environ["AWS_REGION"] = "eu-west-1"
    os.environ["AWS_DEFAULT_REGION"] = "eu-west-1"
    os.environ["AWS_ACCESS_KEY_ID"] = "XXX"
    os.environ["AWS_SECRET_ACCESS_KEY"] = "XXX"


@pytest.fixture()
def different_account_id() -> str:
    return "111111111111"


@pytest.fixture()
def secret_id() -> str:
    return "/test/secret/blogpost/szymonmiks"


@pytest.fixture()
def mock_sts_client() -> Generator[STSSdkClient, None, None]:
    with mock_sts():
        client: STSSdkClient = boto3.client("sts")
        yield client


@pytest.fixture()
def mock_secrets_manager_client(
    secret_id: str, monkeypatch: MonkeyPatch, different_account_id: str
) -> Generator[SecretsManagerSdkClient, None, None]:
    with mock_secretsmanager():
        # http://docs.getmoto.org/en/latest/docs/multi_account.html#configure-an-account-using-sts
        monkeypatch.setenv("MOTO_ACCOUNT_ID", different_account_id)
        client: SecretsManagerSdkClient = boto3.client("secretsmanager")
        client.create_secret(Name=secret_id)

        yield client

from _pytest.monkeypatch import MonkeyPatch

from src.assume_iam_role import LambdaEvent, SecretsManagerSdkClient, STSSdkClient
from src.assume_iam_role.delete_secret import handle

def test_can_delete_secret_using_assumed_role(
    mock_sts_client: STSSdkClient,
    mock_secrets_manager_client: SecretsManagerSdkClient,
    monkeypatch: MonkeyPatch,
    secret_id: str,
    different_account_id: str,
) -> None:
    # given
    event = LambdaEvent({"secret_id": secret_id})
    monkeypatch.setenv("ROLE_ARN_TO_ASSUME", f"arn:aws:iam::{different_account_id}:role/test_role_to_assume")

    # when
    result = handle(event, None, mock_sts_client)

    # then
    assert result == event
    assert mock_secrets_manager_client.list_secrets()["SecretList"] == []

That’s it. As simple as that 😄.

Happy coding 😉 .

Domain model with SQLAlchemy

Szymon — Thu, 16 Feb 2023 23:00:00 +0000

Intro

Welcome to the world of databases and domain modelling!

In this blog post, we will explore the power of SQLAlchemy, a popular ORM library in Python, to model our domain objects.

We will look at how to define our entities, and how to persist them into a relational database.

Whether you're a beginner or an experienced Python developer, this post will provide you with the knowledge and skills you need to build robust, scalable, and maintainable domain models with SQLAlchemy.

So, let's dive in and see how we can bring our domains to life!

Domain model

Before we jump straight into the code let's define what the domain model is, so we are on the same page.

Domain Model is a representation of the concepts and relationships that exist in a particular business or problem domain.

According to Fowler, a domain model is an object-oriented model that encapsulates the behaviour and data of the objects in the domain, and provides a clear and concise representation of the problem being solved.

The purpose of a domain model is to provide a common language and understanding of the problem domain, between the developers and stakeholders.
A well-designed domain model helps to bridge the gap between the technical and business perspectives, and provides a clear and concise representation of the system's requirements and behaviour.

The domain model highlights the importance of focusing on the behaviour and interactions of the objects in the domain, rather than on the implementation details.

Ways of persisting domain models

I will use the domain model from one of the previous posts. This is a simple representation of a library card.

This class is a good candidate.
It is complex enough.
It has some related objects and quite a lot of value objects.
Only business methods are exposed, and implementation details are hidden.

Our goal is simple: save the object to the database, but the object itself should know nothing about the database.

The code is also available on my GitHub here 🚀.

Imperative mapping

SQLAlchemy comes with a nice feature called "Imperative mapping".

Imperative mapping in SQLAlchemy is a technique for defining the relationship between database tables and Python classes using imperative code, rather than declarative code.
This involves creating a set of mappings that describe the relationships between tables, columns, and classes, and using this mapping to perform database operations.
Imperative mapping provides more flexibility than declarative mapping, but requires more manual setup and can be more verbose.

This is how our repository looks using this technique

# blog/examples/src/domain_model_with_sqlalchemy/imperative_mapping/infrastructure/sql_alchemy_library_card_repository.py

from sqlalchemy import CHAR, Column, DateTime, ForeignKey, String, Table
from sqlalchemy.exc import NoResultFound
from sqlalchemy.orm import Session, composite, joinedload, relationship

from src.domain_model_with_sqlalchemy.db import mapper_registry, metadata
from src.domain_model_with_sqlalchemy.errors import ResourceNotFound
from src.domain_model_with_sqlalchemy.imperative_mapping.domain.date_range import DateRange
from src.domain_model_with_sqlalchemy.imperative_mapping.domain.entity_id import EntityId
from src.domain_model_with_sqlalchemy.imperative_mapping.domain.library_card import LibraryCard
from src.domain_model_with_sqlalchemy.imperative_mapping.domain.library_card_repository import ILibraryCardRepository
from src.domain_model_with_sqlalchemy.imperative_mapping.domain.rental import Rental
from src.domain_model_with_sqlalchemy.imperative_mapping.domain.status import Status


def init_mappers() -> None:
    library_card_table = Table(
        "library_cards_imperative_mapping",
        metadata,
        Column("id", CHAR(32), primary_key=True),
        Column("owner_id", CHAR(32), nullable=False),
        Column("status", String(20), nullable=False),
        Column("created_at", DateTime, nullable=False),
    )

    rental_table = Table(
        "rentals_imperative_mapping",
        metadata,
        Column("id", CHAR(32), primary_key=True),
        Column("library_card_id", CHAR(32), ForeignKey("library_cards_imperative_mapping.id"), nullable=False),
        Column("resource_id", CHAR(32), nullable=False),
        Column("rental_period_start_date", DateTime, nullable=False),
        Column("rental_period_end_date", DateTime, nullable=False),
    )

    mapper_registry.map_imperatively(
        LibraryCard,
        library_card_table,
        properties={
            "_id": composite(EntityId.of, library_card_table.c.id),
            "__id": library_card_table.c.id,
            "_owner_id": composite(lambda value: EntityId.of(value), library_card_table.c.owner_id),
            "__owner_id": library_card_table.c.owner_id,
            "_rentals": relationship(Rental, cascade="all, delete-orphan"),
            "_status": composite(Status, library_card_table.c.status),
            "__status": library_card_table.c.status,
        },
        column_prefix="_",
    )

    mapper_registry.map_imperatively(
        Rental,
        rental_table,
        properties={
            "_id": composite(lambda value: EntityId.of(value), rental_table.c.id),
            "__id": rental_table.c.id,
            "_resource_id": composite(lambda value: EntityId.of(value), rental_table.c.resource_id),
            "__resource_id": rental_table.c.resource_id,
            "_rental_period": composite(
                lambda start_date, end_date: DateRange(start_date=start_date, end_date=end_date),
                rental_table.c.rental_period_start_date,
                rental_table.c.rental_period_end_date,
            ),
        },
        column_prefix="_",
    )


class SqlAlchemyLibraryCardRepository(ILibraryCardRepository):
    def __init__(self, session: Session) -> None:
        self._session = session

    def get(self, library_card_id: EntityId) -> LibraryCard:
        try:
            result: LibraryCard = (
                self._session.query(LibraryCard).options(joinedload("*")).filter_by(_id=library_card_id).one()
            )
            return result
        except NoResultFound as error:
            raise ResourceNotFound(str(library_card_id)) from error

    def save(self, library_card: LibraryCard) -> None:
        self._session.merge(library_card)

And the entity itself:

# blog/examples/src/domain_model_with_sqlalchemy/imperative_mapping/domain/library_card.py

from datetime import datetime

from src.domain_model_with_sqlalchemy.errors import BorrowingError, ResourceNotFound
from src.domain_model_with_sqlalchemy.imperative_mapping.domain.date_range import DateRange
from src.domain_model_with_sqlalchemy.imperative_mapping.domain.entity_id import EntityId
from src.domain_model_with_sqlalchemy.imperative_mapping.domain.rental import Rental
from src.domain_model_with_sqlalchemy.imperative_mapping.domain.status import Status


class LibraryCard:
    def __init__(
        self,
        id: EntityId,
        owner_id: EntityId,
        rentals: list[Rental],
        status: Status,
        created_at: datetime = datetime.utcnow(),
    ) -> None:
        self._id = id
        self._owner_id = owner_id
        self._rentals = rentals
        self._status = status
        self._created_at = created_at

    @property
    def id(self) -> EntityId:
        return self._id

    @classmethod
    def create(cls, owner_id: EntityId) -> "LibraryCard":
        instance = cls(id=EntityId.new_one(), owner_id=owner_id, rentals=[], status=Status.ACTIVE)
        return instance

    def borrow(self, resource_id: EntityId, how_long: DateRange) -> None:
        if not self.is_active():
            raise BorrowingError("Can not borrow")

        days_diff = (how_long.end_date - how_long.start_date).days
        if days_diff > 14:
            raise BorrowingError("You cannot borrow for more than 14 days at once!")

        if len(self._rentals) >= 3:
            raise BorrowingError("You can only have 3 resources borrowed!")

        rental = Rental.create(resource_id, how_long)
        self._rentals.append(rental)

    def prolong_for(self, rental_id: EntityId, days: int) -> None:
        rental = next((rental for rental in self._rentals if rental.id == rental_id), None)

        if not rental:
            raise ResourceNotFound(str(rental_id))

        rental.prolong(days)

    def give_back(self, resource_id: EntityId) -> None:
        rental = next((rental for rental in self._rentals if rental.resource_id == resource_id), None)

        if not rental:
            raise ResourceNotFound(str(resource_id))

        self._rentals.remove(rental)

        days_diff = (datetime.utcnow() - rental.rental_period.end_date).days
        if days_diff > 3:
            self._status = Status.SUSPENDED

    def is_active(self) -> bool:
        return self._status == Status.ACTIVE

    def is_owned_by(self, owner: EntityId) -> bool:
        return self._owner_id == owner

    def __str__(self) -> str:
        return f"LibraryCard(id={self._id}, owner_by={self._owner_id}, rentals={self._rentals}, status={self._status})"

As you can see on the first listing we defined the tables and the mapping.

The entity itself is not aware of anything related to the database!

The only thing that is needed and comes directly from SQLAlchemy is __composite_values__ and this is needed to persist value objects.

Before I show you the code example, let me try to explain what __composite_values__ is 😉.

In SQLAlchemy, __composite_values__ is a special method that can be defined on a custom composite type class to define how instances of the composite type are serialized and deserialized.

A composite type is a custom data type that can be used to represent a group of related values as a single unit.
For example, a composite type could be used to represent a physical address as a combination of street, city, state, and zip code.
In our case we use it to represent EntityId or DateRange.

When you define a composite type in SQLAlchemy, you can define the attributes of the composite type and its data types, and you can also define various methods and properties on the class to customize its behaviour.

It is a method that should return a tuple of values which represents the attributes of the composite type.
When an instance of the composite type is serialized (when it is written to a database), SQLAlchemy calls the __composite_values__ method to obtain the values that should be stored.
When an instance of the composite type is deserialized (when it is read from a database), SQLAlchemy calls the constructor of the composite type with the tuple of values obtained from __composite_values__.

You can read more about it here.

The use case looks like that:

# blog/examples/src/domain_model_with_sqlalchemy/imperative_mapping/domain/date_range.py

from dataclasses import dataclass
from datetime import datetime
from typing import Tuple

from dateutil.relativedelta import relativedelta


@dataclass(frozen=True)
class DateRange:
    start_date: datetime
    end_date: datetime

    def __composite_values__(self) -> Tuple[datetime, datetime]:
        return self.start_date, self.end_date

    def __post_init__(self) -> None:
        if self.start_date > self.end_date:
            raise ValueError("Can not create DateRange")

    @classmethod
    def one_week(cls) -> "DateRange":
        start_date = datetime.now()
        end_date = start_date + relativedelta(weeks=1)
        return cls(start_date, end_date)

    @classmethod
    def two_weeks(cls) -> "DateRange":
        start_date = datetime.now()
        end_date = start_date + relativedelta(weeks=2)
        return cls(start_date, end_date)

    @classmethod
    def one_month(cls) -> "DateRange":
        start_date = datetime.now()
        end_date = start_date + relativedelta(months=1)
        return cls(start_date, end_date)

    def is_within_range(self, date: datetime) -> bool:
        return self.start_date <= date <= self.end_date

Separate model

Another technique, I like to call "separate model". The idea is to create SQLAlchemy models and map them manually
onto the entity, on the repository level.

# blog/examples/src/domain_model_with_sqlalchemy/separate_model/infrastructure/sql_alchemy_library_card_repository.py

from sqlalchemy.exc import NoResultFound
from sqlalchemy.orm import Session, joinedload

from src.domain_model_with_sqlalchemy.errors import ResourceNotFound
from src.domain_model_with_sqlalchemy.separate_model.domain.date_range import DateRange
from src.domain_model_with_sqlalchemy.separate_model.domain.entity_id import EntityId
from src.domain_model_with_sqlalchemy.separate_model.domain.library_card import LibraryCard
from src.domain_model_with_sqlalchemy.separate_model.domain.library_card_repository import ILibraryCardRepository
from src.domain_model_with_sqlalchemy.separate_model.domain.rental import Rental
from src.domain_model_with_sqlalchemy.separate_model.domain.status import Status
from src.domain_model_with_sqlalchemy.separate_model.infrastructure.model import LibraryCardModel, RentalModel


class SqlAlchemyLibraryCardRepository(ILibraryCardRepository):
    def __init__(self, session: Session) -> None:
        self._session = session

    def _to_model(self, library_card: LibraryCard) -> LibraryCardModel:
        rentals = [
            RentalModel(
                id=str(rental.id),
                library_card_id=str(library_card.id),
                resource_id=str(rental.resource_id),
                rental_period_start_date=rental.rental_period.start_date,
                rental_period_end_date=rental.rental_period.end_date,
            )
            for rental in library_card.rentals
        ]

        return LibraryCardModel(
            id=str(library_card.id),
            owner_id=str(library_card.owner_id),
            status=library_card.status.value,
            created_at=library_card.created_at,
            rentals=rentals,
        )

    def get(self, library_card_id: EntityId) -> LibraryCard:
        try:
            result: LibraryCardModel = (
                self._session.query(LibraryCardModel).options(joinedload("*")).filter_by(id=str(library_card_id)).one()
            )
        except NoResultFound as error:
            raise ResourceNotFound(str(library_card_id)) from error

        rentals = [
            Rental(
                _id=EntityId.of(rental_model.id),
                _resource_id=EntityId.of(rental_model.resource_id),
                _rental_period=DateRange(
                    start_date=rental_model.rental_period_start_date, end_date=rental_model.rental_period_end_date
                ),
            )
            for rental_model in result.rentals
        ]

        return LibraryCard(
            id=EntityId.of(result.id),
            owner_id=EntityId.of(result.owner_id),
            rentals=rentals,
            status=Status(result.status),
            created_at=result.created_at,
        )

    def save(self, library_card: LibraryCard) -> None:
        model = self._to_model(library_card)
        self._session.merge(model)

Then we have a separate file where we define our SQLAlchemy models.

# blog/examples/src/domain_model_with_sqlalchemy/separate_model/infrastructure/model.py

from sqlalchemy import CHAR, Column, DateTime, ForeignKey, String
from sqlalchemy.orm import relationship

from src.domain_model_with_sqlalchemy.db import Base


class LibraryCardModel(Base):
    __tablename__ = "library_cards_separate_model"

    id = Column(CHAR(32), primary_key=True)
    owner_id = Column(CHAR(32), nullable=False)
    status = Column(String(20), nullable=False)
    created_at = Column(DateTime, nullable=False)
    rentals = relationship("RentalModel", cascade="all, delete-orphan")

    def __str__(self) -> str:
        return (
            f"LibraryCardModel(id={self.id}, owner_id={self.owner_id}, "
            f"status={self.status}, created_at={self.created_at})"
        )


class RentalModel(Base):
    __tablename__ = "rentals_separate_model"

    id = Column(CHAR(32), primary_key=True)
    library_card_id = Column(CHAR(32), ForeignKey("library_cards_separate_model.id"), nullable=False)
    resource_id = Column(CHAR(32), nullable=False)
    rental_period_start_date = Column(DateTime, nullable=False)
    rental_period_end_date = Column(DateTime, nullable=False)

    def __str__(self) -> str:
        return (
            f"RentalModel(id={self.id}, library_card_id={self.library_card_id}, resource_id={self.resource_id}, "
            f"rental_period_start_date={self.rental_period_start_date}, "
            f"rental_period_end_date={self.rental_period_end_date})"
        )

When it comes to our entity, we only had to add public getters to be able to map all properties to our SQLAlchemy model.
The rest stays the same.

# blog/examples/src/domain_model_with_sqlalchemy/separate_model/domain/library_card.py

from datetime import datetime

from src.domain_model_with_sqlalchemy.errors import BorrowingError, ResourceNotFound
from src.domain_model_with_sqlalchemy.separate_model.domain.date_range import DateRange
from src.domain_model_with_sqlalchemy.separate_model.domain.entity_id import EntityId
from src.domain_model_with_sqlalchemy.separate_model.domain.rental import Rental
from src.domain_model_with_sqlalchemy.separate_model.domain.status import Status


class LibraryCard:
    def __init__(
        self,
        id: EntityId,
        owner_id: EntityId,
        rentals: list[Rental],
        status: Status,
        created_at: datetime = datetime.utcnow(),
    ) -> None:
        self._id = id
        self._owner_id = owner_id
        self._rentals = rentals
        self._status = status
        self._created_at = created_at

    @property
    def id(self) -> EntityId:
        return self._id

    @property
    def owner_id(self) -> EntityId:
        return self._owner_id

    @property
    def rentals(self) -> list[Rental]:
        return self._rentals

    @property
    def status(self) -> Status:
        return self._status

    @property
    def created_at(self) -> datetime:
        return self._created_at

    @classmethod
    def create(cls, owner_id: EntityId) -> "LibraryCard":
        instance = cls(id=EntityId.new_one(), owner_id=owner_id, rentals=[], status=Status.ACTIVE)
        return instance

    def borrow(self, resource_id: EntityId, how_long: DateRange) -> None:
        if not self.is_active():
            raise BorrowingError("Can not borrow")

        days_diff = (how_long.end_date - how_long.start_date).days
        if days_diff > 14:
            raise BorrowingError("You cannot borrow for more than 14 days at once!")

        if len(self._rentals) >= 3:
            raise BorrowingError("You can only have 3 resources borrowed!")

        rental = Rental.create(resource_id, how_long)
        self._rentals.append(rental)

    def prolong_for(self, rental_id: EntityId, days: int) -> None:
        rental = next((rental for rental in self._rentals if rental.id == rental_id), None)

        if not rental:
            raise ResourceNotFound(str(rental_id))

        rental.prolong(days)

    def give_back(self, resource_id: EntityId) -> None:
        rental = next((rental for rental in self._rentals if rental.resource_id == resource_id), None)

        if not rental:
            raise ResourceNotFound(str(resource_id))

        self._rentals.remove(rental)

        days_diff = (datetime.utcnow() - rental.rental_period.end_date).days
        if days_diff > 3:
            self._status = Status.SUSPENDED

    def is_active(self) -> bool:
        return self._status == Status.ACTIVE

    def is_owned_by(self, owner: EntityId) -> bool:
        return self._owner_id == owner

    def __str__(self) -> str:
        return f"LibraryCard(id={self._id}, owner_by={self._owner_id}, rentals={self._rentals}, status={self._status})"

Again the entity itself is not aware of anything related to the database!

Summary

We discussed how to use SQLAlchemy to persist domain models to the database.

We started by introducing the concept of a domain model and explaining why it's important to separate domain logic from persistence logic.

Then we discussed different ways of persisting domain models with SQLAlchemy.

Code examples for each of these approaches demonstrated how to define domain models and how to persist them to a database using SQLAlchemy.
They also discussed the pros and cons of each approach.

The techniques we discussed can be used to implement architecture patterns like Domain-Driven Design (DDD) and hexagonal architecture.
By separating domain logic from persistence logic, you can create more modular and maintainable systems that are easier to test and evolve over time.
SQLAlchemy provides a powerful toolkit for building domain models and persisting them to a database, allowing you to choose the right approach for your specific use case and architecture.

I hope you enjoyed it.

Let me know, how you approach the domain model and persistence. Which technique do you use? I would love to hear your opinion 😉.

How I start every new Python backend API project

Szymon — Fri, 20 Jan 2023 07:46:46 +0000

Intro

In today's blog post I would like to show you how I approach every new Python backend API project setup.

This blog post includes:

project structure
tools
best practices
automation

Let's go! 🚀

Poetry

Poetry in my opinion is the best Python packaging and dependency management tool.

I'm using it in all of my projects since 2019.
If you have not heard about Poetry yet, I highly recommend reading about this tool.
In my opinion, it is the best option that we currently have on the Python market.

Project creation

To create a new project with poetry you need to run this command:

$ poetry new <your-project-name>

Project structure

├── .docker
├── .gitignore
├── .pre-commit-config.yaml
├── Makefile
├── README.md
├── docs
│ ├── adr
│ │   └── adr-001-example_adr.md
│ └── api
│     └── openapi.yaml
├── http_requests
│ ├── http-client.env.json
│ └── ping.http
├── iac
├── pyproject.toml
├── setup.cfg
├── src
│ ├── building_blocks
│ │ └── logger.py
│ ├── domain_module_a
│ │ ├── application
│ │ ├── domain
│ │ └── infrastructure
│ ├── domain_module_b
│ └── domain_module_c
└── tests
    ├── domain_module_a
    ├── domain_module_b
    └── domain_module_c

I will explain the purpose of each directory, starting from the top.

📁 .docker

If my project uses docker, I put all docker-related files here.
For example: init scripts for localstack.

📁 docs

Inside this directory, I keep all things related to the project's documentation.

The openapi.yaml file resides inside api sub-directory.

Inside the adr sub-directory, I keep the project's ADRs.
I wrote a separate article about the ADRs, you can read it here.

📁 http_requests

I use IntelliJ HTTP Client
and here is where I keep request's definitions.

If you have not heard about this tool, I recommend you check it out. Having this, you do not have to have any
additional API clients like postman or insomnia.

📁 iac

If my project uses Infrastructure as code
here is where I keep terraform or other files.

📁 src

Here I keep the code responsible for the application itself. This directory contains sub-directories.
I do not like splitting the project into technical layers. In opposition to that, I follow the convention
where each module is named accordingly and responsible for the business domain that it belongs to.

One of the benefits of it is that each module can have a different type of application architecture.
As you can see in the domain_module_a the separation is done by hexagonal architecture rules.
It has no impact on the other modules where such separation is not needed.

You can see there is a module called building_blocks.
Inside it, I keep all the utilities needed in the project, like a logger, serializers, and so on.
I did not make up this name, I borrowed it from this repo.

README.md

It is very important to take care of the README file because:

it is the first thing that will be shown after opening the project's repo
it allows new members of the team to start working with the project more smoothly

Here is my proposition for the README file

# <project_name>

Quick project description.

## Table of contents

* [Stack](#stack)
* [Prerequisites](#prerequisites)
* [Setup](#setup)
* [Intellij / PyCharm configuration](#intellij-/-pycharm-configuration)
* [Tests](#tests)
* [CI/CD](#ci/cd)
* [Monitoring](#monitoring)
* [ADR](#adr)
* [HTTP requests](#http-requests)

## Stack

Description of the technology stack used in the project.

## Prerequisites

Information about all needed tools you have to install before you start the development.

## Setup

Description of how to setup the project to be able to start the development.

## Architecture

Description of the project's architecture. Diagrams, maps, etc.

## Intellij / PyCharm configuration

Info on how to setup a project inside Intellij or other IDE.

## Tests

Description of how to run the tests.

## CI/CD

Description of what the CI/CD process looks like and how it works. What is the deployment strategy, etc.

## Monitoring

Information about tools used to monitor the application, how to use them, what is the purpose, how to access etc.

## ADR

Information about ADRs.

## HTTP requests

Information about IntelliJ HTTP client.

You can find the complete README with some example descriptions for each section here:
https://github.com/szymon6927/szymonmiks.pl/tree/master/blog/examples/example-project

Tools

Here is the list of tools that I always add while creating a new project:

If my project uses AWS services, I also install moto library.

I use pyproject.toml file to keep the config for each of them in one place.

This is what it looks like:

[tool.black]
line-length = 120
target-version = ["py39"]

[tool.coverage.report]
exclude_lines = [
    "pragma: no cover",
    "def __repr__",
    "raise AssertionError",
    "raise NotImplementedError",
    "pass",
    "if 0:",
    "if __name__ == .__main__.:",
    "nocov",
    "if TYPE_CHECKING:",
]
fail_under = 80
show_missing = true

[tool.coverage.run]
branch = true
omit = [
    "tests/*"
]

[tool.isort]
combine_as_imports = "true"
force_grid_wrap = 0
include_trailing_comma = "true"
known_first_party = "src"
line_length = 120
multi_line_output = 3

[tool.mypy]
disallow_untyped_defs = true
follow_imports = "silent"
ignore_missing_imports = true
python_version = "3.9"
warn_return_any = true
warn_unused_configs = true

[tool.pylint.BASIC]
good-names = "id,i,j,k"

[tool.pylint.DESIGN]
max-args = 5
max-attributes = 8
min-public-methods = 1

[tool.pylint.FORMAT]
max-line-length = 120

[tool.pylint."MESSAGES CONTROL"]
disable = "missing-docstring, line-too-long, logging-fstring-interpolation, duplicate-code"

[tool.pylint.MISCELLANEOUS]
notes = "XXX"

[tool.pylint.SIMILARITIES]
ignore-comments = "yes"
ignore-docstrings = "yes"
ignore-imports = "yes"
min-similarity-lines = 6

[tool.pytest.ini_options]
addopts = "-v --cov=src --cov-report term-missing --no-cov-on-fail"
testpaths = ["tests"]

The only exception is flake8 which does not support config inside pyproject.toml so we have to have an additional file
which is setup.cfg.

[flake8]
ignore = E501, W503, E203
max-line-length = 120

pre-commit

pre-commit allows you to add git hooks that will execute before you add your commit.

This is the config file that I use in my projects:

repos:
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.3.0
    hooks:
      - id: trailing-whitespace
      - id: check-merge-conflict
      - id: check-yaml
        args: [--unsafe]
      - id: check-json
      - id: detect-private-key
      - id: end-of-file-fixer

  - repo: https://github.com/timothycrosley/isort
    rev: 5.10.1
    hooks:
      - id: isort

  - repo: https://github.com/psf/black
    rev: 22.8.0
    hooks:
      - id: black

  - repo: https://gitlab.com/pycqa/flake8
    rev: 3.9.2
    hooks:
      - id: flake8

  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v0.971
    hooks:
      - id: mypy
        args: [ --warn-unused-configs, --ignore-missing-imports, --disallow-untyped-defs, --follow-imports=silent, --install-types, --non-interactive ]

Makefile

I like using Make to automate stuff in my projects.

Below you can see the Makefile that I use.


.DEFAULT_GOAL := all

toml_sort:
    toml-sort pyproject.toml --all --in-place

isort:
    poetry run isort .

black:
    poetry run black .

flake8:
    poetry run flake8 .

pylint:
    poetry run pylint src

dockerfile_linter:
    docker run --rm -i hadolint/hadolint < Dockerfile

validate_openapi_schema:
    poetry run openapi-spec-validator example-project/docs/api/openapi.yaml

mypy:
    poetry run mypy --install-types --non-interactive .

audit_dependencies:
    poetry export --without-hashes -f requirements.txt | poetry run safety check --full-report --stdin

bandit:
    poetry run bandit -r . -x ./tests,./test

test:
    poetry run pytest

lint: toml_sort isort black flake8 pylint mypy validate_openapi_schema

audit: audit_dependencies bandit

tests: test

all: lint audit tests

Complete example

You can find the complete example on my GitHub 🚀

https://github.com/szymon6927/szymonmiks.pl/tree/master/blog/examples/example-project

Summary

I hope it was useful to you, and I hope that after reading this article you adopt some concepts for your project.

If you have a different opinion, please let me know. I would like to know what you think about it.

Hexagonal architecture in Python

Szymon — Wed, 11 Jan 2023 14:55:07 +0000

Intro

Hexagonal architecture is one of the styles of application architecture that you can use in your project.
In today’s blog post I would like to show you how you can build your application using a hexagonal architecture with Python.

Instead of focusing on the theoretical side (there are already enough of these articles on the web), we will take a more practical approach.

History

Originally hexagonal architecture was invented by Alistair Cockburn in 2005.
Then around 2008, Jeffrey Palermo invented something very similar called onion architecture.
And then, as the last one, around 2011, Robert C. Martin came up with his idea called clean architecture.

So if you hear the following terms:

hexagonal architecture (aka. ports and adapters)
onion architecture
clean architecture
screaming architecture

It’s generally about the same idea, with some minor variations.

Quick intro

Please consider the following diagram:

Let me quickly introduce you to all the concepts included in the diagram.
The best explanation will be a code example where all these concepts are applied simultaneously.
For the example please have a look at the section below.

Generally speaking, ports are responsible for communication with the outside world.
Usually, they are implemented as interfaces, whereas adapters are concrete implementations of our ports.

input port (aka. driver port or primary port) - exposes the application features to the outside world. It’s an entry point to our business logic.

input adapter - is a concrete implementation of how we want to enter our application, for example via REST endpoint etc.

output port (aka. driven or secondary) - it is used to interact with outbound things, for example, reading/writing data from/to a database.

output adapter - it is the concrete implementation of the above. For example, implementation of communication with a specific database, e.g. MySQL.

use case (aka. application layer) - this layer answers the question "what to do". It controls the flow.

domain - in contrast to the above layer, this one answers the question "how to do".
Here our business logic lives. This is the place where our application makes money. This is the heart of our app.

As you can see on the diagram the use case layer knows what to do but is not aware of how to do it.
The use case layer delegates it to the domain layer.
Because of this, the domain layer is not aware of any ports or adapters.
The domain layer should not have any dependencies to the layers above it.

Example

For the purpose of this article I created a separate project on my GitHub. You can find it here:
https://github.com/szymon6927/hexagonal-architecture-python

This project is a simplified gym management software. We have clients, gym classes and gym passes.

The project consists of four modules:

building_blocks - contains all utilities used across different modules
clients - module responsible for clients. There is no complicated business logic here, but we have a lot of integrations that's why we use hexagonal architecture here
gym_classes - simple CRUD responsible for management of gym classes. No business logic here, therefore there is no need to use hexagonal architecture here
gym_passes - management of gym passes, which is the core of our business. This is where our business is making money

There is no fancy structure for the gym_classes module. The answer is simple - there is no need for that.
We should use hexagonal architecture only where it's really needed.

I developed this example specifically to show you that using hexagonal architecture doesn't mean using it everywhere.
Not every module has to have it.

This is also visible in tests.

There are no unit tests for the gym classes module because unit tests for it would make no sense.
All we do here are CRUD operations.

When it comes to modules that were built using hexagonal architecture. They follow such structure:

application - contains our use cases together with DTOs and other classes that are responsible for coordinating the business processes
domain - contains all domain objects like entities, value objects, etc. The objects there are not anemic, but they follow the principle of the rich domain model which means that we encapsulate data and behaviours together. You can check it by looking at src/gym_passes/domain/gym_pass.py
infrastructure - contains output adapters, which are objects responsible for communication with external world, for example database
bootstrap.py - contains the definition of our DI container. I wrote a separate article about the DI. If you have not heard about a technique called "dependency injection" you can read about it here
controllers.py - definition of our REST endpoints
facade.py (optional) - if your module needs to expose some behaviours for other modules then I recommend using a facade to achieve it. It is the public API of your module

Pros and Cons

Pros:

testability - you can test your domain logic without any dependencies
ability to postpone some decisions - when your project starts you don't have to know which database you will use. You can make this decision later on
it supports easy technology change - you can change your REST adapter to a gRPC adapter without touching your business logic/core domain

Cons:

if your project has multiple adapters it means that you will have more integration tests. This may impact the total execution time of your test suite
it’s harder to navigate through the project with such architecture. You will mostly encounter interfaces instead of real implementation
additional effort is needed to configure the adapters. You need to have some mechanism that will specify: this adapter is on prod env, this on local env, etc

When to use / when not to use

If you are wondering about cases to which hexagonal architecture suits the best, I prepared a simple heuristic for you.

Use if:

your project has complex and frequently changing business domain
you need to implement the core domain of your application

Don't use if:

your project or module has a CRUD-like complexity - in such a scenario, there will be no benefits of hexagonal architecture, and it will be overengineering

Additional resources

Summary

I hope you enjoyed it.
As with everything in our industry, hexagonal architecture is not a silver bullet.
It will not solve all the problems for you,
but it's a great way to improve things in your project especially when it comes to testability or the possibility to postpone some decisions.
If applied correctly it may improve your project and make your life easier.

Let me know what you think, I would love to hear your opinion.
Also, if you have experience with hexagonal architecture let me know what your thoughts/feelings are about it.