Forem: Piotr Tobiasz

Gopher Pythonista #1: Moving From Python To Go

Piotr Tobiasz — Thu, 27 Jul 2023 07:03:58 +0000

Throughout my career, I have primarily focused on developing backend systems using Python. Despite dabbling in front-end work, I always found myself gravitating back toward API development. In order to expand my skill set, I have experimented with writing backend applications using different programming languages such as JavaScript/Typescript and Rust. I have tested various technologies and frameworks, some of which I found more enjoyable than others. Nonetheless, there remains one language that I have yet to explore: Go.

When I first started using Golang, it felt a little unfamiliar due to its simplicity, unique error handling, and some unusual design choices like struct tags. However, even before getting fully accustomed to these "quirks", I quickly realized several things about Go that I genuinely appreciate. Interestingly, these same features can be seen as challenges in my primary programming language, Python.

In this article, I will share my initial thoughts on using Go as a Python engineer. I will highlight the aspects of Go that I appreciate, which may not be as prominent in Python. These include:

built-in tools
packaging
backend focus
setting up Docker
performance

By the end of this article, you will understand why Go is an excellent option for backend applications and how it compares to Python in terms of convenience.

Tooling

When it comes to building software, the main task is writing code. However, there are several potential problems that can arise during this process. These can include:

incorrect formatting
bugs
common code issues

As engineers and humans, it's impossible to guarantee that our work is completely flawless. This is where automated tools like linters, unit tests, and static analysis can be incredibly useful in providing additional assistance.

When working with Python, formatting can be a challenge. While the most popular tool for formatting is black, it requires installation and configuration. However, configuration can be difficult as there are multiple ways to do it, with pyproject.toml being the default option according to PEP 518, but not all tools support it.

In terms of linting, prior to the release of ruff in 2022, Python developers had to rely on flake8 and its various plugins, which were inconvenient to configure. Thankfully, ruff now supports many features and can be configured via the pyproject.toml file.

Similarly, if you want more powerful testing capabilities than what is offered by the built-in unittest, you'll need to install and configure additional tools and addons. It's important to ensure that all of these components work together properly in your CI system as well.

When working with Go, the development experience can be quite different. Upon installation, you will have access to a wide range of tools to aid your local development.

Compared to Python, formatting rules in Go are less strict. For example, there is no fixed line length. However, Go has a built-in formatting tool that can be accessed through the go fmt command, eliminating the need for additional installation or configuration.

Another useful tool in Go is the go vet command, which helps to identify common coding mistakes such as unreachable code or useless comparisons. In addition, external linters like staticcheck can be used to detect bugs and performance issues with ease.

For testing purposes, Go provides a go test command that automatically discovers tests within your application and supports features such as caching and code coverage. However, if you require more advanced testing capabilities such as suites or mocking, you will need to install a toolkit like testify. Overall, while Go provides a highly effective testing experience, it's worth noting that writing tests in Python using pytest is arguably one of the most enjoyable testing experiences I have encountered across all programming languages.

It's also important to remember that there's no need to use type-checking tools such as mypy with Go, as it's a compiled and statically-typed language.

In summary, once you've installed Go, you'll have access to the necessary tools to support your code writing efforts. The next step is to focus on packaging and managing dependencies.

Packaging

Managing dependencies is crucial when building software. While Golang has an extensive standard library, more complex projects often require external packages. Luckily, Go provides excellent tools for effectively managing dependencies.

Go projects come with two files: go.mod and go.sum. The go.mod file is particularly crucial as it lists all the direct and indirect dependencies along with their versions. The go.sum file is automatically generated to serve as a checksum of the dependencies.

To install a package, you use the go get <package> command. The good news is that this command saves the installed packages in the $GOPATH/pkg/mod directory, so there is no need to manually create virtual environments or similar things.

In addition, you can use the go mod tidy command to ensure consistency between the go.mod file and the source code in the module. This command will add any missing module requirements needed to build the current module's packages and dependencies. If there are any unused dependencies, go mod tidy will remove them from go.mod.

Finally, publishing packages with Go is incredibly simple. To install them, you typically just need to use the repository link, such as go get github.com/tobias-piotr/gouth0. So, to publish your own package, all you have to do is properly set up your module in the go.mod file, create a public repository, and request the package at https://pkg.go.dev/. After a short period of time, your package will be available in the Go module index. That's it.

Now compare this with Python. By default, Python doesn't even have a checksum file. If you want more advanced features, you'll need to install a tool like poetry, which has many external dependencies. Managing project dependencies can be done through a file, but you'll need to create and maintain it yourself. Compared to Go, publishing packages to PyPI can also feel inconvenient.

Without further criticism of Python packaging, Go's packaging feels much simpler and more convenient. The next chapter will cover how Go's setup is beneficial when working with Docker.

Docker

When it comes to building web applications in the modern world, Docker is usually involved. It simplifies the process of setting up a project locally and deploying it to production.

Creating a Dockerfile for a Golang project is remarkably easy for two primary reasons: the Go tooling mentioned earlier and the fact that Go is a compiled language. Additionally, the combination of these two factors leads to a very standardized method of setting up a project. This means that if you copy a Dockerfile from one Golang project to another, there is a high probability that it will work without any issues. This is quite different from the situation in Python.

With all of this in mind, take a look at a Dockerfile from one of my Go projects:

FROM golang:1.20.6-alpine AS builder

EXPOSE 8080

WORKDIR /app

COPY go.mod go.sum ./
RUN go mod download

COPY . .

RUN CGO_ENABLED=0 GOOS=linux go build -o /main

# Local development
FROM builder as dev

ENV DEBUG=true

# Install air for hot reloading
RUN go install github.com/cosmtrek/air@latest

ENTRYPOINT ["air"]

# Production
FROM gcr.io/distroless/base-debian11 AS prod

WORKDIR /

COPY --from=builder /main /main

USER nonroot:nonroot

ENTRYPOINT ["/main"]

Without diving too deep into the Docker syntax, this config simply downloads dependencies, compiles code, and runs the binary file. It's important to note that Go, being a compiled language, improves this process. For example, you can use a tool like air to use the binary file and enable hot reloading while developing locally. Additionally, in production, a smaller base image can be used because no OS dependencies are required to start the binary file. This significantly reduces the image size of your app.

Starting a new project in Go is great because it's easy to set up a fully working and optimized Dockerfile. With tooling, packaging and containerizing covered, we can finally focus on the language's features.

Backend Focus

When I first started working with Go, I had doubts about its suitability for backend development. However, it didn't take long for me to realize that Go is actually designed with backend development in mind.

One of the main things that makes Go well-suited for backend development is its standard library. This library includes a variety of useful packages for tasks such as encoding, cryptography, and database management. Most importantly, it also includes packages for building HTTP servers, which means you can create a semi-sophisticated API without needing to install any additional dependencies. For instance, I was able to develop an authentication API for interacting with Auth0 using just one external package (for structured logging). You can view the code here: goloxy.

On top of that, there are many amazing packages available for developing web applications, including:

web frameworks: Gin, Fiber, chi, Echo
validation: validate, ozzo-validation, Valgo
structured logging: zap, logrus, zerolog
database interactions: GORM, sqlx, sqlc, Ent

My preferred combination is Echo/chi as a web framework, Valgo for validation, zap for structured logging, sqlx for database interactions, and Fx for dependency injection. This setup has been effective for most of my projects, though it may vary depending on specific requirements.

To highlight a difference with Python, configuring structured logging can be a challenge, even with a great framework like FastAPI. In an Echo app, I configure the zap logger in the following way:

e.Use(middleware.RequestLoggerWithConfig(middleware.RequestLoggerConfig{
    LogProtocol:  true,
    LogMethod:    true,
    LogURI:       true,
    LogStatus:    true,
    LogRequestID: true,
    LogRemoteIP:  true,
    LogLatency:   true,
    LogError:     true,
    LogValuesFunc: func(c echo.Context, v middleware.RequestLoggerValues) error {
        logger.Info(
            fmt.Sprintf("%s %s %s %v", v.Protocol, v.Method, v.URI, status),
            zap.String("protocol", v.Protocol),
            zap.String("method", v.Method),
            zap.String("uri", v.URI),
            zap.Int("status", status),
            zap.String("request_id", v.RequestID),
            zap.String("remote_ip", v.RemoteIP),
            zap.String("latency", v.Latency.String()),
            zap.Error(v.Error),
        )
        return nil
    },
}))

In addition to creating a logger instance, which is usually just a few lines, this is all that is required to have complete control over the logger for each request. An example of setting up structured logging in a FastAPI app can be found using the structlog package at this link. This is surely overwhelming, even when you are fully aware of what's going on there.

To summarize, Go has excellent built-in and externally installable solutions for developing web apps. However, implementation is just the beginning. The next chapter will cover the performance of Golang code.

Performance

When examining the most popular uses of Go, examples include proxies, infrastructure management, and "typical" web development. One of the reasons for its popularity is the exceptional performance of Go, especially when compared to Python, which is known to have weaker performance.

There are numerous factors that contribute to Go's superior performance. such as:

lightweight design, which reduces runtime overhead
efficient garbage collection
impressive concurrency model with Goroutines
strong typing and static analysis

These are just a few examples of the reasons why Go is so well-performing. All of these features enable the creation of highly optimized applications that can efficiently handle large workloads and easily scale even further.

It's possible that one might argue that performance isn't crucial in many applications. In such cases, you could simply select Python, take code from a previous FastAPI project, base it around asyncio, and it should perform well enough. This argument may be valid. However, performance isn't just about speed; it also relates to resources.

You might not notice speed differences when comparing an app written in Golang to one in Python - the differences are in nanoseconds. However, you will definitely see the difference in resource usage. The app developed in Go will use significantly fewer resources. This will be evident through lower costs. I recommend watching Ben Davis' video where he explains how rewriting a SAAS project in Go resulted in reduced costs.

When it comes to performance, Python doesn't have the best reputation. This is largely due to the GIL, dynamic typing, and awkward concurrency model. While there are optimizations available, such problems are less frequent in Go. Therefore, when prioritizing performance, Go may be the better option.

Summary

Based on my experience with various technologies, I have some personal favorites. However, when building software, it is important to consider multiple factors before choosing the best tech stack. For backend development, Go stands out as an excellent option, especially when performance and cost are crucial. Nevertheless, it's not without any flaws. I believe that in modern Python, FastAPI and Pydantic are its strongest features for web development. They offer features like dependency injection, Swagger documentation, exceptional validation, and error handling right out of the box. I also find pytest to be the most well-designed test writing tool I've encountered. In summary, Go lacks some of the drawbacks that Python has, but Python has a wealth of features and libraries for app development. Ultimately, it depends on your specific needs and trade-offs.

Sigma Python #3: Descriptors

Piotr Tobiasz — Wed, 26 Jul 2023 07:19:52 +0000

Python offers a range of powerful features for object-oriented programming, including descriptors. These enable you to define how class attribute modification and access occur, making it useful for enforcing value constraints, implementing computed properties, and customizing attribute access.

This article covers the fundamentals of descriptors and provides guidance on how to implement them in your Python code. It addresses the following questions:

what is the definition of a descriptor
how to create a descriptor from scratch
how the descriptors actually work
what can be built using descriptors

Descriptors may sound unfamiliar to you, but they are widely used in popular packages such as Django and SQLAlchemy. This article will provide you with a clear understanding of descriptors and enable you to develop your own.

Anatomy of a Descriptor

A descriptor is a type of class that includes either of the __get__, __set__, or __delete__ methods. When an instance of this class is used as an attribute in another class, it is referred to as a managed attribute. To illustrate, here is an example of a basic descriptor class and how it can be used as a managed attribute:

class PositiveInt:
    def __init__(self, name: str) -> None:
        self.name = name

    def __set__(self, obj: object, value: int) -> None:
        if value <= 0:
            raise ValueError(f"Value of '{self.name}' must be a positive int")
        obj.__dict__[self.name] = value

    def __get__(self, obj: object, cls: type) -> int:
        return obj.__dict__[self.name]


class SomeClass:
    x = PositiveInt("x")
    y = PositiveInt("y")

    def __init__(self, x: int, y: int) -> None:
        self.x = x
        self.y = y


some_object = SomeClass(1, 2)
print(some_object.x)
print(some_object.y)
some_object.y = 0

The PositiveInt descriptor is used to set integer attributes in a class. Its primary function is to ensure that the number passed is greater than zero. This check is performed within the __set__ magic method, which is triggered whenever a new value is assigned to an attribute. If the value is valid, it is stored in the instance's dictionary of writable attributes, called __dict__.

There are a few important things to note about the PositiveInt structure. Firstly, it also includes the __get__ method. This is necessary in order to retrieve the value from the original class. However, the __del__ method is not included. It was needed for this particular example, and it is acceptable for a descriptor class to not implement it. In fact, a descriptor class only needs to include one of the three methods. Finally, it is worth noting that inheritance is not required in order to implement a descriptor. This is because descriptors are protocol-based feature.

When the code is executed, the output is as follows:

1
2
Traceback (most recent call last):
    # trunkated
ValueError: Value of 'y' must be a positive int

The final workflow appears like there are regular attributes, but in reality, the power of Python Data Model lies in its ability to handle complex implementation behind the scenes. In this case, the descriptor manages the assignment and retrieval of values. The values are initially set in the __init__ method, and then, an error occurs due to the last assignment.

Upon examining PositiveInt, it may seem more practical to store the value within the descriptor rather than in the original class. This can be achieved by adding self.__dict__[self.name] = value in the __set__ method. To investigate such a case, take a look at the "improved" example:

class NegativeInt:
    def __init__(self, name: str) -> None:
        self.name = name

    def __set__(self, obj: object, value: int) -> None:
        if value >= 0:
            raise ValueError(f"Value of '{self.name}' must be a negative int")
        self.__dict__[self.name] = value

    def __get__(self, obj: object, cls: type) -> int:
        return self.__dict__[self.name]


class DifferentClass:
    x = NegativeInt("x")


some_object = DifferentClass()
some_object.x = -5
print(some_object.x)

different_object = DifferentClass()
different_object.x = -6
print(different_object.x)

print(some_object.x)

The NegativeInt descriptor functions similarly to PositiveInt, but in the opposite direction. It also operates on data that is stored within the descriptor class. The following result is produced when executing this code:

-5
-6
-6

It appears that changing the value of different_object.x to -6 also changed the value of the some_object attribute. This is because descriptors are class attributes that are initialized only once during import time. Therefore, both objects of the DifferentClass share the same descriptor object. To ensure proper functioning, the descriptor object must be smart enough to work with the data of the object it belongs to.

In previous examples, it was necessary to manually set the name and retrieve the value using that name. However, this approach can lead to repetitive code in all descriptors. Thankfully, the __set_name__ magic method provides a solution. When the interpreter calls this method on all descriptors in the class, it eliminates the need for repetitive __init__ and __get__ methods. For instance:

class PositiveInt:
    def __set_name__(self, owner: type, name: str):
        self.name = name

    def __set__(self, obj: object, value: int) -> None:
        if value <= 0:
            raise ValueError(f"Value of '{self.name}' must be a positive int")
        obj.__dict__[self.name] = value


class SomeClass:
    x = PositiveInt()
    y = PositiveInt()

    def __init__(self, x: int, y: int) -> None:
        self.x = x
        self.y = y


some_object = SomeClass(1, 2)
print(some_object.x)
print(some_object.y)

The PositiveInt descriptor remains the same as before, but it no longer contains the __init__ method. Instead, it includes the __set_name__ method, which will be called with the x and y names in this particular case. This modification means that the __get__ method implementation is also unnecessary, as the attribute name is identical in both classes, and the value can be retrieved from the SomeClass instance.

Now with the anatomy of descriptors covered, you should have a solid understanding of how they function. The next section will go over the implementation of a more real-world scenario.

JSON Attribute

Descriptors are an embodiment of encapsulation. They hide implementation and complexity, making them perfect for validation, caching, and interacting with non-in-memory storage like databases and file systems. For example, the JSONAttribute descriptor works with JSON files while the target class declares attributes as usual.

import json
from typing import Any


class JSONAttribute:
    """A descriptor for working with JSON files."""

    def __set_name__(self, owner: type, name: str):
        self.name = name
        self.path = f"{owner.__name__}.json"

    def __set__(self, obj: object, value: Any) -> None:
        """Set the value in the JSON file."""
        try:
            f = open(self.path, "r+")
            content = json.load(f)
        except FileNotFoundError:
            f = open(self.path, "a+")
            content = {}

        content[self.name] = value
        f.seek(0)
        json.dump(content, f, indent=4)
        f.close()

    def __get__(self, obj: object, owner: type) -> Any:
        """Get the value from the JSON file."""
        try:
            f = open(self.path, "r")
        except FileNotFoundError:
            f = open(self.path, "a+")
            json.dump({}, f)
            return None

        value = json.load(f).get(self.name)
        f.close()
        return value


