Domain Model

The concept of domain model can be explained by a simplistic definition of its terms:

• domain refers to the specific subject area of activity (or knowledge) that our software is being built to support for.
• model refers to a simple representation (or abstraction) of the system or process that we are trying to encode in our software.

Thus, the domain model is a fancy (but standard and useful) way to refer to the set of concepts and rules that business owners have in their mind about how the business works. This is what we also, and commonly, refer to as the business logic of the application, including the rules, constraints, and relationships that govern the behavior of the system.

We'll refer to the domain model as the model from now on.

Repository pattern

The repository pattern is a design pattern that allows for the decoupling the model from the data access. The main idea behind the repository pattern is to create an abstraction layer between the data access logic and the business logic of an application. This abstraction layer allows for the separation of concerns, making the code more maintainable and testable.

When implementing the repository pattern, we typically define an interface that specifies the standard methods that any other repository must implement (AbstractRepository). And, then, a particular repository is defined with the concrete implementation of these methods where the data access logic is implemented (e.g., SQLAlchemyRepository). This design pattern aims at isolating the data manipulation methods so that they can be used seamlessly elsewhere in the application, e.g. in our domain model.

Unit of work pattern

The unit of work pattern is the missing part to finally decouple the model from the data access. The unit of work encapsulates the data access logic and provides a way to group all the operations that must be performed on the data source within a single transaction. This pattern ensures that all the operations are performed atomically.

When implementing the unit of work pattern, we typically define an interface that specifies the standard methods that any other unit of work must implement (AbstractUnitOfWork). And, then, a particular unit of work is defined with the concrete implementation of these methods where the data access logic is implemented (e.g., SQLAlchemyUnitOfWork). This design allows for a systematic handling of the connection to the data source, without the need to change the implementation of the business logic of the application.

Data model

Our users data will be stored in a SQLite database (you can use any other database supported by SQLAlchemy). We'll use the following data model to represent the users (you can save this code in a file called models.py under the src folder):

# src/models.pyimport uuid
import sqlalchemyfrom flama.sqlalchemy import metadatafrom sqlalchemy.dialects.postgresql import UUID
__all__ = ["user_table", "metadata"]
user_table = sqlalchemy.Table(    "user",    metadata,    sqlalchemy.Column("id", UUID(as_uuid=True), primary_key=True, nullable=False, default=uuid.uuid4),    sqlalchemy.Column("name", sqlalchemy.String, nullable=False),    sqlalchemy.Column("surname", sqlalchemy.String, nullable=False),    sqlalchemy.Column("email", sqlalchemy.String, nullable=False, unique=True),    sqlalchemy.Column("password", sqlalchemy.String, nullable=False),    sqlalchemy.Column("active", sqlalchemy.Boolean, nullable=False),)

Besides the data model, we need a migration script to create the database and the table. For this, we can save the following code in a file called migrations.py at the root of the project:

# migrations.pyfrom sqlalchemy import create_engine
from src.models import metadata
if __name__ == "__main__":    # Set up the SQLite database    engine = create_engine("sqlite:///models.db", echo=False)
    # Create the database tables    metadata.create_all(engine)
    # Print a success message    print("Database and User table created successfully.")

And then, we can run the migration script by executing the following command:

> poetry run python migrations.py
Database and User table created successfully.

Repository

In this example we are going to need only one repository, namely the repository which will handle the atomic operations on the user table, the name of which will be UserRepository. Thankfully, flama provides a base class for repositories related to SQLAlchemy tables, called SQLAlchemyTableRepository.

The class SQLAlchemyTableRepository provides a set of methods to perform CRUD operations on the table, specifically:

• create: Creates new elements in the table. If the element already exists, it will raise an exception (IntegrityError), otherwise it will return the primary key of the new element.
• retrieve: Retrieves an element from the table. If the element does not exist, it will raise an exception (NotFoundError), otherwise it will return the element. If more than one element is found, it will raise an exception (MultipleRecordsError).
• update: Updates an element in the table. If the element does not exist, it will raise an exception (NotFoundError), otherwise it will return the updated element.
• delete: Deletes an element from the table.
• list: Lists all the elements in the table that match the clauses and filters passed. If no clauses or filters are given, it returns all the elements in the table. If no elements are found, it returns an empty list.
• drop: Drops the table from the database.