class JSONReaderWriter:
    x = JSONAttribute()
    y = JSONAttribute()


rw = JSONReaderWriter()
rw.x = 1
rw.y = 2
print(rw.x, rw.y)

Without delving too deeply into the details of json implementation, JSONAttribute has three core functionalities:

creating a JSON file named after the class it is used in
setting the value of an attribute inside the JSON file
retrieving the value of the attribute from the JSON file

When executed, this code, of course, prints 1 2, but more importantly, the JSONReaderWriter.json file is created and the values are stored inside it. This is impressive because the end user of this code sees the values coming from the instance of the class, but they are actually nowhere to be found within it.

You have now seen a real-world use case and should be able to implement a descriptor for your own needs. The final chapter covers some of the most popular descriptor examples.

Descriptors Are Ubiquitous

As mentioned previously, even if you haven't been familiar with descriptors, you've probably used them before, without realizing it. One common example is the property decorator, which is a descriptor factory that creates a descriptor. The decorated method acts as a getter, but you can also implement setter and deleter methods.

Another popular example includes models and fields in Django. In this case, descriptors are a way to abstract away the details of working with a database, or a file system. However, Django uses metaclasses on top of descriptors, making the process more complex.

An example of a more modern Python library is Pydantic fields. They behave similarly to Django in that they also "hide" field behavior. However, in this case, descriptors are used to provide additional information about the field and attach necessary validation.

Summary

Descriptors are one of the features that make Python's OOP so unique. They allow for powerful customization of attribute access and manipulation. This article explored the topic of descriptors, covering how to build them, what they can be used for, and what descriptors developers are encountering on a daily basis. Hopefully, you gained some valuable insights and you are able to work with descriptors in your projects.

Sources

Code examples used in the article can be found here: link.

The biggest inspiration for these articles and source of my knowledge is the Fluent Python book by Luciano Ramalho. I highly encourage you to check it out; you will not be disappointed.

Patterns and Practices for using SQLAlchemy 2.0 with FastAPI

Piotr Tobiasz — Tue, 25 Jul 2023 18:34:23 +0000

While Django and Flask remain the first choices for many Python engineers, FastAPI has already been recognized as an undeniably reliable pick. It is a highly flexible, well-optimized, structured framework that gives the developer endless possibilities for building backend applications.

Working with databases is an essential aspect of most backend applications. As a result, the ORM plays a critical role in the backend code. However, unlike Django, FastAPI does not have an ORM built-in. It is entirely the developer's responsibility to select a suitable library and integrate it into the codebase.

Python engineers widely consider SQLAlchemy to be the most popular ORM available. It's a legendary library that's been in use since 2006 and has been adopted by thousands of projects. In 2023, it received a major update to version 2.0. Similar to FastAPI, SQLAlchemy provides developers with powerful features and utilities without forcing them to use them in a specific way. Essentially, it's a versatile toolkit that empowers developers to use it however they see fit.

FastAPI and SQLAlchemy are a match made in heaven. They are both reliable, performant, and modern technologies, which enable the creation of powerful and unique applications. This article explores creating a FastAPI backend application that utilizes SQLAlchemy 2.0 as the ORM. The content covers:

building models using Mapped and mapped_column
defining an abstract model
handling database session
using the ORM
creating a common repository class for all models
preparing a test setup and adding tests

Afterward, you will be able to combine the FastAPI application with SQLAlchemy ORM easily. Additionally, you will get familiar with best practices and patterns for creating well-structured, robust, and performant applications.

Prerequisites

Code examples included in the article come from the alchemist project, which is a basic API for creating and reading ingredient and potion objects. The main focus of the article is to explore the combination of FastAPI and SQLAlchemy. It doesn't cover other topics, such as:

configuring Docker setup
starting the uvicorn server
setting up linting

If you are interested in these topics, you can explore them on your own by examining the codebase. To access the code repository of the alchemist project, please follow this link here. Additionally, you can find the file structure of the project below:

alchemist
├─ alchemist
│  ├─ api
│  │  ├─ v1
│  │  │  ├─ __init__.py
│  │  │  └─ routes.py
│  │  ├─ v2
│  │  │  ├─ __init__.py
│  │  │  ├─ dependencies.py
│  │  │  └─ routes.py
│  │  ├─ __init__.py
│  │  └─ models.py
│  ├─ database
│  │  ├─ __init__.py
│  │  ├─ models.py
│  │  ├─ repository.py
│  │  └─ session.py
│  ├─ __init__.py
│  ├─ app.py
│  └─ config.py
├─ requirements
│  ├─ base.txt
│  └─ dev.txt
├─ scripts
│  ├─ create_test_db.sh
│  ├─ migrate.py
│  └─ run.sh
├─ tests
│  ├─ conftest.py
│  └─ test_api.py
├─ .env
├─ .gitignore
├─ .pre-commit-config.yaml
├─ Dockerfile
├─ Makefile
├─ README.md
├─ docker-compose.yaml
├─ example.env
└─ pyproject.toml

Although the tree may seem large, some of the contents are not relevant to the main point of this article. Additionally, the code may appear simpler than what is necessary in certain areas. For instance, the project lacks:

production stage in the Dockerfile
alembic setup for migrations
subdirectories for tests

This was done intentionally to reduce complexity and avoid unnecessary overhead. However, it is important to keep these factors in mind if dealing with a more production-ready project.

API Requirements

When beginning to develop an app, it's critical to consider the models that your app will use. These models will represent the objects and entities that your app will work with and will be exposed in the API. In the case of the alchemist app, there are two entities: ingredients and potions. The API should allow for creating, and retrieving these entities. The alchemist/api/models.py file contains the models that will be used in the API:

import uuid

from pydantic import BaseModel, Field


class Ingredient(BaseModel):
    """Ingredient model."""

    pk: uuid.UUID
    name: str

    class Config:
        orm_mode = True


class IngredientPayload(BaseModel):
    """Ingredient payload model."""

    name: str = Field(min_length=1, max_length=127)


class Potion(BaseModel):
    """Potion model."""

    pk: uuid.UUID
    name: str
    ingredients: list[Ingredient]

    class Config:
        orm_mode = True


class PotionPayload(BaseModel):
    """Potion payload model."""

    name: str = Field(min_length=1, max_length=127)
    ingredients: list[uuid.UUID] = Field(min_items=1)

The API will be returning Ingredient and Potion models. Setting orm_mode to True in the configuration will make it easier to work with the SQLAlchemy objects in the future. Payload models will be utilized for creating new objects.

The use of pydantic makes the classes more detailed and clear in their roles and functions. Now, it's time to create the database models.

Declaring Models

A model is essentially a representation of something. In the context of APIs, models represent what the backend expects in the request body and what it will return in the response data. Database models, on the other hand, are more complex and represent the data structures stored in the database and the relationship types between them.

The alchemist/database/models.py file contains models for ingredient and potion objects:

import uuid

from sqlalchemy import Column, ForeignKey, Table, orm
from sqlalchemy.dialects.postgresql import UUID


class Base(orm.DeclarativeBase):
    """Base database model."""

    pk: orm.Mapped[uuid.UUID] = orm.mapped_column(
        primary_key=True,
        default=uuid.uuid4,
    )


potion_ingredient_association = Table(
    "potion_ingredient",
    Base.metadata,
    Column("potion_id", UUID(as_uuid=True), ForeignKey("potion.pk")),
    Column("ingredient_id", UUID(as_uuid=True), ForeignKey("ingredient.pk")),
)


class Ingredient(Base):
    """Ingredient database model."""

    __tablename__ = "ingredient"

    name: orm.Mapped[str]


class Potion(Base):
    """Potion database model."""

    __tablename__ = "potion"

    name: orm.Mapped[str]
    ingredients: orm.Mapped[list["Ingredient"]] = orm.relationship(
        secondary=potion_ingredient_association,
        backref="potions",
        lazy="selectin",
    )

Every model in SQLAlchemy starts with the DeclarativeBase class. Inheriting from it allows building database models that are compatible with Python type checkers.

It's also a good practice to create an abstract model—Base class in this case—that includes fields required in all models. These fields include the primary key, which is a unique identifier of every object. The abstract model often also stores the creation and update dates of an object, which are set automatically, when an object is created or updated. However, the Base model will be kept simple.

Moving on to the Ingredient model, the __tablename__ attribute specifies the name of the database table, while the name field uses the new SQLAlchemy syntax, allowing model fields to be declared with type annotations. This concise and modern approach is both powerful and advantageous for type checkers and IDEs, as it recognizes the name field as a string.

Things get more complex in the Potion model. It also includes __tablename__ and name attributes, but on top of that, it stores the relationship to ingredients. The use of Mapped[list["Ingredient"]] indicates that the potion can contain multiple ingredients, and in this case, the relationship is many-to-many (M2M). This means that a single ingredient can be assigned to multiple potions.

M2M requires additional configuration, usually involving the creation of an association table that stores the connections between the two entities. In this case, the potion_ingredient_association object stores only the identifiers of the ingredient and potion, but it could also include extra attributes, such as the amount of a specific ingredient needed for the potion.

The relationship function configures the relationship between the potion and its ingredients. The lazy argument specifies how related items should be loaded. In other words: what should SQLAlchemy do with related ingredients when you are fetching a potion. Setting it to selectin means that ingredients will be loaded with the potion, eliminating the need for additional queries in the code.

Building well-designed models is crucial when working with an ORM. Once this is done, the next step is establishing the connection with the database.

Session Handler

When working with a database, particularly when using SQLAlchemy, it is essential to understand the following concepts:

dialect
engine
connection
connection pool
session

Out of all these terms, the most important one is the engine. According to the SQLAlchemy documentation, the engine object is responsible for connecting the Pool and Dialect to facilitate database connectivity and behavior. In simpler terms, the engine object is the source of the database connection, while the connection provides high-level functionalities like executing SQL statements, managing transactions, and retrieving results from the database.

A session is a unit of work that groups related operations within a single transaction. It is an abstraction over the underlying database connections and efficiently manages connections and transactional behavior.

Dialect is a component that provides support for a specific database backend. It acts as an intermediary between SQLAlchemy and the database, handling the details of communication. The alchemist project uses Postgres as the database, so the dialect must be compatible with this specific database type.

The final question mark is the connection pool. In the context of SQLAlchemy, a connection pool is a mechanism that manages a collection of database connections. It is designed to improve the performance and efficiency of database operations by reusing existing connections rather than creating new ones for each request. By reusing connections, the connection pool reduces the overhead of establishing new connections and tearing them down, resulting in improved performance.

With that knowledge covered, you can now take a look at alchemist/database/session.py file, which contains a function that will be used as a dependency for connecting to the database:

from collections.abc import AsyncGenerator

from sqlalchemy import exc
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from alchemist.config import settings


async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
    engine = create_async_engine(settings.DATABASE_URL)
    factory = async_sessionmaker(engine)
    async with factory() as session:
        try:
            yield session
            await session.commit()
        except exc.SQLAlchemyError as error:
            await session.rollback()
            raise

The first important detail to notice is that the function get_db_session is a generator function. This is because the FastAPI dependency system supports generators. As a result, this function can handle both successful and failed scenarios.

The first two lines of the get_db_session function create a database engine and a session. However, the session object can also be used as a context manager. This gives you more control over potential exceptions and successful outcomes.

Although SQLAlchemy handles the closing of connections, it is a good practice to explicitly declare how to handle the connection after it's done. In the get_db_session function, the session is committed if everything goes well and rolled back if an exception is raised.

It's important to note that this code is built around the asyncio extension. This feature of SQLAlchemy allows the app to interact with the database asynchronously. It means that requests to the database won't block other API requests, making the app way more efficient.

Once the models and connection are set up, the next step is to ensure that the models are added to the database.

Quick Migrations

SQLAlchemy models represent the structures of a database. However, simply creating them does not result in immediate changes to the database. To make changes, you must first apply them. This is typically done using a migration library such as alembic, which tracks every model and updates the database accordingly.

Since no further changes to the models are planned in this scenario, a basic migration script will suffice. Below is an example code from the scripts/migrate.py file.

import asyncio
import logging

from sqlalchemy.ext.asyncio import create_async_engine

from alchemist.config import settings
from alchemist.database.models import Base

logger = logging.getLogger()


async def migrate_tables() -> None:
    logger.info("Starting to migrate")

    engine = create_async_engine(settings.DATABASE_URL)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    logger.info("Done migrating")


if __name__ == "__main__":
    asyncio.run(migrate_tables())

To put it simply, the migrate_tables function reads the structure of models and recreates it in the database using the SQLAlchemy engine. To run this script, use the python scripts/migrate.py command.

The models are now present both in the code and in the database and get_db_session can facilitate interactions with the database. You can now begin working on the API logic.

API with the ORM

As mentioned previously, the API for ingredients and potions is meant to support three operations:

creating objects
listing objects
retrieving objects by ID

Thanks to the prior preparations, all these features can already be implemented with SQLAlchemy as the ORM and FastAPI as the web framework. To begin, review the ingredients API located in the alchemist/api/v1/routes.py file.

import uuid

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from alchemist.api import models
from alchemist.database import models as db_models
from alchemist.database.session import get_db_session

router = APIRouter(prefix="/v1", tags=["v1"])


@router.post("/ingredients", status_code=status.HTTP_201_CREATED)
async def create_ingredient(
    data: models.IngredientPayload,
    session: AsyncSession = Depends(get_db_session),
) -> models.Ingredient:
    ingredient = db_models.Ingredient(**data.dict())
    session.add(ingredient)
    await session.commit()
    await session.refresh(ingredient)
    return models.Ingredient.from_orm(ingredient)


@router.get("/ingredients", status_code=status.HTTP_200_OK)
async def get_ingredients(
    session: AsyncSession = Depends(get_db_session),
) -> list[models.Ingredient]:
    ingredients = await session.scalars(select(db_models.Ingredient))
    return [models.Ingredient.from_orm(ingredient) for ingredient in ingredients]


@router.get("/ingredients/{pk}", status_code=status.HTTP_200_OK)
async def get_ingredient(
    pk: uuid.UUID,
    session: AsyncSession = Depends(get_db_session),
) -> models.Ingredient:
    ingredient = await session.get(db_models.Ingredient, pk)
    if ingredient is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Ingredient does not exist",
        )
    return models.Ingredient.from_orm(ingredient)

Under the /ingredients API, there are three routes available. The POST endpoint takes an ingredient payload as an object from a previously created model and a database session. The get_db_session generator function initializes the session and enables database interactions.

In the actual function body, there are five steps taking place:

An ingredient object is created from the incoming payload.
The add method of the session object adds the ingredient object to the session tracking system and marks it as pending for insertion into the database.
The session is committed.
The ingredient object is refreshed to ensure its attributes match the database state.
The database ingredient instance is converted to the API model instance using the from_orm method.

For a quick test, a simple curl can be executed against the running app:

curl -X 'POST' \
  'http://localhost:8000/api/v1/ingredients' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{"name": "Salty water"}'

In the response, there should be an ingredient object that has an ID coming from the database:

{
  "pk":"2eb255e9-2172-4c75-9b29-615090e3250d",
  "name":"Salty water"
}

Although SQLAlchemy's multiple layers of abstraction may seem unnecessary for a simple API, they keep the ORM details separated and contribute to SQLAlchemy's efficiency and scalability. When combined with asyncio, the ORM features perform exceptionally well in the API.

The remaining two endpoints are less complex and share similarities. One part that deserves a deeper explanation is the use of the scalars method inside the get_ingredients function. While querying the database using SQLAlchemy, the execute method is often used with a query as the argument. While execute method returns row-like tuples, scalars returns ORM entities directly, making the endpoint cleaner.

Now, consider the potions API, in the same file:

@router.post("/potions", status_code=status.HTTP_201_CREATED)
async def create_potion(
    data: models.PotionPayload,
    session: AsyncSession = Depends(get_db_session),
) -> models.Potion:
    data_dict = data.dict()
    ingredients = await session.scalars(
        select(db_models.Ingredient).where(
            db_models.Ingredient.pk.in_(data_dict.pop("ingredients"))
        )
    )
    potion = db_models.Potion(**data_dict, ingredients=list(ingredients))
    session.add(potion)
    await session.commit()
    await session.refresh(potion)
    return models.Potion.from_orm(potion)


@router.get("/potions", status_code=status.HTTP_200_OK)
async def get_potions(
    session: AsyncSession = Depends(get_db_session),
) -> list[models.Potion]:
    potions = await session.scalars(select(db_models.Potion))
    return [models.Potion.from_orm(potion) for potion in potions]


@router.get("/potions/{pk}", status_code=status.HTTP_200_OK)
async def get_potion(
    pk: uuid.UUID,
    session: AsyncSession = Depends(get_db_session),
) -> models.Potion:
    potion = await session.get(db_models.Potion, pk)
    if potion is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Potion does not exist",
        )
    return models.Potion.from_orm(potion)

The GET endpoints for potions are identical to those for ingredients. However, the POST function requires additional code. This is because creating potions involves including at least one ingredient ID, which means that the ingredients must be fetched and linked to the newly created potion. To achieve this, the scalars method is used again, but this time with a query that specifies the IDs of the fetched ingredients. The remaining part of the potion creation process is identical to that of the ingredients.

To test the endpoint, again a curl command can be executed.

curl -X 'POST' \
  'http://localhost:8000/api/v1/potions' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{"name": "Salty soup", "ingredients": ["0b4f1de5-e780-418d-a74d-927afe8ac954"}'

It results in the following response:

{
  "pk": "d4929197-3998-4234-a5f7-917dc4bba421",
  "name": "Salty soup",
  "ingredients": [
    {
      "pk": "0b4f1de5-e780-418d-a74d-927afe8ac954",
      "name": "Salty water"
    }
  ]
}

It's important to notice that each ingredient is represented as a complete object within the potion, thanks to the lazy="selectin" argument specified in the relationship.

The APIs are functional, but there is a major issue with the code. While SQLAlchemy gives you the freedom to interact with the database as you please, it does not offer any high-level "manager" utility similar to Django's Model.objects. As a result, you will need to create it yourself, which is essentially the logic used in the ingredient and potion APIs. However, if you keep this logic directly in the endpoints without extracting it into a separate space, you will end up with a lot of duplicated code. Additionally, making changes to the queries or models will become increasingly difficult to manage.

The upcoming chapter introduces the repository pattern: an elegant solution for extracting ORM code.

Repository

The repository pattern allows to abstract away the details of working with the database. In the case of using SQLAlchemy, such as in the example of the alchemist, the repository class would be responsible for managing multiple models and interacting with the database session.

Take a look at the following code from the alchemist/database/repository.py file:

import uuid
from typing import Generic, TypeVar

from sqlalchemy import BinaryExpression, select
from sqlalchemy.ext.asyncio import AsyncSession

from alchemist.database import models

Model = TypeVar("Model", bound=models.Base)


class DatabaseRepository(Generic[Model]):
    """Repository for performing database queries."""

    def __init__(self, model: type[Model], session: AsyncSession) -> None:
        self.model = model
        self.session = session

    async def create(self, data: dict) -> Model:
        instance = self.model(**data)
        self.session.add(instance)
        await self.session.commit()
        await self.session.refresh(instance)
        return instance

    async def get(self, pk: uuid.UUID) -> Model | None:
        return await self.session.get(self.model, pk)

    async def filter(
        self,
        *expressions: BinaryExpression,
    ) -> list[Model]:
        query = select(self.model)
        if expressions:
            query = query.where(*expressions)
        return list(await self.session.scalars(query))

The DatabaseRepository class holds all the logic that was previously included in the endpoints. The difference is that it allows for the specific model class to be passed in the __init__ method, making it possible to reuse the code for all models instead of duplicating it in each endpoint.

Furthermore, the DatabaseRepository uses Python generics, with the Model generic type bounded to the abstract database model. This allows for the repository class to benefit more from static type checking. When used with a specific model, the return types of the repository methods will reflect this specific model.

Because the repository needs to use the database session, it must be initialized along with the get_db_session dependency. Consider the new dependency in the alchemist/api/v2/dependencies.py file.

from collections.abc import Callable

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from alchemist.database import models, repository, session


def get_repository(
    model: type[models.Base],
) -> Callable[[AsyncSession], repository.DatabaseRepository]:
    def func(session: AsyncSession = Depends(session.get_db_session)):
        return repository.DatabaseRepository(model, session)

    return func

Simply put, the get_repository function is a dependency factory. It first takes the database model that you will use the repository with. Then, it returns the dependency, which will be used to receive the database session and initialize the repository object. To gain a better understanding, check out the new API from the alchemist/api/v2/routes.py file. It shows only the POST endpoints, but it should be enough to give you a clearer idea of how the code gets improved:

from typing import Annotated

from fastapi import APIRouter, Depends, status

from alchemist.api import models
from alchemist.api.v2.dependencies import get_repository
from alchemist.database import models as db_models
from alchemist.database.repository import DatabaseRepository

router = APIRouter(prefix="/v2", tags=["v2"])

IngredientRepository = Annotated[
    DatabaseRepository[db_models.Ingredient],
    Depends(get_repository(db_models.Ingredient)),
]
PotionRepository = Annotated[
    DatabaseRepository[db_models.Potion],
    Depends(get_repository(db_models.Potion)),
]


@router.post("/ingredients", status_code=status.HTTP_201_CREATED)
async def create_ingredient(
    data: models.IngredientPayload,
    repository: IngredientRepository,
) -> models.Ingredient:
    ingredient = await repository.create(data.dict())
    return models.Ingredient.from_orm(ingredient)


@router.post("/potions", status_code=status.HTTP_201_CREATED)
async def create_potion(
    data: models.PotionPayload,
    ingredient_repository: IngredientRepository,
    potion_repository: PotionRepository,
) -> models.Potion:
    data_dict = data.dict()
    ingredients = await ingredient_repository.filter(
        db_models.Ingredient.pk.in_(data_dict.pop("ingredients"))
    )
    potion = await potion_repository.create({**data_dict, "ingredients": ingredients})
    return models.Potion.from_orm(potion)

The first important feature to note is the use of Annotated, a new way of working with FastAPI dependencies. By specifying the dependency's return type as DatabaseRepository[db_models.Ingredient] and declaring its usage with Depends(get_repository(db_models.Ingredient)) you can end up using simple type annotations in the endpoint: repository: IngredientRepository.

Thanks to the repository, the endpoints don't have to store all the ORM-related burden. Even in the more complicated potion case, all you need to do is to use two repositories at the same time.

You may wonder if initializing two repositories will initialize the session twice. The answer is no. FastAPI dependency system caches the same dependency calls in a single request. This means that session initialization gets cached and both repositories use the exact same session object. Yet another great feature of the combination of SQLAlchemy and FastAPI.

The API is fully functional and has a reusable, high-performing data-access layer. The next step is to make sure the requirements are met by writing some end-to-end tests.

Testing

Tests play a crucial role in software development. Projects can contain unit, integration, and end-to-end (E2E) tests. While it is usually best to have a high number of meaningful unit tests, it is also good to write at least a few E2E tests to ensure the entire workflow is functioning correctly.

To create some E2E tests for the alchemist app, two additional libraries are required:

pytest to actually create and run the tests
httpx to make async requests inside the tests

Once these are installed, the next step is to have a separate, test database in place. You don't want your default database to be polluted or dropped. Since alchemist includes a Docker setup, only a simple script is needed to create a second database. Take a look at the code from the scripts/create_test_db.sh file:

#!/bin/bash

psql -U postgres
psql -c "CREATE DATABASE test"

In order for the script to be executed, it must be added as a volume to the Postgres container. This can be achieved by including it in the volumes section of the docker-compose.yaml file.

The final step of preparation is to create pytest fixtures within the tests/conftest.py file:

from collections.abc import AsyncGenerator

import pytest
import pytest_asyncio
from fastapi import FastAPI
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from alchemist.app import app
from alchemist.config import settings
from alchemist.database.models import Base
from alchemist.database.session import get_db_session


@pytest_asyncio.fixture()
async def db_session() -> AsyncGenerator[AsyncSession, None]:
    """Start a test database session."""
    db_name = settings.DATABASE_URL.split("/")[-1]
    db_url = settings.DATABASE_URL.replace(f"/{db_name}", "/test")

    engine = create_async_engine(db_url)

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    session = async_sessionmaker(engine)()
    yield session
    await session.close()


@pytest.fixture()
def test_app(db_session: AsyncSession) -> FastAPI:
    """Create a test app with overridden dependencies."""
    app.dependency_overrides[get_db_session] = lambda: db_session
    return app


@pytest_asyncio.fixture()
async def client(test_app: FastAPI) -> AsyncGenerator[AsyncClient, None]:
    """Create an http client."""
    async with AsyncClient(app=test_app, base_url="http://test") as client:
        yield client

One thing that is essential to be changed in the tests, is how the app interacts with the database. This includes not only changing the database URL but also ensuring that each test is isolated by starting with an empty database.

The db_session fixture accomplishes both of these goals. Its body takes the following steps:

Create an engine with a modified database URL.
Delete all existing tables to make sure that the test has a clean database.
Create all the tables inside the database (same code as in the migration script).
Create and yield a session object.
Manually close the session when the test is done.

Although the last step could also be implemented as a context manager, manual closing works just fine in this case.

The two remaining fixtures should be quite self-explanatory:

test_app is the FastAPI instance from the alchemist/app.py file, with the get_db_session dependency replaced with the db_session fixture
client is the httpx AsyncClient that will make API requests against the test_app

With all this being set up, finally the actual tests can be written. For conciseness, the example below from the tests/test_api.py file shows only a test for creating an ingredient:

from fastapi import status


class TestIngredientsAPI:
    """Test cases for the ingredients API."""

    async def test_create_ingredient(self, client):
        response = await client.post("/api/v2/ingredients", json={"name": "Carrot"})
        assert response.status_code == status.HTTP_201_CREATED
        pk = response.json().get("pk")
        assert pk is not None

        response = await client.get("/api/v2/ingredients")
        assert response.status_code == status.HTTP_200_OK
        assert len(response.json()) == 1
        assert response.json()[0]["pk"] == pk

The test uses a client object created in a fixture, that makes requests to the FastAPI instance with overridden dependency. As a result, the test is able to interact with a separate database that will be cleared after the test is done. The structure of the remaining test suite for both APIs is pretty much the same.

Summary

FastAPI and SQLAlchemy are excellent technologies for creating modern and powerful backend applications. The freedom, simplicity, and flexibility they offer make them one of the best options for Python-based projects. If developers follow best practices and patterns, they can create performant, robust, and well-structured applications that handle database operations and API logic with ease. This article aimed to provide you with a good understanding of how to set up and maintain this amazing combination.

Sources

The source code for the alchemist project can be found here: link.

Sigma Python #2: Generators

Piotr Tobiasz — Mon, 24 Jul 2023 05:37:32 +0000

Python generators are a crucial feature with multiple uses, from lazy iteration to continuous data streaming. They are heavily adopted in popular packages such as FastAPI, SQLAlchemy, pytest, and others, highlighting their power and the importance of understanding how they work.

This article provides answers to the following questions about generators:

What are they exactly?
How do they work?
What are generator expressions and subgenerators?
What are some common use cases for generators?

Each section covers both theoretical and practical aspects of Python generators. By the end, you will have a strong comprehension of generators and their importance. The first section begins by defining what a generator is and how it operates.

Anatomy of a Generator

At first glance, the generator function appears to be no different from a regular function. Consider the following example:

from typing import Generator

def calculate(a: int, b: int) -> Generator[int | float, None, None]:
    print(f"Starting calculations for {a} and {b}")

    print("Sum")
    yield a + b

    print("Subtraction")
    yield a - b

    print("Multiplication")
    yield a * b

    print("Division")
    yield a / b

    print("Done calculating")

The defining characteristic of a generator function is the use of the yield keyword. When called, the generator function produces a generator, meaning that it can be seen as a generator factory.

To work with a generator, there are two methods. Consider the first example of how to use the calculate function:

calculations = calculate(10, 5)
for result in calculations:
    print("Result:", result)

To start, the calculate generator function is used to create a generator object. It doesn't do anything else: without the for loop, no print outputs would appear. The code then goes through the generator output by iterating over it. This causes the generator to stop at each yield and pass it as a result in the loop. Overall, running this code produces the following result:

Starting calculations for 10 and 5
Sum
Result: 15
Subtraction
Result: 5
Multiplication
Result: 50
Division
Result: 2.0
Done calculating

If you want to work with a generator in a more direct way, you can use the next function. Here's an example of how to use the calculate generator in this manner:

calculations = calculate(10, 5)
print(next(calculations))
print(next(calculations))
print(next(calculations))
print(next(calculations))
print(next(calculations))

Although the outcome is identical, the code now throws a StopIteration exception. This occurs because the generator gets exhausted after the final yield statement. As a result, it cannot be used again, and a new generator object is required to start over. When using a for loop to iterate, it automatically stops when the generator is exhausted, which is why there was no exception in the first example.

When using a generator, it's crucial to pay attention to when it stops and continues the process. If you only have one next call, you'll only see the print statement for the sum, indicating that the generator stopped at the first yield. This means that any remaining computations and print statements will not be executed. This feature can be very useful for adding some "laziness" to your code.

It is important to mention that the final print statement with the "Done calculating" message is called after the fifth next call. This is the same call that exhausts the generator, but it's also the only way to reach the code after the last "yield" statement. To sum up, when using a for loop or the next function, the generator jumps over subsequent yield statements and the code between them until there are no more yields. At that point, the final code is executed, and the generator is exhausted.

Although generators are not an easy topic, this introduction should provide you with a good understanding of how they operate. With this knowledge covered, you can now move on to further interactions with the generator object.

Sending and Returning

Generators use the yield keyword to temporarily pause execution and output a value, which is their core feature. However, generators also have the functionalities of sending and returning.

In the first example, the type annotation Generator[int | float, None, None] indicates that the generator yields values of type integer or float, and the remaining two are types for sending or returning.

The send method of a generator object allows you to input a value into a generator during execution, which is then processed by the yield statement where the generator paused. Here is an example to illustrate this functionality:

import random
from typing import Generator

def send_message(message: str) -> str:
    print(f"Sending: {message}")
    return str(random.randint(1, 1000))

def chat(message: str) -> Generator[str, str, None]:
    print("Starting a new chat")
    history = []

    while True:
        history.append(message)
        response = send_message(message)
        history.append(response)
        message = yield response

quick_chat = chat("hello")
print(next(quick_chat))
print(quick_chat.send("how are you doing?"))
print(quick_chat.send("oh, that is nice!"))

The chat generator function acts like a conversation. It starts with an initial message and uses the send_message function to send and receive messages while keeping a record of the conversation history. The generator object won't get exhausted because of the while True loop. When executed, the code produces the following output:

Starting a new chat
Sending: hello
786
Sending: how are you doing?
392
Sending: oh, that is nice!
416

To send values into a generator, you need to assign the result of yield to a variable (e.g. message = yield response). However, before sending anything, you must first ensure that the generator object proceeds to the first yield statement. Otherwise, there will be nothing to handle the sending. In the example above, this is achieved by calling the next function, which also sends the "hello" message. Once the generator is stopped at the yield statement, you can use the send method. In the example, the send method is used twice to reassign the message variable, which is then sent using the send_message function.

Adding a send feature can enhance the interaction of a generator. By including a return statement within the generator, you can further expand its functionalities. Check out the tweaked chat generator for an example:

... # trunkated code

def chat(message: str) -> Generator[str, str, list[str]]:
    print("Starting a new chat")
    history = []

    while True:
        history.append(message)
        response = send_message(message)
        history.append(response)
        message = yield response
        if not message:
            return history

quick_chat = chat("hello")
print(next(quick_chat))
print(quick_chat.send("how are you doing?"))
print(quick_chat.send("oh, that is nice!"))

try:
    quick_chat.send("")
except StopIteration as e:
    print(e.value)

A new type annotation, list[str] has been added as the return type of the generator. If an empty value is passed via the send method, the generator will now return conversation history. The updated output is as follows:

Starting a new chat
Sending: hello
459
Sending: how are you doing?
817
Sending: oh, that is nice!
84
['hello', '459', 'how are you doing?', '817', 'oh, that is nice!', '84']

The generator got exhausted due to the return statement. However, you can retrieve the returned value from the exception by using the try/except syntax. This method may seem unusual, as it is not the typical way of using a generator. Generally, some iteration is used to go over the generator, which handles the exhaustion behind the scene. The above examples make use of the next and send methods just to showcase the underlying processes.

This knowledge should help you understand how to interact with generators better. Additionally, there are two more techniques to build and exhaust a generator to follow.

Generator Expressions

Generator expressions provide a concise way of creating generators. They enable you to define generators in just one line of code, which is a sophisticated method for generating sequences of data. The syntax for creating a generator expression is like that of a list comprehension. For example, suppose you have a decorator that takes a list of numbers (which might be a large list), squares them, and yields only even numbers. In that case, you can use a generator function as follows:

from typing import Generator

large_numbers_dataset = list(range(1, 11))
print(large_numbers_dataset)

def square_even(numbers: list[int]) -> Generator[int, None, None]:
    for n in numbers:
        if n % 2 == 0:
            yield n**2

squared = square_even(large_numbers_dataset)
print(squared)
print(list(squared))

And the output is the following:

[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
<generator object square_even at 0x7fc5789f4110>
[4, 16, 36, 64, 100]

The results are as expected. In this scenario, using a generator allows for the computation of a portion of the dataset. For instance, you can iterate over the squared generator and stop at the second outcome. In such case, you would receive only two results without processing the entire dataset. Hence, while a one-line list comprehension is feasible, it is not as efficient. Nonetheless, a one-line generator expression can be utilized. To do so, replace the square_even function with the implementation below:

large_numbers_dataset = list(range(1, 11))
squared_exp = (n**2 for n in large_numbers_dataset if n % 2 == 0)
print(squared_exp)
print(list(squared_exp))

The outcome is now nearly the same:

<generator object <genexpr> at 0x7fc5789f5cb0>
[4, 16, 36, 64, 100]

The generator expression ultimately creates a generator object, similar to the generator function. However, it is more concise and expressive. It can be used like a list comprehension, but still maintains the laziness functionality of a generator. Moreover, generator expressions are suitable to be used as a function argument. For instance, consider the following example:

large_numbers_dataset = list(range(1, 11))
print(sum(n**2 for n in large_numbers_dataset if n % 2 == 0))

The result is 220, which means that the generator yielded all the numbers as previously and they were all passed to the sum function. This is a clean and memory-efficient way of processing data without creating intermediate lists.

Subgenerators

In Python, you can create hierarchical generators by using subgenerators. These are generators that are nested within other generators, making it possible to structure and organize your code in a modular and reusable manner. Essentially, subgenerators serve as an alternative method for "unpacking" a generator within another generator. Take a look at the code example below for reference:

numbers_board = [
    [1, 2, 3],
    [4, 5, 6],
    [7, 8, 9],
]

for row in numbers_board:
    for cell in row:
        print(cell)

Naturally, the code displays numbers 1 to 9. Nonetheless, using a double for loop in your code is not the most optimal and elegant solution. A more efficient approach would be to use a generator:

def cells(board: list[list[int]]) -> Generator[int, None, None]:
    for row in board:
        for cell in row:
            yield cell

for cell in cells(numbers_board):
    print(cell)

Although the outcome remains unchanged, your code outside of the function is now more concise and takes benefits from using a generator. Additionally, there is another enhancement you can make in this scenario: separate the two loops and delegate for cell in row to a distinct generator. This can be accomplished by implementing a subgenerator and the yield from syntax:

def columns(row: list[int]) -> Generator[int, None, None]:
    for column in row:
        yield column

def cells(board: list[list[int]]) -> Generator[int, None, None]:
    for row in board:
        yield from columns(row)

for cell in cells(numbers_board):
    print(cell)

The program will print the numbers 1 through 9 again, but this time, the cells generator will use the columns subgenerator to generate the cells. The yield from statement allows the outer generator to transfer control to the inner generator and obtain values from it without having to iterate over each one. Although this may seem like an unnecessary addition in this simple example, it is included to demonstrate the use of a subgenerator.

With the information covered in the previous two chapters, you should now have a good understanding of the generators topic. The next section will explain why it is crucial to have a correct comprehension of how generators work.

Use Cases

Python's generators are definitely a unique aspect of the language that is, however, not used that frequently in the code. This is likely due to the fact that the default iteration and list comprehension syntax are sufficient in most cases. However, incorporating lazy iteration and other generator-related enhancements can sometimes be a game-changer in the codebase.

Despite their limited use in code, generators are essential when working with external libraries such as the FastAPI web framework or pytest testing library. To go through some examples, first, take a look at a class that mimics a database and is aware of its connected status:

class Database:
    def __init__(self) -> None:
        self._entries: list[dict] = []
        self._connected = False

    def connect(self) -> None:
        self._connected = True

    def disconnect(self) -> None:
        self._connected = False

    def add(self, entry: dict) -> None:
        if self._connected is False:
            raise exceptions.DatabaseError("Database is disconnected")
        self._entries.append(entry)

    @property
    def entries(self) -> list[dict]:
        if self._connected is False:
            raise exceptions.DatabaseError("Database is disconnected")
        return self._entries

If you want to implement this class in a FastAPI example, it would appear as follows:

from fastapi import FastAPI, APIRouter

from sigma_py.generators.utils import Database

app = FastAPI()
router = APIRouter()

@router.get("/entries")
async def get_entries() -> list[dict]:
    db = Database()
    db.connect()
    entries = db.entries
    db.disconnect()
    return entries

The issue in this case, is that you need to manually:

initialize the database instance
connect the database
disconnect the database

FastAPI solves this problem by introducing a dependency system, that uses generators. Take a look at the enhanced version of the get_entries endpoint:

from typing import Generator
from fastapi import FastAPI, APIRouter, Depends

from sigma_py.generators.utils import Database

app = FastAPI()
router = APIRouter()

def get_db() -> Generator[Database, None, None]:
    db = Database()
    db.connect()
    yield db
    db.disconnect()

@router.get("/entries")
async def get_entries(db: Database = Depends(get_db)) -> list[dict]:
    return db.entries

The get_db dependency is responsible for all database-related tasks, allowing the endpoint to focus on its intended logic. This dependency is a generator function that creates and connects to a database object, and disconnects it once the generator is exhausted. The framework handles the generator's mechanics behind the scenes.

When using pytest to write tests, a similar process occurs. The library also uses generators to create fixtures that resemble FastAPI dependencies:

import pytest
from typing import Generator

from sigma_py.generators.utils import Database

@pytest.fixture
def db() -> Generator[Database, None, None]:
    db = Database()
    db.connect()
    yield db
    db._entries = []
    db.disconnect()


def test_add_entry(db: Database) -> None:
    assert db.entries == []
    db.add({"name": "Jon"})
    assert db.entries == [{"name": "Jon"}]

As you can see, pytest also leverages a decorator to create and initialize dependencies and add a tear-down code which is executed before the generator gets exhausted.

Summary

Generators are a powerful feature in Python that enable efficient and flexible handling of data generation, iteration, and processing. They offer lazy evaluation, saving memory and enabling infinite sequences. Generators provide advantages such as improved memory utilization, enhanced code readability, and the ability to handle large or infinite data streams. Popular libraries like FastAPI, pytest, Django, TensorFlow, or asyncio utilize generators for dependency management, fixtures, pagination, batch processing, and asynchronous iteration respectively. By understanding and leveraging generators, you can optimize code, improve performance, and design modular and reusable solutions. Embrace generators to write efficient, readable, and scalable code in various domains.

Sources

Code examples used in the article can be found here: link.

The biggest inspiration for these articles and source of my knowledge is the Fluent Python book by Luciano Ramalho. I highly encourage you to check it out; you will not be disappointed.

Documentation page of each library:

Debugging Python Apps in VSCode

Piotr Tobiasz — Sun, 23 Jul 2023 10:24:56 +0000

Debugging is an essential part of software development. It can be a time-consuming and frustrating process. There is nothing wrong with putting print statements inside the code to check something quickly. If it works for you in most cases, then why should you worry about overcomplicating this process? However, in more complex scenarios, using just a logging mechanism may be insufficient or even unfeasible.

This article covers the usage of Visual Studio Code debugging tools for Python applications. Examples are shown on the FastAPI app running on top of the uvicorn server. The article addresses the following scenarios:

debugging a locally running code
running a container and attaching a debugger to it
attaching to a running container and debugging the code inside it

After reading this article, you will have a comprehensive understanding of the different debugging methods available in VSCode. This will enable you to choose the most appropriate strategy for any given scenario.

Prerequisites

The main focus of this article is the configuration and usage of the VSCode debugger. It doesn't focus too much on the uvicorn, FastAPI, or Docker specifics. Everything related to the debugged app can be found in the debug_hell project, which is available here: link. You can set it up by yourself, by cloning the repository and running the commands from the readme file. From a high-level perspective, debug_hell is a straightforward, create-read API that uses an ORM to store records in the database.

Problem Statement

In the modern programming world, most projects contain Docker configuration for quick and reliable local setup. Often you just need to prepare the environment variables and run two commands: docker compose build and docker compose up. This will result in building and running all the containers required for the app to be functional. This is also the case for debug_hell project, which consists of the FastAPI app and Postgres database.

Docker makes setting up apps locally a breeze, however, things get more complicated when it comes to debugging. Your code is now running in the container and it also uses its version of installed packages, not your local virtual environment. Placing breakpoints in your code or in external packages will simply not work.

Debugging App Locally

The first method for debugging an app, such as debug_hell, is to run the server locally, which means replacing the Docker container with a local process. This approach has a clear disadvantage: installing dependencies locally. Although it may seem counterproductive, it could be the easiest and quickest solution in certain simpler situations.

This is precisely the case for the debug_hell app, which requires only two dependencies: installed packages and the Postgres database. To set up the former, create a new virtual environment, by running the following commands:

python -m venv venv
source venv/bin/activate
pip install requirements/dev.txt

When it comes to the database, the local process can also utilize the one from the Postgres Docker container. You can start it up just like before by executing the docker compose up command.

With that in place, the next step is configuring the debugger. Take a look at the configuration for the debug_hello app, located in the .vscode/launch.json file:

{
    "configurations": [
        {
            "name": "Run Server",
            "type": "python",
            "request": "launch",
            "module": "uvicorn",
            "args": [
                "debug_hell.app:app",
                "--reload",
                "--host",
                "0.0.0.0",
                "--port",
                "8001"
            ],
            "env": {
                "DATABASE_URL": "postgresql+asyncpg://postgres:postgres@localhost:5432/debug_hell"
            },
            "justMyCode": false
        }
    ]
}

Since the packages are available locally, the debugger can start its own uvicorn server. Besides the debugger-specific configuration, you can also notice two important changes:

the port is set to 8001 to not interfere with the in-container app (if one is not stopped)
database URL is pointing to the localhost instead of the postgres, which was only accessible within the Docker network

Now that you have everything set up, you can begin debugging. Navigate to the "Run and Debug" section and run the "Run Server" configuration or press the F5 key. A floating menu will appear for the debugger, allowing you to easily place bookmarks in your code and external libraries within the virtual environment. See the example result below:

From now on, when your request passes through the bookmarked code, the debugger will stop and grant you access to all the runtime insights. That includes both the "Variables" section in the debug tab and the debug console, which enables you to execute Python commands from the stopped line.

Implementing this approach is a fast and efficient way to produce the desired outcome. However, as previously mentioned, it may not always be that easy to set up apps locally. Docker is so popular because it simplifies the setup process. If there is a complicated installation process in place, avoiding Docker may be too time-consuming. The next section demonstrates how to utilize the debugger in conjunction with Docker containers.

Running Container Debugger

If the application you're dealing with has a more complicated setup, using Docker containers is likely the more logical choice. Luckily, the VSCode debugger offers built-in features to collaborate with Docker.

To get started, first specify the tasks in the .vscode/tasks.json file that will be executed to build and start the debugging container:

{
    "version": "2.0.0",
    "tasks": [
        {
            "type": "docker-build",
            "label": "docker-build",
            "platform": "python",
            "dockerBuild": {
                "dockerfile": "${workspaceFolder}/infra/Dockerfile",
                "context": "${workspaceFolder}",
                "target": "dev",
                "pull": true
            }
        },
        {
            "type": "docker-run",
            "label": "docker-run: debug",
            "dependsOn": [
                "docker-build"
            ],
            "dockerRun": {
                "ports": [
                    {
                        "containerPort": 8001,
                        "hostPort": 8001
                    }
                ],
                "volumes": [
                    {
                        "containerPath": "/code",
                        "localPath": "${workspaceFolder}"
                    }
                ],
                "network": "debug_hell_default"
            },
            "python": {
                "args": [
                    "debug_hell.app:app",
                    "--reload",
                    "--host",
                    "0.0.0.0",
                    "--port",
                    "8001"
                ],
                "module": "uvicorn"
            }
        }
    ]
}

If you have prior knowledge of Docker, this configuration should be comprehensible to you. In addition to some debugger-specific details and the port being modified, there is one more crucial adjustment: the Docker configuration now has the network attribute set to debug_hell_default to enable the container to access the database from the Docker Compose configuration.

Now that you have it set up, you can add the next launch configuration to the .vscode/launch.json file:

{
    "configurations": [
        {
            "name": "Run Docker",
            "type": "docker",
            "request": "launch",
            "preLaunchTask": "docker-run: debug",
            "python": {
                "pathMappings": [
                    {
                        "remoteRoot": "/code",
                        "localRoot": "${workspaceFolder}"
                    }
                ]
            }
        }
    ]
}

You should now be able to start the "Run Docker" configuration under the "Run and Debug" section with that basic config in place. Running it will cause the Docker image to be built and the container to be started. You should see the debugger menu again, and you can begin inserting breakpoints in the code.

This approach is quite flexible, does not require running anything locally, and already works fairly well. When it comes to debugging, in many cases it will be enough. However, one major drawback is that the code from external libraries cannot be debugged. Because the container runs its own version of Python, you simply cannot place bookmarks in the installed packages, because you don't have access to them.

The following chapter explores attaching to a running container in order to debug the code inside it.

Debugging App Inside Container

Debugging code from a running container is convenient, but it can be a little more challenging when dealing with third-party libraries. Luckily, there are tools available that enable working with VSCode inside the Docker container and debugging code within it.

There are several steps you must take to get ready for debugging within the container:

Install the "Dev Containers" extension. It allows you to enter a Docker container and open a folder inside it, just like you would normally do with your project.
Start the debugging container with the "Run Docker" configuration created in the previous section.
Enter the running Docker container by doing the following:
1. Press CTRL + Shift + P key combination.
2. Select the "Dev Containers: Attach to Running Container…" option.
3. Select the debugging container started in step 2.
Open the /code directory, if one is not opened already.

Now that it's set up, you'll be able to view all the code from the container's perspective. Any changes made to files within it will also be reflected in your local VSCode window.

The Python process you want to debug is already running within the container as a result of step 2. You may connect a debugger to an already running process by supplying its id to the VSCode configuration. The Python process id may be found in the original VSCode window, under the "Call Stack" part of the "Run and Debug" tab. Here's an example of a process with id 33:

With that information, you may return to the in-container VSCode window. A few more steps are required to connect the debugger to the running process.

To begin, ensure that the Python extension in VSCode is installed. When you have it, you must also choose a proper virtual environment. It should be /usr/local/bin/python by default. That is where the Docker container stores its Python version along with all installed libraries.

Another dependency is the gdp package. Without it, you cannot run the debugger correctly. To install gdp library, simply run apt-get install gdp command.

Now that the required setup is complete, it's time to include a new configuration in the .vscode/launch.json file:

{
    "configurations": [
        {
            "name": "Attach",
            "type": "python",
            "request": "attach",
            "processId": "${command:pickProcess}",
            "justMyCode": false
        }
    ]
}

When started from the "Run and Debug" tab, that simple configuration will prompt you for the process id you obtained previously. The debugger should correctly attach to the running process after passing it. That configuration is eventually supposed to function exactly like the one from the first chapter. You should be able to place breakpoints in your code and external libraries, and the debugger should respect both.

Summary

Debugging Python applications or libraries can be a challenging task, but Visual Studio Code offers useful tools to help developers debug Python code effectively. While more advanced IDEs like PyCharm come with powerful debuggers, they often require additional costs, have steep learning curves, or consume more resources. In contrast, VSCode is a free and lightweight option, but configuring its debugger requires some extra effort. The three methods presented here offer good choices for different situations where a debugger is needed.

Sources

Source code for the debug_hell project can be found here: link.

Sigma Python #1: Decorators

Piotr Tobiasz — Sat, 22 Jul 2023 15:20:07 +0000

One of the most interesting and useful features of Python are decorators, which are callable objects that modify the behavior of other pieces of code without any additional changes. Decorators are a fundamental concept in Python, and they are used extensively in the language's standard library, as well as in third-party libraries and frameworks.

Some of the most common usages of decorators include:

logging
timing
caching
error handling

Decorators are also a great way to create reusable code since they can be applied to multiple functions.

This article explores the topic of Python decorators, including their following aspects:

structure
behavior
closures
parametrization
class-based implementation

Each section offers simple code examples related to a specific subject, with one more complex and practical example concluding the article in the last section.

Whether you are a beginner or an experienced Python programmer, understanding decorators is essential for writing clean, concise, and maintainable code. By the end of this article, you will have a solid understanding of Python decorators and be able to use them in your own code to improve its functionality and readability.

Anatomy of a Decorator

A decorator, as previously said, is a piece of code that adds or modifies the functionality of another piece of code. Such behavior is achieved by creating either a function or a class that accepts an existing function as a parameter and 'decorates' it. The following example shows a simple decorator and two ways to decorate a function:

from typing import Any, Callable

def announcer(func: Callable[..., Any]) -> Callable[..., Any]:
    print(f"Decorating {func.__name__}")

    def wrapper(*args, **kwargs) -> Any:
        print(f"Calling {func.__name__} with {args} and {kwargs}")
        result = func(*args, **kwargs)
        print(f"The result is {result}")
        return result

    return wrapper

@announcer
def to_fahrenheit(celsius: float) -> float:
    return (celsius * 1.8) + 32

def to_celsius(fahrenheit: float) -> float:
    return (fahrenheit - 32) / 1.8

to_celsius = announcer(to_celsius)

if __name__ == "__main__":
    to_fahrenheit(42.0)
    to_fahrenheit(celsius=11.0)
    to_celsius(452.0)
    to_celsius(fahrenheit=111.0)

announcer is a decorator that replaces the decorated function with a new function that takes the following steps:

Prints the name of called function and passed arguments.
Calls the decorated function and saves the result.
Prints and returns the result.

Executing the preceding code yields the following result:

Decorating to_fahrenheit
Decorating to_celsius
Calling to_fahrenheit with (42,) and {}
The result is 107.60000000000001
Calling to_fahrenheit with () and {'celsius': 11}
The result is 51.8
Calling to_celsius with (452,) and {}
The result is 233.33333333333331
Calling to_celsius with () and {'fahrenheit': 111}
The result is 43.888888888888886

The output reveals that wrapper function from the decorator replaced both to_fahrenheit and to_celsius functions. On top of the calculations happening in the initial implementation, the functions now make the print calls added in the decorator.

It is also worth noticing when the first two prints actually appear. Even if both to_fahrenheit and to_celsius were removed, those two prints would still be present. This is because decorators are being called at the import time. When you consider how to_celsius is decorated, where the announcer decorator is explicitly called, it makes a lot of sense.

One caveat that isn't visible in the output is what both decorated functions actually became and how IDE (like VSCode or PyCharm) might be reacting to them afterward. Considering the following changes to the previous code:

if __name__ == "__main__":
    print(to_fahrnheit)
    print(to_fahrenheit.__annotations__)
    print(to_celsius)
    print(to_celsius.__annotations__)

Now, the outcome is as follows:

Decorating to_fahrenheit
Decorating to_celsius
<function announcer.<locals>.wrapper at 0x7f452835a7a0>
{'return': typing.Any}
<function announcer.<locals>.wrapper at 0x7f452821aa70>
{'return': typing.Any}

The output now may be unexpected, and it also reveals a larger issue: decorated functions were not only 'enriched' with new features, but they were actually replaced by the decorator's wrapper function. In this context, 'replaced' means that any information about the initial function, such as name or type annotations, is gone. That is why your IDE may be unable to assist you with autocompletion while dealing with decorated functions. Code just does not include the necessary information any longer.

While you could manually add all the necessary information to the wrapper function, there is a built-in solution to address this problem:

import functools

def announcer(func: Callable[..., Any]) -> Callable[..., Any]:
    print(f"Decorating {func.__name__}")

    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> Any:
        print(f"Calling {func.__name__} with {args} and {kwargs}")
        result = func(*args, **kwargs)
        print(f"The result is {result}")
        return result

    return wrapper

functools is a Python standard library that contains several high-order functions – ones that are used for modifying existing functions. wraps is a decorator that accepts the decorated function as an argument and then updates the wrapper that was meant to replace the original function. Under the hood, wraps invokes the update_wrapper function, which provides probably the clearest description of this process:

Update a wrapper function to look like the wrapped function

With these announcer changes in place, the output now is the following:

Decorating to_fahrenheit
Decorating to_celsius
<function to_fahrenheit at 0x7f540e3ae7a0>
{'celsius': <class 'float'>, 'return': <class 'float'>}
<function to_celsius at 0x7f540e26ea70>
{'fahrenheit': <class 'float'>, 'return': <class 'float'>}

In this case, wrapper function from the decorator still replaces the decorated functions, but now it contains all the essential information copied from the initial function.

Next comes the topic of free variables and closures, so how decorated functions are able to 'share state' between themselves.

Closures

Take a look back at the very first example and its output. If the wrapper function replaced the decorated functions, then how did it still have access to their original names? You can see it did in the first print call in the wrapper. To make it more clear, consider the following example:

from typing import Any, Callable

def register(func: Callable[..., Any]) -> Callable[..., Any]:
    results = []

    def wrapper(*args, **kwargs) -> Any:
        result = func(*args, **kwargs)
        results.append(result)
        print(f"Results: {results}")
        return result

    return wrapper

@register
def to_fahrenheit(celsius: float) -> float:
    return (celsius * 1.8) + 32

if __name__ == "__main__":
    to_fahrenheit(10.0)
    to_fahrenheit(20.0)
    to_fahrenheit(30.0)

The register decorator stores the results of all the calls that happen on the functions it decorates. Calling this code results in the following output:

Results: [50.0]
Results: [50.0, 68.0]
Results: [50.0, 68.0, 86.0]

This outcome should come as no surprise, given the code's simplicity. However, another question that may arise is how the wrapper function is able to reach the results variable. The register decorator returns the wrapper function, hence results should be outside its scope. The func variable in the first example was similar in that it belonged to the announcer decorator rather than the wrapper function. That behavior is possible thanks to Python features known as free variables and closures.

When you create a nested function that references a variable that is neither local nor global, Python creates a closure to bind the variable to the nested function. Such variables are known as free variables, and they are stored in a __closure__ attribute. This is particularly common with decorators, where the decorated function is a free variable. To illustrate this on a previous example, consider the following changes:

if __name__ == "__main__":
    to_fahrenheit(10.0)
    to_fahrenheit(20.0)
    to_fahrenheit(30.0)
    for cell in to_fahrenheit.__closure__:
        print(cell.cell_contents)

__closure__ attribute contains objects called cells that store the values of free variables in a cell_contents attribute. Here is the output produced by the above code:

Results: [50.0]
Results: [50.0, 68.0]
Results: [50.0, 68.0, 86.0]
<function to_fahrenheit at 0x7fdca98d1da0>
[50.0, 68.0, 86.0]

The last two lines show that the decorated function does, in fact, save the additional values. The original function is the first one – that explains how the wrapper in the first example was able to print the name of the decorated function. The second is the result variable created in the decorator. This demonstrates how powerful feature the closures are: since result is mutable, all the decorated function can share a certain state.

The next section covers parametrization: how to provide arguments to a decorator before the decoration happens.

Parametrization

It is fairly common to use a decorator that requires some configuration. One of the most popular examples is the lru_cache decorator from the functools standard library. It is used to cache the output of a decorated function based on its parameters. lru_cache comes with a default configuration, but you can also provide two additional arguments: maxsize and typed. The first indicates how many entries can be cached before the LRU policy begins to remove the oldest items. The latter determines whether the same values of various types should be cached separately. Take a look at this simple example:

from functools import lru_cache

@lru_cache(maxsize=2)
def to_fahrenheit(celsius: float) -> float:
    print(f"Calling to_fahrenheit with {celsius}")
    return (celsius * 1.8) + 32

if __name__ == "__main__":
    to_fahrenheit(10.0)
    to_fahrenheit(20.0)
    to_fahrenheit(20.0)
    to_fahrenheit(30.0)
    to_fahrenheit(40.0)
    to_fahrenheit(20.0)

When maxsize=2 is specified, only two results can be cached at the same time. The output of this code is the following:

Calling to_fahrenheit with 10
Calling to_fahrenheit with 20
Calling to_fahrenheit with 30
Calling to_fahrenheit with 40
Calling to_fahrenheit with 20

First, you can see that print with 20 was called just once, despite the fact that to_fahrenheit was called twice with 20 as the argument, indicating that the cache worked. But, at the very end, to_fahrenheit(20.0) didn't use the cache anymore. This is simply because calls with 30 and 40 took all the cache seats, forcing the output of to_fahrenheit(20.0) to be removed.

This basic example should demonstrate the potential of decorators' parametrization: not only you can add functionality to existing code, but you can also configure how it should behave.

Implementing a decorator that accepts some parameters can be tricky, and the syntax may appear confusing. Consider the following example:

from typing import Any, Callable, TypeAlias

DecoratedFunc: TypeAlias = Callable[..., Any]

def retry(max_tries: int = 3) -> DecoratedFunc:
    def decorator(fn: DecoratedFunc) -> DecoratedFunc:
        def wrapper(*args, **kwargs) -> Any:
            for i in range(max_tries):
                try:
                    return fn(*args, **kwargs)
                except Exception as exc:
                    if i + 1 == max_tries:
                        raise exc
                    print(f"{fn.__name__} failed with {exc}, retrying")

        return wrapper

    return decorator

@retry(max_tries=2)
def to_celsius(fahrenheit: float) -> float:
    print(f"Calling to_celsius with {fahrenheit}")
    return (fahrenheit - 32) / 1.8

if __name__ == "__main__":
    to_celsius(100.0)
    to_celsius(200.0)
    to_celsius("300")

retry is a simple decorator that causes the decorated function to be called again if something fails inside it. It has a max_tries parameter, which specifies how many times should a decorated function be called before giving up if an exception persists. Here is the output of running the code:

Calling to_celsius with 100.0
Calling to_celsius with 200.0
Calling to_celsius with 300
to_celsius failed with unsupported operand type(s) for -: 'str' and 'int', retrying
Calling to_celsius with 300
Traceback (most recent call last):
  (trunkated error traceback)
  File "/sigma_py/decorators/decorators04.py", line 31, in to_celsius
    return (fahrenheit - 32) / 1.8
            ~~~~~~~~~~~^~~~
TypeError: unsupported operand type(s) for -: 'str' and 'int'

As you can see, the first two calls worked normally, but the third one triggered the retry mechanism. On the second unsuccessful retry, it gave up and raised the original exception. Of course, these kinds of retry decorators are usually way more sophisticated. The goal here was to test the max_tries parameter in action.

What can be confusing is how 'nested' the decorator became. The reason is fairly simple: in this case, retry is not actually a decorator, but rather a decorator factory. After receiving the arguments, it returns the decorator, which is the decorator function in this example. The @ syntax then decorates the function underneath it. The rest remains the same: wrappers, closures, free variables, and so forth.

Because max_tries has a default value of 3, you might expect that you could use this decorator without the brackets, i.e., @retry. With this implementation, this is not possible, or at least will not produce the expected outcomes. Since retry is actually a decorator factory, using it without brackets would replace the decorated function with the decorator function. This would of course completely break the program.

However, it is possible to implement a decorator in a way that it would use the default parameters without the need to call it explicitly. This is done in the final section of the article, where retry decorator is going to be expanded.

With parametrization covered, the next topic is the implementation of decorators using a class-based approach.

Class-based Decorators

In Python, class instances that implement the __call__ method can be invoked just like functions. This means, that it is possible to implement class-based decorators. Even though it doesn't provide any major advantages over the regular, functional approach, it can sometimes be easier to organize the code in a class.

Here is how the retry decorator from the previous example looks like when implemented as a class:

from typing import Any, Callable, TypeAlias

DecoratedFunc: TypeAlias = Callable[..., Any]

class Retry:
    def __init__(self, max_tries: int = 3) -> None:
        self.max_tries = max_tries

    def __call__(self, fn: DecoratedFunc) -> DecoratedFunc:
        def wrapper(*args, **kwargs) -> Any:
            for i in range(self.max_tries):
                try:
                    return fn(*args, **kwargs)
                except Exception as exc:
                    if i + 1 == self.max_tries:
                        raise exc
                    print(f"{fn.__name__} failed with {exc}, retrying")

        return wrapper

@Retry(max_tries=2)
def to_celsius(fahrenheit: float) -> float:
    print(f"Calling to_celsius with {fahrenheit}")
    return (fahrenheit - 32) / 1.8

if __name__ == "__main__":
    to_celsius(100.0)
    to_celsius(200.0)
    to_celsius("300")

This code produces exactly the same output as the one in the preceding example.

Similar to the functional approach, calling __init__ method of the Retry class is analogous to using a decorator factory. The class instance is then invoked using the __call__ method to replace the decorated function with wrapper.

Using a decorator with a capitalized name might not appear Pythonic, but naming a class using the snake case convention is probably even worse. This can, however, be easily solved by adding a variable that references the class, i.e., retry_that_thing = RetryThatThing. Following that, a class-based decorator looks just like a function.

The final section goes back to the functional retry decorator in order to make it more powerful and customizable.

Advanced Scenario

The first version of the retry decorator was relatively straightforward. Its updated version includes the following features:

ability to use the decorator without brackets
parameter to specify exceptions for which to retry
parameter to define a callback replacing the final raise

The code for the enriched retry decorator is as follows:

from typing import Any, Callable, TypeAlias

DecoratedFunc: TypeAlias = Callable[..., Any]

def retry(
    max_tries: int | DecoratedFunc = 3,
    exceptions: tuple[type[Exception], ...] | None = None,
    on_giveup: Callable[[Exception], None] | None = None,
) -> DecoratedFunc:
    # Catch all exception by default
    if exceptions is None:
        exceptions = (Exception,)

    decorated_func = None

    # If decorator was used without brackets, then function is the first arg
    if callable(max_tries):
        decorated_func, max_tries = max_tries, 3

    def decorator(fn: DecoratedFunc) -> DecoratedFunc:
        def wrapper(*args, **kwargs) -> Any:
            for i in range(max_tries):
                try:
                    return fn(*args, **kwargs)
                except exceptions as exc:
                    if i + 1 == max_tries:
                        if on_giveup is None:
                            raise exc
                        return on_giveup(exc)

                    print(f"{fn.__name__} failed with {exc}, retrying")

        return wrapper

    # If decorated_func is already set, then run decoration immediately
    if decorated_func is not None:
        return decorator(decorated_func)

    return decorator

@retry
def to_celsius(fahrenheit: float) -> float:
    print(f"Calling to_celsius with {fahrenheit}")
    return (fahrenheit - 32) / 1.8

@retry(
    max_tries=2,
    exceptions=(TypeError,),
    on_giveup=lambda exc: print(f"Giving up: {exc}"),
)
def to_fahrenheit(celsius: float) -> float:
    print(f"Calling to_fahrenheit with {celsius}")
    return (celsius * 1.8) + 32

if __name__ == "__main__":
    to_celsius(100.0)
    to_fahrenheit(40)
    to_fahrenheit("50")

The retry function now accepts three parameters:

max_tries, which works just as previously
exceptions, which describes what exception classes should be handled by the retry mechanism
on_giveup, that specifies a callable to use in case of exceeding the max_tries

One notable change is that max_tries is now annotated as int | DecoratedFunc. This is due to the behavior of using retry without the brackets. In such a scenario, the decorated function is passed as the first argument. For the same reason, if callable(max_tries) line checks whether max_tries is something that can be called. If it is, all of the arguments should use their default values. The decorated function is then saved in a separate variable and at the very end it is directly decorated instead of returning the decorator.

The remaining two arguments are used in the except clause to add the extra behavior.

The output of the final retry version is the following:

Calling to_celsius with 100.0
Calling to_fahrenheit with 40
Calling to_fahrenheit with 50
to_fahrenheit failed with can't multiply sequence by non-int of type 'float', retrying
Calling to_fahrenheit with 50
Giving up: can't multiply sequence by non-int of type 'float'

It demonstrates that both usages of the decorator are working as expected and that the new functionalities are present as well.

Summary

In conclusion, Python decorators offer a versatile way to modify the behavior of Python functions without modifying their original code. Decorators can be used to improve the functionality and maintainability of your Python code if you have a strong understanding of how they work and some common use cases. Decorators can help you build cleaner, more modular, and extensible Python code by harnessing the power of closures and higher-order functions.

Sources

Code examples used in the article can be found here: link.

The biggest inspiration for these articles and source of my knowledge is the Fluent Python book by Luciano Ramalho. I highly encourage you to check it out; you will not be disappointed.

Using Textual to Build a ChatGPT TUI App

Piotr Tobiasz — Fri, 21 Jul 2023 07:05:21 +0000

TUI stands for "text user interface". It refers to a type of user interface that relies primarily on text and symbols to interact with users, rather than visual elements like icons and images. TUI applications typically run in a command-line interface (CLI) environment, which is a type of interface where users interact with a computer by typing commands into a text-based interface. TUI applications have been around for a long time and despite their seemingly archaic design, they have certain advantages over graphical user interface (GUI) applications. These include faster response times, lower system requirements, and easier automation.

Is It Really in the Terminal?

Several libraries, such as urwid or PyTermGUI, allow the development of TUI applications in Python. For enhancing the functionality and aesthetics of TUI apps, they offer some fundamental and more sophisticated utilities. But there is one package that is truly exceptional and might even be so amazing that it sparks a TUI renaissance (I really wanted to put "TUI renaissance" somewhere in this article).

Textual is a package created by Will McGugan, the creator of the rich, extremely popular terminal text formatting library. Textual is actually built on top of rich to support all of those lovely features, add interactivity to them, and enable the creation of more complicated apps and components. When you first encounter an example of Textual app (and you can find them directly in the Textual repository), you may wonder: is it really a built-in terminal? Yes, it is.

It's no coincidence that Textual has reached approximately 20k GitHub stars in less than two years. You don't win the hearts of the community by just giving a tool that does something. In the case of Textual, it is obviously much more than that. First and foremost, the code quality is excellent; everything is type-hinted and documented. Another point is how practical and simple Textual is to use. This is because there are so many built-in features. Async support is also included right out of the box. Eventually, working with Textual produces effects that are simply remarkable. In conclusion, you may easily and quickly create the most stunning TUI apps that have ever been seen.

ChatGPT and Textual

Using Textual to create TUI applications may be an interesting and rewarding experience. Without investing a lot of time in the "frontend" work, you can easily create something usable with an attractive interface. Here, we'll concentrate on chatui, an in-terminal ChatGPT app. chatui will connect with the ChatGPT engine using the openai package, and the interface will be made, of course, with Textual. The most important parts of the project structure will include the following:

chatui
├─ requirements
│  ├─ base.txt
│  └─ dev.txt
├─ chatui
│  ├─ static
│  │  └─ styles.css
│  ├─ __init__.py
│  ├─ chat.py
│  ├─ tui.py
│  ├─ settings.py
│  └─ main.py
└─ Makefile

To avoid creating an unnecessary burden, some of the extra information is hidden. Nevertheless, to see how everything is wired, you can find the complete project here.

Setting up the Chat

After creating a new directory and spinning up a new virtual environment, the first thing to deal with is the dependencies. In this case, only two are required: textual==0.14.0 and openai==0.27.1. These should be added to requirements/base.txt file, and installed with pip install -r requirements/base.txt command.

The OpenAI API key, which is required to communicate with the ChatGPT model, comes next. It can be acquired on the OpenAI account page. Once it's obtained you certainly don't want it to end up in the project repository. Instead, by running export OPENAI_KEY=key> it may be loaded as an environment variable, which can then be read in the chatui/settings.py file:

import os

OPENAI_KEY = os.environ["OPENAI_KEY"]

With that in place, the first step is to develop the actual chat logic. It's not too difficult thanks to the openai library. It provides a ChatCompletion.acreate method that accepts the name of the AI model and a collection of messages. However, some additional code is required to keep the whole context of the conversation:

import openai

from chatui import settings

openai.api_key = settings.OPENAI_KEY

class Conversation:
    model: str = "gpt-3.5-turbo"

    def __init__(self) -> None:
        self.messages: list[dict] = []

    async def send(self, message: str) -> list[str]:
        self.messages.append({"role": "user", "content": message})
        r = await openai.ChatCompletion.acreate(
            model=self.model,
            messages=self.messages,
        )
        return [choice["message"]["content"] for choice in r["choices"]]

    def pick_response(self, choice: str) -> None:
        self.messages.append({"role": "assistant", "content": choice})

    def clear(self) -> None:
        self.messages = []

As a result, the above class keeps track of all messages, and the ChatGPT model is fully aware of the ongoing dialogue. There are a few things worth mentioning. The first is the OpenAI model: gpt-3.5-turbo, which powers the ChatGPT and is also recommended for the majority of use cases. Another factor is that messages must include information about the role in addition to text. As a result, the AI model will know who said what. It would be too risky to rely solely on the message order. Finally, the Conversation class returns a list of response options. Despite the fact that, in most circumstances (at least in my experience), this list has just one option, the send method returns the complete list and allows the user to select the desired option.

It may be difficult to believe, but these 32 lines of code can already execute a genuine dialogue. It can already be seen in action, by creating a chatui/main.py file with an alpha version of the app:

import asyncio

from chatui.chat import Conversation

async def main() -> None:
    conversation = Conversation()
    while True:
        msg = input("Type your message: ")
        choices = await conversation.send(msg)
        print("Here are your choices:", choices)
        choice_index = input("Pick your choice: ")
        conversation.pick_response(choices[int(choice_index)])

if __name__ == "__main__":
    asyncio.run(main())

Running python chatui/main.py can produce the following results:

$ python chatui/main.py
Type your message: Tell a joke
Here are your choices: ['\\\\n\\\\nWhy did the tomato turn red? \\\\nBecause it saw the salad dressing!']
Pick your choice: 0
Type your message: Tell another one
Here are your choices: ['Why did the chicken cross the playground?\\\\nTo get to the other slide.']
Pick your choice: ^C

If you have any problems running this script, please make sure to: activate the virtual environment; export the API key; export the output of pwd command as the PYTHONPATH environment variable if any "Module Not Found" error occurs.

This actually already satisfies the functional requirements of chatui:

Asking questions,
Receiving answers,
Persisting context.

With chat handling ready, the next step is the user-facing part: TUI.

Textual in Action

Building user interfaces with Textual is done by utilizing classes provided by the library. The base class for creating the app is called (surprisingly) App and it is used to connect different components and handle various events and actions. Starting with something simple, chatui/tui.py will contain the following code:

from textual.app import App, ComposeResult
from textual.widgets import Footer, Header, Placeholder

class ChatApp(App):
    TITLE = "chatui"
    SUB_TITLE = "ChatGPT directly in your terminal"
    CSS_PATH = "static/styles.css"

    def compose(self) -> ComposeResult:
        yield Header()
        yield Placeholder()
        yield Footer()

The Textual app must now be run in place of the CLI chat in the chatui/main.py file:

from chatui.tui import ChatApp

if __name__ == "__main__":
    app = ChatApp()
    app.run()

With that in place, you can use the python chatui/main.py command to launch the app. The expected output is as follows:

This is already quite intriguing. The ChatApp class begins with some configuration that will be utilized by the interface. The compose method then yields widgets to the parent container. In this scenario, there are three components: Header which contains configuration-related information, Placeholder, which simply fills the empty space, and Placeholder, which is now just a sticker at the bottom. To exit the app, use CTRL+C combination.

Even though this is a really simple example, it already opens the door to building something far more powerful.

Handling Input

Core functionality of chatui is sending messages. It obviously cannot be done without the ability to take and process input. To make this happen, chatui needs to have an Input component:

from textual.app import App, ComposeResult
from textual.widgets import Footer, Header, Input, Button
from textual.containers import Container, Horizontal

class ChatApp(App):
    TITLE = "chatui"
    SUB_TITLE = "ChatGPT directly in your terminal"
    CSS_PATH = "static/styles.css"

    def compose(self) -> ComposeResult:
        yield Header()
        with Horizontal(id="input_box"):
            yield Input(placeholder="Enter your message", id="message_input")
            yield Button(label="Send", variant="success", id="send_button")
        yield Footer()

The widgets will already be added to the app, but in order for them to resize and position correctly, they require some additional CSS. static/styles.css is where styling is stored:

#input_box {
    dock: bottom;
    height: auto;
    width: 100%;
    margin: 0 0 2 0;
    align_horizontal: center;
}

#message_input {
    width: 50%;
    background: #343a40;
}