For the purposes of our example, we don't need any further action on the table, so the methods provided by the SQLAlchemyTableRepository are sufficient. We can save the following code in a file called repositories.py under the src folder:

# src/repositories.pyfrom flama.ddd import SQLAlchemyTableRepository
from src import models
__all__ = ["UserRepository"]
class UserRepository(SQLAlchemyTableRepository):    _table = models.user_table

As you can see, the UserRepository class is a subclass of SQLAlchemyTableRepository, and it only requires the table to be set in the _table attribute. This is the only thing we need to do to have a fully functional repository for the user table.

If we wanted to add custom methods beyond the standard CRUD operations, we could do so by defining them in the UserRepository class. For example, if we wanted to add a method to count the number of active users, we could do so as follows:

# src/repositories.pyfrom flama.ddd import SQLAlchemyTableRepository
from src import models
__all__ = ["UserRepository"]
class UserRepository(SQLAlchemyTableRepository):    _table = models.user_table
    async def count_active_users(self):        return len((await self._connection.execute(self._table.select().where(self._table.c.active == True))).all())

Although we won't be using this method in our example, it's good to know that we can add custom methods to the repository if needed, and how they are implemented in the context of the repository pattern. This is a powerful design pattern as we can already see, since we can implement here all the data access logic without having to change the business logic of the application (which is implemented in the corresponding resource methods).

Worker

The unit-of-work pattern is used to encapsulate the data access logic and provide a way to group all the operations that must be performed on the data source within a single transaction. In flama the UoW pattern is implemented with the name of Worker. In the same way as with the repository pattern, flama provides a base class for workers related to SQLAlchemy tables, called SQLAlchemyWorker. In essence, the SQLAlchemyWorker provides a connection and a transaction to the database, and instantiates all its repositories with the worker connection. In this example, our worker will only use a single repository (namely, the UserRepository) but we could add more repositories if needed.

Our worker will be called RegisterWorker, and we can save the following code in a file called workers.py under the src folder:

# src/workers.pyfrom flama.ddd import SQLAlchemyWorker
from src import repositories
__all__ = ["RegisterWorker"]

class RegisterWorker(SQLAlchemyWorker):    user: repositories.UserRepository

Thus, if we had more repositories to work with, for instance ProductRepository and OrderRepository, we could add them to the worker as follows:

# src/workers.pyfrom flama.ddd import SQLAlchemyWorker
from src import repositories
__all__ = ["RegisterWorker"]
class RegisterWorker(SQLAlchemyWorker):    user: repositories.UserRepository    product: repositories.ProductRepository    order: repositories.OrderRepository

As simple as that, we have implemented the repository and worker patterns in our application. Now, we can move on to implement the resource methods that will provide the API endpoints needed to interact with the user data.

Resources

Resources are one of the main building blocks of a flama application. They are used to represent application resources (in the sense of RESTful resources) and to define the API endpoints that interact with them.

In our example, we will define a resource for the user, called UserResource, which will contain the methods to create, activate, sign in, and deactivate users. Resources need to derive, at least, from the flama built-in Resource class, although flama provides more sophisticated classes to work with such as RESTResourceand CRUDResource.

We can save the following code in a file called resources.py under the src folder:

# src/resources.pyimport hashlibimport httpimport uuid
from flama import typesfrom flama.ddd.exceptions import NotFoundErrorfrom flama.exceptions import HTTPExceptionfrom flama.http import APIResponsefrom flama.resources import Resource, resource_method
from src import models, schemas, worker
__all__ = ["AdminResource", "UserResource"]
ENCRYPTION_SALT = uuid.uuid4().hexENCRYPTION_PEPER = uuid.uuid4().hex
class Password:    def __init__(self, password: str):        self._password = password
    def encrypt(self):        return hashlib.sha512(            (hashlib.sha512((self._password + ENCRYPTION_SALT).encode()).hexdigest() + ENCRYPTION_PEPER).encode()        ).hexdigest()