#send_button {
    width: auto;
}

Ultimately, this will result in the interface seen below:

With things looking much better now, a quick recap of what actually happened. First, Placeholder widget was replaced with 3 different components. Horizontal is a container widget that organizes the items inside it horizontally. Using it as a context manager (with keyword) is simply an easier way of passing the widgets to the container __init__ method. Finally, within the Horizontal container, there are Input and Button widgets, which are quite self-explanatory.

There are a few things worth highlighting in the static/styles.css file as well. Aside from standard CSS code, there are other things unique to Textual:

The dock: bottom attribute causes the container to stick to the bottom,
height: auto instructs the container to take only as much height as it needs, rather than occupying the entire area,
If there is some free space inside the container, align_horizontal: center causes the items to move to the center.

Having decent aesthetics is the first step to making this UI usable. Following that, it must actually handle the input. Thankfully, Textual offers a variety of options to handle events occurring within the app:

...  # trunkated imports
from textual.widget import Widget

class ChatApp(App):
    ...  # trunkated code

    async def on_button_pressed(self) -> None:
        await self.process_conversation()

    async def on_input_submitted(self) -> None:
        await self.process_conversation()

    async def process_conversation(self) -> None:
        message_input = self.query_one("#message_input", Input)
        # Don't do anything if input is empty
        if message_input.value == "":
            return
        button = self.query_one("#send_button")

        self.toggle_widgets(message_input, button)

        # Clean up the input without triggering events
        with message_input.prevent(Input.Changed):
            message_input.value = ""

    def toggle_widgets(self, *widgets: Widget) -> None:
        for w in widgets:
            w.disabled = not w.disabled

The on_button_pressed and on_input_submitted methods serve as the starting points for submitting input. They are referred to as "message handlers" in Textual terminology. These methods have three elements to their names:

on indicates that a method is actually a message handler,
button describes the namespace: what produced the message,
pressed names the class of the message - what actually happened.

Because of this naming approach, relatively minimal code is required to handle events within the app. There are two possible messages in this case: input being submitted after clicking the enter button while the input is focused, and the send button being pressed. In a more sophisticated case with many inputs and buttons, the code would have to be more accurate in processing the messages, but because chatui only has one of each, it doesn't have to worry about it.

Following that is the process_conversation method, which is the root of the app domain logic and is used by both handlers. It doesn't have many responsibilities for now, but there is some intriguing Textual code. It first looks for the Input widget by using its id. It is comparable to an ORM searching the database for a certain record. When the input widget is obtained, the code checks to see if it contains any text; if it doesn't, there is no need to perform any more actions. Otherwise, it continues by retrieving the next component: Button. There is also a helper method for toggling a list of widgets. In this example, it will deactivate both the input and button components, preventing the user from typing anything while waiting for a response. The method concludes with a code for deleting the text from the input. It uses the widget.prevent context manager to prevent events from being emitted when the text inside the input changes.

The handlers do not currently include any chat-related code, but they do respond appropriately and have access to the text and widgets. The next step is to display messages.