class UserResource(Resource):    name = "user"    verbose_name = "User"
    @resource_method("/", methods=["POST"], name="create")    async def create(self, worker: worker.RegisterWorker, data: types.Schema[schemas.UserDetails]):        """        tags:            - User        summary:            User create        description:            Create a user        responses:            200:                description:                    User created in successfully.        """        async with worker:            try:                await worker.user.retrieve(email=data["email"])            except NotFoundError:                await worker.user.create({**data, "password": Password(data["password"]).encrypt(), "active": False})
        return APIResponse(status_code=http.HTTPStatus.OK)
    @resource_method("/signin/", methods=["POST"], name="signin")    async def signin(self, worker: worker.RegisterWorker, data: types.Schema[schemas.UserCredentials]):        """        tags:            - User        summary:            User sign in        description:            Create a user        responses:            200:                description:                    User signed in successfully.            401:                description:                    User not active.            404:                description:                    User not found.        """        async with worker:            password = Password(data["password"])            try:                user = await worker.user.retrieve(email=data["email"])            except NotFoundError:                raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND)
            if user["password"] != password.encrypt():                raise HTTPException(status_code=http.HTTPStatus.UNAUTHORIZED)
            if not user["active"]:                raise HTTPException(                    status_code=http.HTTPStatus.BAD_REQUEST, detail=f"User must be activated via /user/activate/"                )
        return APIResponse(status_code=http.HTTPStatus.OK, schema=types.Schema[schemas.User], content=user)
    @resource_method("/activate/", methods=["POST"], name="activate")    async def activate(self, worker: worker.RegisterWorker, data: types.Schema[schemas.UserCredentials]):        """        tags:            - User        summary:            User activate        description:            Activate an existing user        responses:            200:                description:                    User activated successfully.            401:                description:                    User activation failed due to invalid credentials.            404:                description:                    User not found.        """        async with worker:            try:                user = await worker.user.retrieve(email=data["email"])            except NotFoundError:                raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND)
            if user["password"] != Password(data["password"]).encrypt():                raise HTTPException(status_code=http.HTTPStatus.UNAUTHORIZED)
            if not user["active"]:                await worker.user.update({**user, "active": True}, id=user["id"])
        return APIResponse(status_code=http.HTTPStatus.OK)
    @resource_method("/deactivate/", methods=["POST"], name="deactivate")    async def deactivate(self, worker: worker.RegisterWorker, data: types.Schema[schemas.UserCredentials]):        """        tags:            - User        summary:            User deactivate        description:            Deactivate an existing user        responses:            200:                description:                    User deactivated successfully.            401:                description:                    User deactivation failed due to invalid credentials.            404:                description:                    User not found.        """        async with worker:            try:                user = await worker.user.retrieve(email=data["email"])            except NotFoundError:                raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND)
            if user["password"] != Password(data["password"]).encrypt():                raise HTTPException(status_code=http.HTTPStatus.UNAUTHORIZED)
            if user["active"]:                await worker.user.update({**user, "active": False}, id=user["id"])
        return APIResponse(status_code=http.HTTPStatus.OK)

Base application with DDD

Now that we have implemented the data model, the repository and worker patterns, and the resource methods, we need to modify the base application we introduced before, so that everything works as expected. We need to:

• Add the SQLAlchemy connection to the application, and this is achieved by adding the SQLAlchemyModule to the application constructor as a module.
• Add the worker to the application, and this is achieved by adding the RegisterWorker to the application constructor as a component.

This will leave the app.py file as follows:

# src/app.py
from flama import Flamafrom flama.ddd import WorkerComponentfrom flama.sqlalchemy import SQLAlchemyModule
from src import resources, worker
DATABASE_URL = "sqlite+aiosqlite:///models.db"
app = Flama(    title="Domain-driven API",    version="1.0.0",    description="Domain-driven design with Flama 🔥",    docs="/docs/",    modules=[SQLAlchemyModule(DATABASE_URL)],    components=[WorkerComponent(worker=worker.RegisterWorker())],)
app.resources.add_resource("/user/", resources.UserResource)
@app.get("/", name="info")def info():    """    tags:        - Info    summary:        Ping    description:        Returns a brief description of the API    responses:        200:            description:                Successful ping.    """    return {"title": app.schema.title, "description": app.schema.description, "public": True}