Adding Components Dynamically

For the chat to be functional, messages need to appear on the screen. Adding content dynamically can often be tricky when working with UI frameworks. However, with Textual it is no rocket science. You only need a container widget to which you will be adding child components. Additionally, chatui will also have a custom widget for the message, so that it's a little fancier than just a text:

...  # trunkated imports
from textual.widgets import Footer, Header, Input, Button, Static
from textual.containers import Horizontal, Container

class MessageBox(Widget):
    def __init__(self, text: str, role: str) -> None:
        self.text = text
        self.role = role
        super().__init__()

    def compose(self) -> ComposeResult:
        yield Static(self.text, classes=f"message {self.role}")

class ChatApp(App):
    ...  # trunkated code

    def compose(self) -> ComposeResult:
        yield Header()
        yield Container(id="conversation_box")  # 🆕
        with Horizontal(id="input_box"):
            yield Input(placeholder="Enter your message", id="message_input")
            yield Button(label="Send", variant="success", id="send_button")
        yield Footer()

    async def process_conversation(self) -> None:
        message_input = self.query_one("#message_input", Input)
        # Don't do anything if input is empty
        if message_input.value == "":
            return
        button = self.query_one("#send_button")
        conversation_box = self.query_one("#conversation_box")  # 🆕

        self.toggle_widgets(message_input, button)

        # 🆕 Create question message, add it to the conversation and scroll down
        message_box = MessageBox(message_input.value, "question")
        conversation_box.mount(message_box)
        conversation_box.scroll_end(animate=False)

        # Clean up the input without triggering events
        with message_input.prevent(Input.Changed):
            message_input.value = ""

        # 🆕 Add answer to the conversation
        conversation_box.mount(
            MessageBox(
                "Answer",
                "answer",
            )
        )

        self.toggle_widgets(message_input, button)
        conversation_box.scroll_end(animate=False)  # 🆕

Similarly to the last example, components need a little bit of styling in the chatui/styles.css file:

/* ... trunkated styling */

MessageBox {
    layout: horizontal;
    height: auto;
    align-horizontal: center;
}

.message {
    width: auto;
    min-width: 25%;
    border: tall black;
    padding: 1 3;
    margin: 1 0;
    background: #343a40;
}

.question {
    margin: 1 25 1 0;
}

.answer {
    margin: 1 0 1 25;
}

The output after launching the app and sending a few messages should be the following:

The conversation UI in the terminal is alive. Well, at least the part about asking questions is alive because the response is for now static. In any case, a little bit of code was added to the app. First goes a widget called Container, and it is simply a vertical container for holding additional widgets. As you can see on the screen, it occupies the entire height of the space between the header and the input box. As a result, only that portion is scrollable and the input box is glued to the bottom.

Then, there is a new, custom MessageBox widget that serves as a container for the static text. To render the component within properly, it takes two additional parameters. Thanks to that, Static widget scales and positions correctly and it has additional styling based on the provided role CSS class. With that in place, the process_conversation method is now inserting MessageBox components into the conversation container, by using mount method. Presence of scroll_end method is also worth noting - it is used to scroll down to the end of the conversation container whenever a new message appears.

At this point, it appears that ChatApp has most of the UI elements and behaviors needed to support a genuine conversation with the ChatGPT. The following step is to connect it with the previously developed Conversation class.

Connecting TUI with ChatGPT

Now that the UI is complete, ChatGPT can be finally integrated into the TUI. The ChatApp class must use the actual Conversation instance to accomplish this:

...  # trunkated imports
from chatui.chat import Conversation

...  # trunkated code

class ChatApp(App):
    ...  # trunkated code

    def on_mount(self) -> None:
        self.conversation = Conversation()
        self.query_one("#message_input", Input).focus()

    async def process_conversation(self) -> None:
        ...  # trunkated code down to cleaning up the input

        # 🆕 Take answer from the chat and add it to the conversation
        choices = await self.conversation.send(message_box.text)
        self.conversation.pick_response(choices[0])
        conversation_box.mount(
            MessageBox(
                choices[0].removeprefix("\\n").removeprefix("\\n"),
                "answer",
            )
        )

        self.toggle_widgets(message_input, button)
        conversation_box.scroll_end(animate=False)

The new code begins with the on_mount handler for the Mount event, which is triggered when the application is first mounted. By doing so, the app initializes a new conversation and additionally focuses the input component so that the user can start typing immediately. In the process_conversation method, mounting static answer messages was replaced with sending questions to ChatGPT and rendering the actual response from the model. Because of AI occasionally responding with line-broken messages, you may notice double remove_prefix call, to clean up the text. This could, of course, be replaced with something more reliable. All of this together leads to pretty much the final result:

Finally, a genuine chatbot experience, with Textual managing the TUI and Conversation object handling communicating ChatGPT model and persisting context.

Usability Improvements

While the end result may be acceptable at this stage, there are a few things that could be added or improved. To begin, it would be wonderful to be able to leave the app using a method other than the CTRL+C combo. Another useful feature would be to clear the entire dialogue. Both of these can be accomplished by incorporating a custom key binding into the app. Nevertheless, there is one issue: because the input component is continually focused, key bindings may not work as planned. To address this issue and add additional actions using key bindings, the following changes must be made:

...  # trunkated imports
from textual.binding import Binding

class FocusableContainer(Container, can_focus=True):  # 🆕
    ...

class MessageBox(Widget, can_focus=True):  # 🆕
    ...  # trunkated code

class ChatApp(App):
    ...  # trunkated code

    BINDINGS = [  # 🆕
        Binding("q", "quit", "Quit", key_display="Q / CTRL+C"),
        ("ctrl+x", "clear", "Clear"),
    ]

    def compose(self) -> ComposeResult:
        yield Header()
        with FocusableContainer(id="conversation_box"):
            yield MessageBox(  # 🆕
                "Welcome to chatui!\\n"
                "Type your question, click enter or 'send' button "
                "and wait for the response.\\n"
                "At the bottom you can find few more helpful commands.",
                role="info",
            )
        with Horizontal(id="input_box"):
            yield Input(placeholder="Enter your message", id="message_input")
            yield Button(label="Send", variant="success", id="send_button")
        yield Footer()

    def action_clear(self) -> None:  # 🆕
        self.conversation.clear()
        conversation_box = self.query_one("#conversation_box")
        conversation_box.remove()
        self.mount(FocusableContainer(id="conversation_box"))

The first thing you can notice is that Container has been replaced with a custom FocusableContainer, which, like the MessageBox widget, is now subclassed with the can_focus=True option. As a result, these components that take up space between the header and input box can now receive focus after being clicked, causing the Input widget to lose focus. That adjustment allows for the use of custom key bindings without entering random letters into the input field.

The second change is the previously mentioned key bindings, which are configured as a class attribute. Textual custom key bindings are often stated as a three-element tuple, with the key/key combination coming first, the action name coming second, and the description coming last. This is the case with the newly added custom clear action. But, if you wish to further tweak the binding, you can utilize the Binding class which accepts a few more parameters. This is the case with the quit action, which in addition to q key, can also be triggered with a CTRL+C combination, and hence both options should be displayed.

Coming back to the clear action, you can see that it's declared by the action_clear method, which is then recognized by the key binding. Its purpose is to simply clear the Conversation state and remount the conversation container.

The final update is a new MessageBox that will appear at the beginning of the conversation. It is there to fill the empty space with a little bit of introduction. To style it differently than the other messages, a new CSS class was introduced:

.info {
    width: auto;
    text-align: center;
}

With all the new pieces together, the ultimate output of the app will look as follows:

As you can see, the key bindings are explained in the footer, the helper message is located directly in the middle, and you can click anywhere to switch the focus away from the input component.

Summary

Textual is genuinely a remarkable framework. It enables the creation of beautiful and powerful TUI apps while maintaining clean and current code. Even though TUI apps aren't so widely used, perhaps as a result of excellent tools and libraries like Textual, we will start to see them more frequently. They have already proven handy for developing some simple and semi-advanced user interfaces for various backend tasks. In either case, creating a TUI app is unquestionably an exciting experience that can result in creating something truly original, and Textual might be the best route to take.

Next Steps

While the app met the basic requirements, it had the potential to be much more. Further features may include, for example, keeping the conversation in a database and moving between them via tabs or a sidebar. It could also handle multiple choices supplied by the ChatGPT, allowing the user to select the one he prefers. The user interface might also be made more sophisticated and user-friendly. The major goal here, though, was to demonstrate the most intriguing and crucial features of Textual and how to leverage them to create something interesting. Ideally, the essay laid the groundwork for something even bigger, finer, and more powerful.

Sources

Code for the chatui project used in the article can be found here: link.

I also highly encourage you to take a look at the Textual documentation, code examples, and Will's Twitter.

5 Python Must-Know Packages

Piotr Tobiasz — Thu, 20 Jul 2023 06:51:13 +0000

Saying "you must know it" in pretty much any field of science can be quite contentious. There is no exception in software development. You may frequently see tweets or posts stating things like "you must learn Blockchain if you don't want to be left behind" or "you must know Kubernetes because it is so popular right now." Spoiler: You don't. However, if you want to be an expert in your field, there might be some topics that are almost universal or are used so frequently that it can be challenging without at least a fundamental understanding.

Python has gathered an incredibly strong community after over 30 years of its existence. As a result, Python has a plethora of well-known libraries and frameworks. Only by looking at the list of the most popular backend frameworks, you can see 2 Python libraries - Django and Flask. Right behind the corner, there is also FastAPI, which, given its popularity, will probably overtake Flask. With numerous excellent libraries available, there must be at least a few that are universal, or ubiquitous enough to qualify as "must-knows".

Managing Linters with pre-commit

Writing code is, of course, a major part of software development. Fortunately, formatting and style maintenance are not a major part of writing the code. There are numerous tools that lint the code to ensure that it appears consistently throughout the project, and determine whether the style guide is being followed. You've probably seen or used at least one of these libraries:

black
flake8
isort
mypy
ruff (recently, my favorite one)

Each of them has its own responsibilities, but in general, they are here to ensure that the code is properly styled and structured. The problem is that by default you need to install and run them separately and manually. This is time-consuming and suboptimal. Luckily, there is a tool for managing and running them all together: pre-commit.

As the name implies, the main idea behind pre-commit is to run your code checks right before you commit the code, so that changes can be made when it is not too late. However, the tool is more powerful than simply running x and y when a git commit occurs. It enables you to manage linters and checkers via a .pre-commit-config.yaml file rather than installing them as project dependencies. It handles caching installed tools as well as updating them. Here's an example of such a yaml file:

repos:
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.4.0
    hooks:
      - id: trailing-whitespace
      - id: end-of-file-fixer
      - id: mixed-line-ending
      - id: check-json
      - id: check-toml
      - id: check-yaml

  - repo: https://github.com/psf/black
    rev: 22.12.0
    hooks:
      - id: black
        name: Black

  - repo: https://github.com/charliermarsh/ruff-pre-commit
    rev: v0.0.185
    hooks:
      - id: ruff
        name: Ruff
        args: ["--fix"]

  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v0.991
    hooks:
      - id: mypy
        name: MyPy
        exclude: ^.*\b(migrations)\b.*$
        additional_dependencies: ["types-requests", "types-redis"]

All you have to do is specify which repositories (hooks) you want 'pre-commit' to install and run. There are four hooks visible here:

Basic ones, provided by pre-commit, which address some of the common issues,
black for code formatting,
ruff for enforcing code style guide,
mypy for static type checking.

pre-commit also enables you to add some extra arguments or dependencies for each hook, as seen in the ruff and mypy cases. All hooks can, of course, be further customized with a config file, such as pyproject.toml.

With pre-commit installed and .pre-commit-config.yaml in place, you should be able to run pre-commit run --all-files command which will install and run the hooks. Here is an example output:

$ pre-commit run --all-files
trim trailing whitespace.................................................Passed
fix end of files.........................................................Passed
mixed line ending........................................................Passed
check json...............................................................Passed
check toml...............................................................Passed
check yaml...............................................................Passed
Black....................................................................Passed
Ruff.....................................................................Passed
MyPy.....................................................................Passed

All the checks passed, and you can now commit & push the code to the remote repository, without wasting money on failed CI checks.

Writing Tests Using pytest

Tests are a must-know not only in Python but in software development in general. They verify that your code meets the functional requirements. They give you the assurance that you won't alter the working code. Unit tests are used to verify the smallest pieces of code, integration tests examine the integration with dependencies, and end-to-end tests check the complete processes.

unittest is a standard library in Python used for creating tests, providing both basic and advanced testing tools. However, relying solely on unittest may not be optimal in the long run as it may lead to a lot of redundant "util" code, a lack of suitable extensions, and difficulties in creating tests, ultimately limiting scalability. Fortunately, pytest is a popular testing framework that addresses these limitations and provides several benefits for developers.

pytest allows developers to focus on testing their code without writing any extraneous code that is irrelevant to the test. Unlike unittest, you don't need to create a TestCase-like class or manually mark the tests - pytest will automatically detect them. Additionally, you don't need to use specific assertion functions like assertEqual as pytest makes use of Python's built-in assert keyword. Lastly, pytest offers a more efficient and effective way of managing the setup and teardown dependencies through fixtures.

Fixtures are one of the most important pytest features. According to pytest documentation, fixtures are defined as follows:

In testing, a fixture provides a defined, reliable and consistent context for the tests.

In practice, a fixture is written as a function that provides the necessary resources to make a test functional. For example, in web development, a common fixture is a database connection. When testing something that involves requests to the database, a fixture can be created that starts the connection, passes it to the test, and then cleans and closes the connection after the test is completed.

For the sake of simplicity consider a case of an in-memory database, that is aware of "being connected":

from typing import Any
from must_knows import exceptions

class Database:
    """Simple, in-memory database."""

    def __init__(self) -> None:
        self._entries: list[Any] = []
        self._connected = False

    def connect(self) -> None:
        self._connected = True

    def disconnect(self) -> None:
        self._connected = False

    def add(self, entry: Any) -> None:
        if self._connected is False:
            raise exceptions.DatabaseError("Database is disconnected")
        self._entries.append(entry)

    @property
    def entries(self) -> list:
        if self._connected is False:
            raise exceptions.DatabaseError("Database is disconnected")
        return self._entries

This database will be utilized in a user service - a simple class that exposes only the most basic operations:

import uuid
from dataclasses import dataclass
from must_knows import exceptions, database

@dataclass
class User:
    """User model."""

    id: uuid.UUID
    email: str
    password: str

class UserService:
    """User operations manager."""

    def __init__(self, db: database.Database) -> None:
        self.db = db

    def register(self, email: str, password1: str, password2: str) -> None:
        if password1 != password2:
            raise exceptions.ValidationError("Password must match")
        user = User(id=uuid.uuid4(), email=email, password=password1)
        self.db.add(user)

    def login(self, email: str, password: str) -> uuid.UUID:
        user = next((u for u in self.db.entries if u.email == email), None)
        if user is None or user.password != password:
            raise exceptions.LoginError("Invalid credentials")
        return user.id

If you were to test this class without using pytest, you would have to manually create the database object and execute the connect method for each test. Additionally, you would need to implement a teardown method that runs the disconnect method after each test. This process can become cumbersome, especially if the connect and disconnect methods take a longer time to execute or if there are failures that prevent the database from disconnecting properly. This is where fixtures come to the rescue:

# tests/conftest.py
import pytest
from typing import Generator
from must_knows import database

@pytest.fixture()
def connected_db() -> Generator[database.Database, None, None]:
    """Yield connected database.

    Disconnect on teardown.
    """
    db = database.Database()
    db.connect()
    yield db
    db.disconnect()

# tests/test_user_service.py
from must_knows.main_dataclass import UserService

class TestUserService:
    """Test cases for the UserService class."""

    def test_register(self, connected_db):
        srv = UserService(db=connected_db)
        srv.register("test@user.com", "testpassword", "testpassword")
        uid = srv.login("test@user.com", "testpassword")
        assert uid is not None

To use fixtures in pytest, all you need to do is define them in the conftest.py file, and pytest will automatically recognize them by name, eliminating the need to import anything. One key aspect of fixtures is that they are structured as Python generators. The code before the yield statement is executed before the test, while the code after yield serves as the teardown, which runs after the test completes or if the test fails. This means that the test itself doesn't need to include any code related to the database connection, nor does it need to worry about the cleanup process after completion.

The fixture decorator also includes a scope parameter that specifies when the fixture should be initialized and torn down. By default, the scope is set to function, which implies the fixture is initialized and torn down before and after every test. However, you can modify this behavior by setting the scope to session, which ensures that the same database is utilized for all tests and is only disconnected once all tests are finished.

Another nice feature of pytest is the abundance of excellent plugins. One of the most popular is definitely pytest-cov, which computes test coverage and generates a report. It can afterward be used to detect untested code or to brag about the great quality of the tests (albeit high coverage does not always imply quality).

Data Structures with pydantic

Python is mostly used in data science and web development, but no matter what you want to accomplish with it, you will eventually need to parse and validate some data. Of course, you may start from scratch and implement everything, as you can in pretty much any circumstance. You can also use a dataclass module, a Python's builtin. Take a look at the class from the prior example:

import uuid
from dataclasses import dataclass, asdict

@dataclass
class User:
    id: uuid.UUID
    email: str
    password: str

user = User(id=uuid.uuid4(), email="test@user.com", password="testpass")
user_dict = asdict(user)

Since implementing __init__ function is no longer required, using the dataclass module is simple and reduces the need for extra code. A dataclass instance may also be easily transformed into a dictionary. Dataclasses might not be adequate, though, when more sophisticated functionality is needed. Several limitations are made clear even in this straightforward scenario.

First of all, no real validation is taking place. Even though a password should be a string, you may still use an integer and everything will work out. Not to mention the email field, which should take a genuine email address but may be filled with whatever you choose. Also, you often want the id field to be filled up automatically if nothing is provided. In this manner, the user service's "register" method wouldn't need to create the id. Such behavior is actually possible with the dataclasses:

import uuid
from dataclasses import dataclass, field

@dataclass
class User:
    email: str
    password: str
    id: uuid.UUID = field(default_factory=uuid.uuid4)

user = User(email="test@user.com", password="testpass")

By using the field function and providing a factory to generate a default value, you can make the id field populate automatically when it is not provided. However, you may have noticed that the id field had to be moved to the end of the class. This is because, in dataclasses, fields with default values cannot appear before fields without default values. While this may seem like a minor inconvenience, it can become frustrating when dealing with longer classes where you want to group values together in a more meaningful way. The pydantic package offers a more powerful solution to help you deal with data structures.

pydantic is a library whose main purpose is data parsing and validation. It is much more powerful than dataclasses, though. With pydantic, you construct a new model using a BaseModel class and then specify the fields with type hints:

import uuid
from pydantic import BaseModel, EmailStr, Field

class User(BaseModel):
    id: uuid.UUID = Field(default_factory=uuid.uuid4())
    email: EmailStr
    password: str = Field(min_length=8, max_length=32)

user = User(email="test@user.com", password="testpass")
user_dict = user.dict()

Upon closer inspection, you'll notice that pydantic operates quite differently. Firstly, unlike with dataclasses, you can place the id field at the beginning of the class without any issues. Similar to dataclasses, pydantic offers a Field function that allows you to configure individual fields. Additionally, pydantic provides extra types that can be used to validate input. In this example, the EmailStr type is utilized to ensure that the email field is a valid email address. While your IDE may prefer email=EmailStr("test@user.com"), instead of the email="test@user.com", pydantic can handle both cases seamlessly. Finally, pydantic validates not only the default types but also any constraints specified in the Field function. For instance, in this example, the password field must be a string between 8 and 32 characters in length, effectively disallowing anything else.

Furthermore, pydantic allows you to quickly add your own validators. Recall the register method once again. It would only create a user if the passwords were the same. Although this logic fits well in the register method, it may alternatively be placed in the user model:

import uuid
from pydantic import BaseModel, EmailStr, root_validator, Field
from must_knows import database