It should be apparent to you already how the DDD pattern has allowed us to separate the business logic of the application (which is easily readable in the resource methods) from the data access logic (which is implemented in the repository and worker patterns). It's also worth noting how this sepration of concerns has made the code more maintainable and testable, and how the code is now more aligned with the business requirements we were given at the beginning of this example.

Running the application

Before running any command, please check that your development environment is set up correctly, and that the folder structure is as follows:

.├──README.md├──migration.py├──models.db└──src├──__init__.py├──__main__.py├──app.py├──models.py├──repositories.py├──resources.py├──schemas.py└──worker.py

If everything is set up correctly, you can run the application by executing the following command (remember to run the migration script before running the application):

> poetry run flama run --server-reload src.app:app
INFO:     Will watch for changes in these directories: [...]INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)INFO:     Started reloader process [35369] using WatchFilesINFO:     Started server process [35373]INFO:     Waiting for application startup.INFO:     Application startup complete.

Now we can try the business logic we've just implemented. Remember, you can try this either by using a tool like curl or Postman, or by using the auto-generated docs UI provided by flama by navigating to http://localhost:8000/docs/ in your browser and trying the endpoints from there.

{  "name": "John",  "surname": "Doe",  "email": "[email protected]",  "password": "123456"}

curl --request POST \  --url http://localhost:8000/user/ \  --header 'Accept: application/json' \  --header 'Content-Type: application/json' \  --data '{  "name": "John",  "surname": "Doe",  "email": "[email protected]",  "password": "123456"}'

{  "email": "[email protected]",  "password": "123456"}

curl --request POST \  --url http://localhost:8000/user/signin/ \  --header 'Accept: application/json' \  --header 'Content-Type: application/json' \  --data '{  "email": "[email protected]",  "password": "123456"}'

{  "status_code": 400,  "detail": "User must be activated via /user/activate/",  "error": "HTTPException"}

curl --request POST \  --url http://localhost:8000/user/signin/ \  --header 'Accept: application/json' \  --header 'Content-Type: application/json' \  --data '{    "email": "[email protected]",    "password": "1234567"    }'

{  "status_code": 401,  "detail": "Unauthorized",  "error": "HTTPException"}

curl --request POST \  --url http://localhost:8000/user/signin/ \  --header 'Accept: application/json' \  --header 'Content-Type: application/json' \  --data '{    "email": "[email protected]",    "password": "123456"    }'

{  "status_code": 404,  "detail": "Not Found",  "error": "HTTPException"}

curl --request POST \  --url http://localhost:8000/user/activate/ \  --header 'Accept: application/json' \  --header 'Content-Type: application/json' \  --data '{  "email": "[email protected]",  "password": "123456"}'

curl --request POST \  --url http://localhost:8000/user/activate/ \  --header 'Accept: application/json' \  --header 'Content-Type: application/json' \  --data '{    "email": "[email protected]",    "password": "1234567"}'

{  "status_code": 401,  "detail": "Unauthorized",  "error": "HTTPException"}

curl --request POST \  --url http://localhost:8000/user/activate/ \  --header 'Accept: application/json' \  --header 'Content-Type: application/json' \  --data '{    "email": "[email protected]",    "password": "123456"}'

{  "status_code": 404,  "detail": "Not Found",  "error": "HTTPException"}

curl --request POST \  --url http://localhost:8000/user/signin/ \  --header 'Accept: application/json' \  --header 'Content-Type: application/json' \  --data '{    "email": "[email protected]",     "password": "123456"    }'

{  "email": "[email protected]",  "name": "John",  "surname": "Doe",  "id": "d73d4a62-dfe9-4907-91f4-f6b06f33c534",  "active": true,  "password": "..."}

curl --request POST \  --url http://localhost:8000/user/deactivate/ \  --header 'Accept: application/json' \  --header 'Content-Type: application/json' \  --data '{    "email": "[email protected]",    "password": "123456"}'