class User(BaseModel):
 """User model."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4)
    email: EmailStr
    password: str = Field(min_length=8, max_length=32)
    password_confirmation: str

    @root_validator(pre=True)
    def validate_passwords(cls, values: dict) -> dict:
        if values["password"] != values["password_confirmation"]:
            raise ValueError("Passwords must match")
        return values

class UserSerivce:
    """User operations manager."""

    def __init__(self, db: database.Database) -> None:
        self.db = db

    def register(self, email: EmailStr, password1: str, password2: str) -> None:
        user = User(email=email, password=password1, password_confirmation=password2)
        self.db.add(user)

This way, pydantic first checks if the passwords match, and then if the password field matches the initial criteria, i.e., it has to be a string with a valid length. You can also write validators for a specific field only using the validator decorator. However, in this case, validity depended on two fields simultaneously, which is why the root_validator was used - because it has access to all values. If you want to read more about validators or how powerful pydantic is in general, be sure to check out the official documentation - it's just as great as the library itself.

Better Requests Using httpx

Regardless of what software you are building, you will likely end up doing some requests. The essential capabilities for dealing with external APIs are provided by the standard Python requests module. Nevertheless, it does not support a number of essential Python features.

httpx is a modern, alternative HTTP client for Python, which offers a more intuitive and user-friendly API than the standard requests library. The most significant feature of httpx is its async support. It can handle HTTP requests asynchronously, which is beneficial for making several requests at the same time without blocking the main thread. This functionality can improve the performance of applications that require high network I/O. httpx is also built on top of the standard Python typing module, which allows for better code readability and clarity.

httpx provides an elegant and straightforward API to make HTTP requests:

import asyncio
from httpx import AsyncClient

async def main() -> None:
    client = AsyncClient(base_url="https://pokeapi.co/api/v2")

    # Two calls, one after another
    r1 = await client.get("/pokemon/ditto")
    print(f"r1: {r1.status_code}")  # 200
    # You can use 'request' for better configurability
    r2 = await client.request("GET", "/pokemon/jynx")
    print(f"r2: {r2.status_code}")  # 200

    # Two calls made asynchronously
    r3, r4 = await asyncio.gather(
        client.get("/pokemon/lucario"),
        client.get("/pokemon/snorlax"),
    )
    print(f"r3: {r3.status_code}")  # 200
    print(f"r4: {r4.status_code}")  # 200

    # Call resulting in 404 and raising an 'HTTPStatusError'
    r5 = await client.get("/pokemon/john-wick")
    print(f"r5: {r5.status_code}")  # 404
    r5.raise_for_status()

if __name__ == "__main__":
    asyncio.run(main())

The preceding example is clearly quite simple, but what's also great about httpx is how easily configurable it is. This includes headers, cookies, authentication, and other request parameters:

async def configurability_example() -> None:
    client = AsyncClient(
        base_url="https://pokeapi.co/api/v2",
        auth=("ash", "ketchum123"),
        cookies={"access_token": "Bearer PIKACHU"},
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json",
        },
    )

    # All requests share the same configuration
    await client.get("/pokemon/ditto")
    await client.request("GET", "/pokemon/jynx")
    await client.get("/pokemon/snorlax")

All of this makes httpx an easy and powerful way for Python scripts to interact with the APIs.

Dynamic HTML with Jinja

As previously mentioned, Python is primarily used for data science and backend development, which means that frontend tasks are rarely performed within Python scripts. However, there are common scenarios where a little bit of frontend functionality is needed in a backend task. A good example can be email notifications, where you want to notify a user when something occurs on the backend and include details about the event in the email. In most cases, it's desirable to use HTML format for the email, allowing for structured and styled content.

To address this need, Jinja was created by Armin Ronacher, the creator of Flask and click. Jinja is a templating engine that allows for working with HTML files in Python scripts. However, the capabilities of Jinja go beyond simple file manipulation - it's a powerful tool for creating dynamic HTML content in a Python environment.

One of the primary benefits of using Jinja is the ability to use templates. Templates allow you to design a structure for your HTML files that may be reused. This is especially beneficial if you have a consistent layout for different emails or web pages.

To use Jinja, you first need to define a template. This is usually an HTML file with some placeholders for the dynamic content. For example, you could have a template for a simple email like this:

<html>

<body>
    <h1>{{title}}</h1>
    <p>{{content}}</p>
</body>

</html>

In this template, {{title}} and {{content}} are placeholders that will be replaced with actual values at runtime. To render the template, you need to create a Jinja environment and load the template from a file:

from jinja2 import Environment, FileSystemLoader

env = Environment(loader=FileSystemLoader("must_knows/templates/"))

template = env.get_template("example.html")
data = {"title": "Hello", "content": "World!"}
html = template.render(data)
print(html)

In the example above, the FileSystemLoader is used to load templates from the templates directory. Once you have the environment and the template, you can render it with some data. The render method replaces the placeholders in the template with the values from the data dictionary and returns the rendered HTML.

Another useful feature is the ability to establish "base" templates that may be extended by more customized templates. In this manner, you may put shared code in the base template and reuse it in all templates. Then you simply provide the blocks that will be injected into particular templates:

<!-- base.html -->
<html>

<body>
    <h1>{% block title %}{% endblock %}</h1>

    <p>{% block content %}{% endblock %}</p>

    <footer>{% block footer %}Made using @Jinja{% endblock %}</footer>
</body>

</html>

<!-- extended_example.html -->
{% extends "base.html" %}

{% block title %}{{title}}{% endblock %}

{% block content %}{{content}}{% endblock %}

Of course, Jinja is considerably more powerful than merely changing placeholders. You may build intricate templates with conditionals, loops, macros, and filters to suit a variety of use situations. Also, it works nicely with Django and Flask, two other Python web frameworks. The most common use cases include generating reports and emails, but it can even be used for building web pages.

Summary

Python is a versatile and user-friendly language that offers a wealth of libraries for virtually any type of use case. Identifying packages that are considered "must-know" might, however, be challenging. Backend developers might recommend Flask or FastAPI, while data scientists would suggest Tensorflow or pandas. While these packages are undoubtedly valuable in their respective fields, they may not be suitable for everyone. Presented packages are just a small fraction of the countless packages available in the Python ecosystem, but they are among the most important and extensively used. By learning and mastering these packages, you can become a more proficient Python developer, regardless of your field.

Sources

Code examples used in the article can be found here: link.

Documentation page of each library:

Practical Introduction to Domain-Driven Design

Piotr Tobiasz — Wed, 19 Jul 2023 18:06:46 +0000

It should not be innovative to say that writing software is not merely about writing code - it is about solving a particular problem. Even though it's developers who eventually implement the solution, it is not developers who define what is the problem in the first place. That task is carried out by various business people, who consider processes, risks, and outcomes to describe what the problem is, why it exists, and how it should be addressed. In a domain-driven context, these business people are referred to as Domain Experts.

From an engineering perspective, it appears that Domain Experts hold a valuable asset: their knowledge about the domain. However, this knowledge is rarely shared in its raw form. Instead, it's usually translated into requirements so that developers can understand and implement it. The issue with this approach is that the domain knowledge of business people and developers can diverge. This means that the perspectives of those who define the problem and those who work on solving it may not align, leading to misunderstandings and conflicts.

So, what's the way out? Make sure that business and technical people use the same language and terminology.

What is Domain-Driven Design?

Domain-Driven Design (DDD) is a methodology that emphasizes the importance of creating a shared understanding between domain experts and technical stakeholders and aligning the software solution with the underlying business requirements. This appears to be a high-level, non-technical definition, but it can also be broken down into something more developer-friendly:

DDD is representing real-world concepts in the code, and its structure by cultivating and using ubiquitous language, that is built by modeling the business domain.

There is still some terminology to be introduced, so it might not be 100% clear right now. What's most important is that DDD provides tools and activities that allow writing and structuring code that is aligned with the business vision. It is then not only about communication but also about making design decisions, that actually shape the common language.

Terminology

It should come as no surprise that the most important term in the DDD world is the Domain. Here is how Vlad Khononov, author of "Learning Domain-Driven Design" describes it:

A business domain defines a company's main area of activity.

It means that the business domain could also be considered as:

The main source of a company's revenue,
What the company is best known for,
Anything that the company does better than its competitors.

The domain can also be divided into subdomains - more specific ranges of activities. While there are three different types of subdomains, the most important one is the core. It describes how the company achieves the business advantage. The other two are about more common, generic problems, like authentication systems or internal admin panels.

Having a deep understanding of a company's business domain is absolutely crucial for fully utilizing the benefits of Domain-Driven Design. The best source of this understanding is no one else than the Domain Experts. These are the individuals whose problem is being addressed with the software - stakeholders, various business people, and even users. It is not to say that engineers are uninformed about the domain they are working on, but rather that the Experts are the source of truth of the domain knowledge. By working together with the Domain Experts, developers can ensure that the models of the domain remain accurate and up-to-date.

This leads to another critical yet potentially ambiguous term: model. Eric Evans, in his book about the DDD, describes the model as follows:

It is an interpretation of reality that abstracts the aspects relevant to solving the problem at hand and ignores extraneous detail.

Vlad Khononov goes on to explain this idea in terms that are more relatable:

A model is not a copy of the real world, but a human construct that helps us make sense of real-world systems.

In conclusion, a model is a representation of a business concept or process that facilitates the understanding and communication of the domain's underlying complexity.

Vlad used a map to illustrate the concept of a domain model effectively. Maps are a perfect example of how they only display information that is relevant to the type of map, like topography, roads, or borders. A map that displays all the details at once would be overwhelming and pretty much useless. Domain models can also be found in other forms, such as:

Customer orders, which represent the simplified version of all the processes happening in the background,
Restaurant menus, where the items listed on the menu are the final products, instead of listing every ingredient and step in the preparation process,
Travel bookings, where the booked trip only highlights the most critical details, even though much more goes into planning the travel, hotel, etc.

The last piece of the Domain-Driven Design (DDD) terminology puzzle is the Ubiquitous Language. It refers to the shared language used by both technical and business stakeholders in a project. Having a common language to describe the business domain, derived from the domain model, is crucial in DDD. It helps ensure that all team members have a clear understanding of the problem space, its concepts, and their relationships. This leads to better alignment and reduces the risk of misunderstandings. By using Ubiquitous Language, the software solution can accurately reflect the underlying business requirements, making it a critical component of DDD.

With most of the terminology covered, it should be easier to understand what is Domain-Driven Design. Now it's time to delve into the actual how - the building blocks of the DDD.

Building Blocks

DDD building blocks serve as the foundation for creating an effective and efficient Domain Model. Vlad Khononov defines the Domain Model in the following way:

A domain model is an object model of the domain that incorporates both behavior and data.

Domain Model consists of various building blocks and structures. The most important ones are:

Value objects,
Entities,
Aggregates,
Domain events,
Repositories,
Domain Services.

Value Objects

Value Objects are the most basic building blocks available. These are objects that are defined by a set of attributes and values. They don't have a unique identifier - their values define their identity. They are immutable in the sense that different values are already representing a different value object. Examples of Value Objects include:

Monetary amount,
Date range,
Postal address.

Here is how a simple Value Object could be implemented in Python:

from pydantic import BaseModel

class Address(BaseModel):
    """Customer address."""

    country: str
    city: str
    street: str
    house_number: str

    class Config:
        frozen = True

It means that using the equality operator (==) to compare two addresses will return True only if both objects have exactly the same values assigned.

Entities

Entities are the next type of building block. Entities represent individual objects in the domain with a distinct identity, such as a person or an order. They are similar to the Value Objects in the way that they also store the data, but their attributes can, and are expected, to change, and thus they need a unique identifier. Orders and personal information are just two simple instances of the Entities:

import uuid
from pydantic import BaseModel, Field
from practical_ddd.building_blocks.value_objects import Address

class Person(BaseModel):
    """Personal data."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4)
    first_name: str
    last_name: str
    address: Address

class Order(BaseModel):
    """Customer order."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4)
    description: str
    value: float

Because the values of the instances are modifiable in both cases, they need an identification, which can be a UUID. What's more important is that in most cases, entities are not meant to be managed directly, but through an aggregate.

Aggregate

An Aggregate is a type of entity because it is mutable and requires a unique identifier. Its primary responsibility, however, is not to store data, but to group a set of related objects (Entities and Value Objects) together as a single unit of consistency. The Aggregate is the root object, with a well-defined boundary that encapsulates its internal state and enforces invariants to ensure the consistency of the whole group. Aggregates allow reasoning about the domain in a more natural and intuitive way, by focusing on the relationships between objects rather than the objects themselves.

Following on from the preceding examples, an aggregate could be represented as a customer:

import uuid
from pydantic import BaseModel, Field
from practical_ddd.building_blocks.entities import Person, Order
from practical_ddd.building_blocks.value_objects import Address

class Customer(BaseModel):
    """Customer aggregate.

    Manages personal information as well as orders.
    """

    id: uuid.UUID = Field(default_factory=uuid.uuid4)
    person: Person
    orders: list[Order] = Field(default_factory=list)

    def change_address(self, new_address: Address) -> None:
        self.person.address = new_address

    def add_order(self, order: Order) -> None:
        if self.total_value + order.value > 10000:
            raise ValueError("Order cannot have value higher than 10000")
        self.orders.append(order)

    def remove_order(self, order_id: uuid.UUID) -> None:
        order = next((order for order in self.orders if order.id == order_id), None)
        if order is None:
            raise IndexError("Order not found")
        self.orders.remove(order)

    @property
    def total_value(self) -> float:
        return sum(order.value for order in self.orders)

The customer is directly linked to the personal data and it stores all the orders. On top of that, the aggregate exposes an interface for managing the person's address as well as adding and removing orders. This is due to the fact that the aggregate's state can only be changed by executing the corresponding methods.

While the previous example is relatively straightforward, with only one constraint (order value cannot be greater than 10000), it should demonstrate the use of DDD building blocks and their relationships. In actual systems, aggregates are often more complex, with more constraints, boundaries, and possibly more relationships. After all, their very existence is to manage this complexity. Additionally, in the real world, aggregates would typically be persisted in a data store, such as a database. This is where the repository pattern comes into play.

Repository

The aggregate's state changes should all be committed transactionally in a single atomic operation. However, it is not the aggregate's responsibility to "persist itself". Repository pattern allows to abstract away the details of data storage and retrieval, and instead, work with aggregates at a higher level of abstraction. Simply put, a repository can be considered as a layer between aggregate and data storage. A JSON file is a fairly simple example of such a store. Customer aggregate could have a repository that operates on JSON files:

import json
import uuid
from practical_ddd.building_blocks.aggregates import Customer

class CustomerJSONRepository:
    """Customer repository operating on JSON files."""

    def __init__(self, path: str) -> None:
        self.path = path

    def get(self, customer_id: uuid.UUID) -> Customer:
        with open(self.path, "r") as file:
            database = json.load(file)
            customer = database["customers"].get(str(customer_id))
            if customer is None:
                raise IndexError("Customer not found")

            person = database["persons"][str(customer["person"])]
            orders = [database["orders"][order_id] for order_id in customer["orders"]]

        return Customer(
            id=customer["id"],
            person=person,
            orders=orders,
        )

    def save(self, customer: Customer) -> None:
        with open(self.path, "r+") as file:
            database = json.load(file)
            # Save customer
            database["customers"][str(customer.id)] = {
                "id": customer.id,
                "person": customer.person.id,
                "orders": [o.id for o in customer.orders],
            }
            # Save person
            database["persons"][str(customer.person.id)] = customer.person.dict()
            # Save orders
            for order in customer.orders:
                database["orders"][str(order.id)] = order.dict()

            file.seek(0)
            json.dump(database, file, indent=4, default=str)

Of course, this class could (and perhaps should) do a lot more, but it's not intended to be a perfect, multifunctional ORM. It should give an idea about repository responsibilities, which in this case are storage and retrieval of the customer aggregate in the JSON file. It is also worth noting how the repository handles entities associated with the aggregate. Because personal data and orders are tightly linked to the customer lifecycle, they must be managed precisely when the aggregate is being processed.

Domain Service

Another case to consider is when there is business logic that simply does not fit into the aggregate or any of its entities or value objects. It could be logic that is dependent on multiple aggregates or the state of the data store. In such cases, a structure known as Domain Service can come in handy. Domain Service must be able to manage aggregates, for example, by using the repository, and then it can store domain logic that doesn't belong to the aggregate. For example, a customer may require logic to avoid losing too many orders:

import uuid
from typing import Protocol
from practical_ddd.building_blocks.aggregates import Customer

class CustomerRepository(Protocol):
    """Customer repository interface."""

    def get(self, customer_id: uuid.UUID) -> Customer:
        ...

    def save(self, customer: Customer) -> None:
        ...

class CustomerService:
    """Customer service."""

    def __init__(self, repository: CustomerRepository) -> None:
        self.repository = repository

    def get_customer(self, customer_id: uuid.UUID) -> Customer | None:
        try:
            return self.repository.get(customer_id)
        except IndexError:
            return None

    def save_customer(self, customer: Customer) -> None:
        existing_customer = self.get_customer(customer.id)
        # If customer is already in the database and has more than 2 orders,
        # he cannot end up with half of them after a single save.
        if (
            existing_customer is not None
            and len(existing_customer.orders) > 2
            and len(customer.orders) < (len(existing_customer.orders) / 2)
        ):
            raise ValueError(
                "Customer cannot lose more than half of his orders upon single save!"
            )

        self.repository.save(customer)

Aggregate cannot ensure how its state differs from the one in the JSON file because it has no knowledge of the JSON file in the first place. That is why the comparison logic must be included in the Domain Service. It is also important to note that Domain Service should work with repository abstraction. This makes it simple to swap out the concrete implementation with an alternative one, by using dependency injection.

Putting it All Together

With all the pieces have now been covered, they can be now seen as a working program:

import uuid
from practical_ddd.building_blocks import aggregates, entities, value_objects
from practical_ddd.database.repository import CustomerJSONRepository
from practical_ddd.service import CustomerService

# Initialize domain service with json repository
srv = CustomerService(repository=CustomerJSONRepository("test.json"))

# Create a new customer
customer = aggregates.Customer(
    person=entities.Person(
        first_name="Peter",
        last_name="Tobias",
        address=value_objects.Address(
            country="Germany",
            city="Berlin",
            street="Postdamer Platz",
            house_number="2/3",
        ),
    ),
)
srv.save_customer(customer)

# Add orders to existing customer
customer = srv.get_customer(uuid.UUID("a32dd73a-6c1b-4581-b1d3-2a1247320938"))
assert customer is not None
customer.add_order(entities.Order(description="Order 1", value=10))
customer.add_order(entities.Order(description="Order 2", value=210))
customer.add_order(entities.Order(description="Order 3", value=3210))
srv.save_customer(customer)

# Remove orders from existing customer
# If there are only 3 orders, it's gonna fail
customer = srv.get_customer(uuid.UUID("a32dd73a-6c1b-4581-b1d3-2a1247320938"))
assert customer is not None
customer.remove_order(uuid.UUID("0f3c0a7f-67fd-4309-8ca2-d007ac003b69"))
customer.remove_order(uuid.UUID("a4fd7648-4ea3-414a-a344-56082e00d2f9"))
srv.save_customer(customer)

Everything has its responsibilities and boundaries. Aggregate is in charge of managing its entities and value objects, as well as enforcing its constraints. Domain Service uses the injected JSON repository to persist the data in the JSON file and enforce additional domain boundaries. In the end, each component has a distinct function and significance within the specified domain.

Summary

Domain-Driven Design is without any doubt a complex idea to grasp. It provides practices, patterns, and tools to help software teams tackle the most challenging business problems by placing a strong emphasis on the business domain. DDD is more than just a set of building blocks, however. It is a mindset that requires collaboration and communication between technical and business stakeholders. A shared understanding of the domain, expressed through ubiquitous language, is critical to the success of a DDD project. When done well, DDD can lead to software that is better aligned with the needs of the business and more effective at solving complex problems.

Afterword and the Next Steps

This article was never intended to be anything like "DDD: From Zero To Hero," but rather to serve as an introduction to the DDD universe. I wanted to demonstrate the most important concepts of Domain-Driven Design in a very straightforward and practical manner. I believe that learning Domain-Driven Design is an excellent way to boost programming expertise. However, you don't hear about it too often - at least not as much as "11 INSANE JavaScript tips and tricks - a thread 🧵". In any case, if you found any of this interesting, you can look through the sources section for books and articles that inspired me to write this article in the first place. There are some concepts I didn't cover because I thought they were beyond the scope of this introduction, but they are well worth investigating:

Bounded Contexts
Domain Events
Event Storming

You will undoubtedly find them in the sources listed below.

Sources

Code examples used in the article can be found here: link.

Learning Domain-Driven Design by Vlad Khononov. An amazing book that served as a major source of inspiration for me. Explains all of the concepts discussed in this article in greater depth.

Architecture Patterns in Python by Harry Percival and Bob Gregory. I read the book almost two years ago, and it had a significant impact on me as a developer. I went back to it while writing this article, and it helped me once more.

DDD in Python by Przemysław Górecki. I discovered this blog near the end of writing the article, but it piqued my interest because of how insanely professional it is. Fun fact: I worked in the same company as Przemysław, and I was completely unaware of it.