Blog

Introduction

Feeling the fatigue of endlessly repeating CRUD code in your projects? 😫

Are you done with the mystery of installing packages from maintainers who’ve jumped ship? 😒

What if I told you there’s a simpler way to amp up your CRUD operations in Python without sacrificing elegance? 😎

Hold tight, because in this article, we’re venturing into the realm of Python, SQLAlchemy, and Pydantic to unveil the magic of the Factory and Repository Patterns.

Our mission: Cook up a sleek CRUD (Create, Read, Update, Delete) application, seasoned with savvy design patterns.

Say goodbye to the mundane and embrace a codebase that’s not just functional but a joy to manage on your own. Ready for the ride? 🚀

All the code, plus step-by-step tests showcasing how to wield this coding magic, are ready and waiting for you in this GitHub repository. Please feel free to copy and paste as much as your heart desires.

But Why the Design Patterns?

Imagine this: a world where business logic and database details stay in their own lanes. That’s where the Factory and Repository Patterns come into play.

Before we dive in, let’s nail down the basics.

The Factory Pattern, a design trick, crafts objects without the fuss of revealing their origins.

For us that means having a factory function take a database model class and return a new class that has all the CRUD functionality we need attached to it.

The perks include:

• Standardizing CRUD operations across models.
• Centralizing CRUD operations — do it once, and attach it to the model.
• Unravelling business logic from database intricacies.
• Smoothly expanding CRUD operations with new methods or giving them a facelift in child classes.

Now, meet it’s cousin the Repository Pattern, a design maestro simplifying the logic of retrieving and storing data. Perks include:

• Isolating query logic.
• Standardizing the database interface.
• Abstracting up database access — the rest of the application blissfully unaware of the data storage details.
• The freedom to switch data sources without causing chaos in the rest of your application.

So the repository pattern lets us have a access layer for a single model, and the factory pattern lets us create an access layer for any model.

The Application Tools

To showcase the Factory and Repository Pattern in action, we’re gearing up to build a straightforward CRUD application using some trusted Python tools:

• Python 3.12: The go-to language for its simplicity, readability, and versatility across a range of applications.
• SQLAlchemy: An open-source SQL toolkit and Object-Relational Mapping (ORM) library, providing a high-level API for smooth interactions with relational databases.
• AsyncPG: An asynchronous PostgreSQL driver for Python, tailored for PostgreSQL databases and designed with the asynchronous programming paradigm in mind.
• Pydantic: A powerhouse for data validation and settings management using Python type annotations. It lets you define data schemas and ensures data aligns with those schemas. Note: Version 2.0.0 or higher is powered by Rust for a speed boost.

Assuming you have a basic grasp of these tools, we’ll straight into the action. If you need a refresher or want to explore more about them, check out the Medium article below for an in-depth guide and tips on setting up the perfect project.

Let’s get started!

Finally, the Application

Let’s dive into crafting a straightforward CRUD application designed for handling posts and users. First, let’s explore the models that set the stage for our application:

from datetime import datetime
from typing import TypeAlias
from uuid import UUID as UuidType

from pydantic import BaseModel
from sqlalchemy import DateTime, Text
from sqlalchemy.dialects.postgresql import UUID as UuidColumn
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.sql.expression import FunctionElement
from uuid6 import uuid7

from snippets.database import Base

# Some generic types for the SQLAlchemy and Pydantic models
SnippetModel: TypeAlias = Base
SnippetSchema: TypeAlias = BaseModel


# some mixins to make our life easier
class UuidMixin:
    uuid: Mapped[UuidType] = mapped_column(
        "uuid",
        UuidColumn(as_uuid=True),
        primary_key=True,
        default=uuid7,
        nullable=False,
        sort_order=-1000,
    )


class UuidMixinSchema(BaseModel):
    uuid: UuidType = None


class utcnow(FunctionElement):
    type = DateTime()
    inherit_cache = True


@compiles(utcnow, "postgresql")
def pg_utcnow(element, compiler, **kw):
    return "TIMEZONE('utc', CURRENT_TIMESTAMP)"


class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(
        DateTime,
        nullable=False,
        server_default=utcnow(),
        sort_order=9999,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime,
        nullable=True,
        index=True,
        server_default=utcnow(),
        server_onupdate=utcnow(),
        sort_order=10000,
    )


class TimestampMixinSchema(BaseModel):
    created_at: datetime | None = None
    updated_at: datetime | None = None


# now the real models
# first posts
class Post(Base, UuidMixin, TimestampMixin):
    __tablename__ = "post"

    title: Mapped[str] = mapped_column(nullable=False, unique=True)
    content: Mapped[str] = mapped_column(Text, nullable=False)
    published: Mapped[bool] = mapped_column(nullable=False, server_default="False")
    views: Mapped[int] = mapped_column(nullable=False, server_default="0")


class PostSchema(UuidMixinSchema, TimestampMixinSchema):
    title: str
    content: str
    published: bool = False
    views: int = 0


class PostUpdateSchema(BaseModel):
    title: str | None = None
    content: str | None = None
    published: bool | None = None
    views: int | None = None


# now the users
class User(Base, UuidMixin, TimestampMixin):
    __tablename__ = "user"

    username: Mapped[str] = mapped_column(nullable=False, unique=True, index=True)
    email: Mapped[str] = mapped_column(unique=True, index=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(nullable=False)
    is_active: Mapped[bool] = mapped_column(default=True, server_default="TRUE")


class UserBaseSchema(BaseModel):
    username: str
    email: str
    is_active: bool = True


class UserSchema(UserBaseSchema, UuidMixinSchema, TimestampMixinSchema):
    hashed_password: str


class UserInSchema(UserBaseSchema):
    password: str


class UserUpdateSchema(BaseModel):
    username: str | None = None
    email: str | None = None
    password: str | None = None
    is_active: bool | None = None

In this snippet, you’ll find the Post and User database models.

We’ve added some nifty mixins for features like a UUID primary key, along with created_atand updated_at fields.

Not stopping there, we’ve also set up Pydantic schemas for these models. These schemas pull double duty: validating data before it hits the database and shaping data for a smooth journey back to the client.

When it comes to users, we’ve got distinct schemas for what’s headed into the database versus what’s making a triumphant exit.

Why the separation?

Because we’re adding a touch of magic — hashing the password before storing it. Okay, for demo purposes, we’re keeping it simple and saying the hashed password is just the reverse of the original (a big no-no in real applications, of course).

Now let’s check out the factory and repository patterns in action😯

The Factory and Repository Pattern for CRUD Operations

Rather than crafting a repository for each model, let’s spice things up by employing the Factory Pattern. We’ll whip up a generic repository capable of handling any model.

This abstract repository will feature methods for the usual suspects: creating, reading, updating, and deleting data from the database.

Now, here’s the code for this abstract repository (copy and paste it and use it at your leisure 😃)

from uuid import UUID

from sqlalchemy import delete, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from snippets.models import (
    SnippetModel,
    SnippetSchema,
    User,
    UserInSchema,
    UserSchema,
    UserUpdateSchema,
)


class SnippetException(Exception):
    pass


class IntegrityConflictException(Exception):
    pass


class NotFoundException(Exception):
    pass


def CrudFactory(model: SnippetModel):
    class AsyncCrud:
        @classmethod
        async def create(
            cls,
            session: AsyncSession,
            data: SnippetSchema,
        ) -> SnippetModel:
            """Accepts a Pydantic model, creates a new record in the database, catches
            any integrity errors, and returns the record.

            Args:
                session (AsyncSession): SQLAlchemy async session
                data (SnippetSchema): Pydantic model

            Raises:
                IntegrityConflictException: if creation conflicts with existing data
                SnippetException: if an unknown error occurs

            Returns:
                SnippetModel: created SQLAlchemy model
            """
            try:
                db_model = model(**data.model_dump())
                session.add(db_model)
                await session.commit()
                await session.refresh(db_model)
                return db_model
            except IntegrityError:
                raise IntegrityConflictException(
                    f"{model.__tablename__} conflicts with existing data.",
                )
            except Exception as e:
                raise SnippetException(f"Unknown error occurred: {e}") from e

        @classmethod
        async def create_many(
            cls,
            session: AsyncSession,
            data: list[SnippetSchema],
            return_models: bool = False,
        ) -> list[SnippetModel] | bool:
            """_summary_

            Args:
                session (AsyncSession): SQLAlchemy async session
                data (list[SnippetSchema]): list of Pydantic models
                return_models (bool, optional): Should the created models be returned
                    or a boolean indicating they have been created. Defaults to False.

            Raises:
                IntegrityConflictException: if creation conflicts with existing data
                SnippetException: if an unknown error occurs

            Returns:
                list[SnippetModel] | bool: list of created SQLAlchemy models or boolean
            """
            db_models = [model(**d.model_dump()) for d in data]
            try:
                session.add_all(db_models)
                await session.commit()
            except IntegrityError:
                raise IntegrityConflictException(
                    f"{model.__tablename__} conflict with existing data.",
                )
            except Exception as e:
                raise SnippetException(f"Unknown error occurred: {e}") from e

            if not return_models:
                return True

            for m in db_models:
                await session.refresh(m)

            return db_models

        @classmethod
        async def get_one_by_id(
            cls,
            session: AsyncSession,
            id_: str | UUID,
            column: str = "uuid",
            with_for_update: bool = False,
        ) -> SnippetModel:
            """Fetches one record from the database based on a column value and returns
            it, or returns None if it does not exist. Raises an exception if the column
            doesn't exist.

            Args:
                session (AsyncSession): SQLAlchemy async session
                id_ (str | UUID): value to search for in `column`.
                column (str, optional): the column name in which to search.
                    Defaults to "uuid".
                with_for_update (bool, optional): Should the returned row be locked
                    during the lifetime of the current open transactions.
                    Defaults to False.

            Raises:
                SnippetException: if the column does not exist on the model

            Returns:
                SnippetModel: SQLAlchemy model or None
            """
            try:
                q = select(model).where(getattr(model, column) == id_)
            except AttributeError:
                raise SnippetException(
                    f"Column {column} not found on {model.__tablename__}.",
                )

            if with_for_update:
                q = q.with_for_update()

            results = await session.execute(q)
            return results.unique().scalar_one_or_none()

        @classmethod
        async def get_many_by_ids(
            cls,
            session: AsyncSession,
            ids: list[str | UUID] = None,
            column: str = "uuid",
            with_for_update: bool = False,
        ) -> list[SnippetModel]:
            """Fetches multiple records from the database based on a column value and
            returns them. Raises an exception if the column doesn't exist.

            Args:
                session (AsyncSession): SQLAlchemy async session
                ids (list[str  |  UUID], optional): list of values to search for in
                    `column`. Defaults to None.
                column (str, optional): the column name in which to search
                    Defaults to "uuid".
                with_for_update (bool, optional): Should the returned rows be locked
                    during the lifetime of the current open transactions.
                    Defaults to False.

            Raises:
                SnippetException: if the column does not exist on the model

            Returns:
                list[SnippetModel]: list of SQLAlchemy models
            """
            q = select(model)
            if ids:
                try:
                    q = q.where(getattr(model, column).in_(ids))
                except AttributeError:
                    raise SnippetException(
                        f"Column {column} not found on {model.__tablename__}.",
                    )

            if with_for_update:
                q = q.with_for_update()

            rows = await session.execute(q)
            return rows.unique().scalars().all()

        @classmethod
        async def update_by_id(
            cls,
            session: AsyncSession,
            data: SnippetSchema,
            id_: str | UUID,
            column: str = "uuid",
        ) -> SnippetModel:
            """Updates a record in the database based on a column value and returns the
            updated record. Raises an exception if the record isn't found or if the
            column doesn't exist.

            Args:
                session (AsyncSession): SQLAlchemy async session
                data (SnippetSchema): Pydantic schema for the updated data.
                id_ (str | UUID): value to search for in `column`
                column (str, optional): the column name in which to search
                    Defaults to "uuid".
            Raises:
                NotFoundException: if the record isn't found
                IntegrityConflictException: if the update conflicts with existing data

            Returns:
                SnippetModel: updated SQLAlchemy model
            """
            db_model = await cls.get_one_by_id(
                session, id_, column=column, with_for_update=True
            )
            if not db_model:
                raise NotFoundException(
                    f"{model.__tablename__} {column}={id_} not found.",
                )

            values = data.model_dump(exclude_unset=True)
            for k, v in values.items():
                setattr(db_model, k, v)

            try:
                await session.commit()
                await session.refresh(db_model)
                return db_model
            except IntegrityError:
                raise IntegrityConflictException(
                    f"{model.__tablename__} {column}={id_} conflict with existing data.",
                )

        @classmethod
        async def update_many_by_ids(
            cls,
            session: AsyncSession,
            updates: dict[str | UUID, SnippetSchema],
            column: str = "uuid",
            return_models: bool = False,
        ) -> list[SnippetModel] | bool:
            """Updates multiple records in the database based on a column value and
            returns the updated records. Raises an exception if the column doesn't
            exist.

            Args:
                session (AsyncSession): SQLAlchemy async session
                updates (dict[str  |  UUID, SnippetSchema]): dictionary of id_ to
                    Pydantic update schema
                column (str, optional): the column name in which to search.
                    Defaults to "uuid".
                return_models (bool, optional): Should the created models be returned
                    or a boolean indicating they have been created. Defaults to False.
                    Defaults to False.

            Raises:
                IntegrityConflictException: if the update conflicts with existing data

            Returns:
                list[SnippetModel] | bool: list of updated SQLAlchemy models or boolean
            """
            updates = {str(id): update for id, update in updates.items() if update}
            ids = list(updates.keys())
            db_models = await cls.get_many_by_ids(
                session, ids=ids, column=column, with_for_update=True
            )

            for db_model in db_models:
                values = updates[str(getattr(db_model, column))].model_dump(
                    exclude_unset=True
                )
                for k, v in values.items():
                    setattr(db_model, k, v)
                session.add(db_model)

            try:
                await session.commit()
            except IntegrityError:
                raise IntegrityConflictException(
                    f"{model.__tablename__} conflict with existing data.",
                )

            if not return_models:
                return True

            for db_model in db_models:
                await session.refresh(db_model)

            return db_models

        @classmethod
        async def remove_by_id(
            cls,
            session: AsyncSession,
            id_: str | UUID,
            column: str = "uuid",
        ) -> int:
            """Removes a record from the database based on a column value. Raises an
            exception if the column doesn't exist.

            Args:
                session (AsyncSession): SQLAlchemy async session
                id (str | UUID): value to search for in `column` and delete
                column (str, optional): the column name in which to search.
                    Defaults to "uuid".

            Raises:
                SnippetException: if the column does not exist on the model

            Returns:
                int: number of rows removed, 1 if successful, 0 if not. Can be greater
                    than 1 if id_ is not unique in the column.
            """
            try:
                query = delete(model).where(getattr(model, column) == id_)
            except AttributeError:
                raise SnippetException(
                    f"Column {column} not found on {model.__tablename__}.",
                )

            rows = await session.execute(query)
            await session.commit()
            return rows.rowcount

        @classmethod
        async def remove_many_by_ids(
            cls,
            session: AsyncSession,
            ids: list[str | UUID],
            column: str = "uuid",
        ) -> int:
            """Removes multiple records from the database based on a column value.
            Raises an exception if the column doesn't exist.

            Args:
                session (AsyncSession): SQLAlchemy async session
                ids (list[str  |  UUID]): list of values to search for in `column` and
                column (str, optional): the column name in which to search.
                    Defaults to "uuid".

            Raises:
                SnippetException: if ids is empty to stop deleting an entire table
                SnippetException: if column does not exist on the model

            Returns:
                int: _description_
            """
            if not ids:
                raise SnippetException("No ids provided.")

            try:
                query = delete(model).where(getattr(model, column).in_(ids))
            except AttributeError:
                raise SnippetException(
                    f"Column {column} not found on {model.__tablename__}.",
                )

            rows = await session.execute(query)
            await session.commit()
            return rows.rowcount

    AsyncCrud.model = model
    return AsyncCrud

Check it out: the function takes a model class, creates an abstract class featuring standard CRUD operations, each bound to the model class and then returns the class.

So to use the factory all we have to do is pass in a SQLAlchemy model and we have a class that provides 99% of the crud operations we’ll ever need!

Witness the Factory Pattern in action:

from snippets.crud import CrudFactory
from snippets.database import create_async_engine, create_pg_url_from_env
from snippets.models import Post

engine = create_async_engine(create_pg_url_from_env())

SessionLocal = async_sessionmaker(
    bind=engine,
    autoflush=False,
    expire_on_commit=False,
    autocommit=False,
)

PostCrud = Crudfactory(Post)

async with SessionLocal() as session:
    post = await PostCrud.get_one_by_id(
        session=session,
        column="title",
        id_="The Factory and Repository Pattern with SQLAlchemy and Pydantic",
    )
    post.published = True
    await session.commit()
  

Notice the flexibility: we don’t tie ourselves to performing CRUD operations solely on the primary key. Any column in the table can be the star of the show.

This is handy — think usernames or emails that are unique across the table, so why do we want to create separate methods for them …

Let’s bring it to life with some slick unit tests:

import pytest
from snippets.crud import CrudFactory, IntegrityConflictException
from snippets.models import Post, PostSchema, PostUpdateSchema

PostCrud = CrudFactory(Post)


@pytest.mark.asyncio
class TestPostCrud:
    async def test_create_post(self, session):
        post_create = PostSchema(title="my first post", content="hello world")
        post = await PostCrud.create(session, post_create)
        assert post.uuid is not None
        assert post.created_at is not None
        assert post.updated_at is not None
        assert post.title == "my first post"
        assert post.content == "hello world"

    async def test_get_many_by_ids(self, session):
        posts_create = [
            PostSchema(title="1: my first post", content="hello world"),
            PostSchema(title="2: my second post", content="hello world again"),
            PostSchema(title="3: my third post", content="foo bar baz"),
        ]
        posts = await PostCrud.create_many(session, posts_create, return_models=True)
        posts = sorted(posts, key=lambda x: x.title)

        posts_selected = await PostCrud.get_many_by_ids(
            session, [posts[0].uuid, posts[2].uuid]
        )
        posts_selected = sorted(posts_selected, key=lambda x: x.title)

        assert posts_selected[0].title == "1: my first post"
        assert posts_selected[1].title == "3: my third post"

    async def test_update_by_id(self, session):
        posts_create = [
            PostSchema(title="1: my first post", content="hello world"),
            PostSchema(title="2: my second post", content="hello world again"),
        ]
        posts = await PostCrud.create_many(session, posts_create, return_models=True)
        posts = sorted(posts, key=lambda x: x.title)

        post_update = PostUpdateSchema(content="i changed my mind")
        post1 = await PostCrud.update_by_id(session, posts[0].uuid, post_update)
        assert post1.content == "i changed my mind"

    async def test_remove_by_id(self, session):
        posts_create = [
            PostSchema(title="1: my first post", content="hello world"),
            PostSchema(title="2: my second post", content="hello world again"),
        ]
        posts = await PostCrud.create_many(session, posts_create, return_models=True)
        posts = sorted(posts, key=lambda x: x.title)

        row_count = await PostCrud.remove_by_id(session, posts[0].uuid)
        assert row_count == 1

        all_posts = await PostCrud.get_many_by_ids(
            session, [posts[0].uuid, posts[1].uuid]
        )
        assert len(all_posts) == 1

        row_count = await PostCrud.remove_by_id(session, posts[0].uuid)
        assert row_count == 0

You can now see that this factory/repository combo allows us to greater simplify and abstract out the data access layer of the application, giving a simple and repeatable interface for all models.

Just take the factory function, pass in a SQLAlchemy model, and bam!

Expanding a Repository

The beauty of the Factory Pattern shines as we extend our repository effortlessly.

Simply create a new class that inherits from the abstract repository and tweak the methods as needed.

This flexibility allows us to add new methods or modify existing ones with ease.

Consider this: our User repository requires the ability to take a password in plain text, hash it, and then gracefully store it in the database.

Achieving this is a breeze with a new class that inherits from the abstract repository, wherein we override the create and update_by_id methods.

class UserCrud(CrudFactory(User)):

    @classmethod
    async def create_many(cls, *args, **kwargs) -> list[User]:
        raise NotImplementedError("Create many not implemented for users.")

    @classmethod
    async def update_many_by_ids(cls, *args, **kwargs) -> list[User] | bool:
        raise NotImplementedError("Update many not implemented for users.")

    @classmethod
    def format_user_with_password(cls, user: UserInSchema) -> UserSchema:
        """Take a Pydantic UserInSchema and return a UserSchema with the password
        hashed.

        Args:
            user (UserInSchema): Pydantic UserInSchema holding the user information

        Returns:
            UserSchema: Pydantic UserSchema with the password hashed
        """
        user_data = user.model_dump()
        password = user_data.pop("password")
        db_user = UserSchema(
            **user_data, hashed_password=cls.get_password_hash(password)
        )
        return db_user

    @classmethod
    def get_password_hash(cls, password: str) -> str:
        """Perform hashing of passwords. This is a simple example and should not be used
        in production. A simple example:

            ```python
            from passlib.context import CryptContext
            pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

            def verify_password(plain_password: str, hashed_password: str) -> bool:
                return pwd_context.verify(plain_password, hashed_password)

            def get_password_hash(password: str) -> str:
                return pwd_context.hash(password)
            ```

        Args:
            password (str): user's password

        Returns:
            str: user's hashed password
        """
        # NOTE: do not ever do this in production
        return password[::-1]

    @classmethod
    async def create(cls, session: AsyncSession, data: UserInSchema) -> User:
        """Create a user in the database. This method is overridden to hash the password
        and then calls the parent create method, with the hashed password.

        Args:
            session (AsyncSession): SQLAlchemy async session
            data (UserInSchema): Pydantic UserInSchema holding the user information

        Returns:
            User: SQLAlchemy User model
        """
        db_user = cls.format_user_with_password(data)
        return await super(cls, cls).create(session, data=db_user)

    @classmethod
    async def update_by_id(
        cls,
        session: AsyncSession,
        data: UserUpdateSchema,
        id_: str | UUID,
        column: str = "uuid",
    ) -> User:
        """Updates a user in the database based on a column value and returns the
        updated user. Raises an exception if the user isn't found or if the column
        doesn't exist.

        Overrides the parent method to hash the password if it is included in the
        update.

        Args:
            session (AsyncSession): SQLAlchemy async session
            data (UserUpdateSchema): Pydantic schema for the updated data.
            id_ (str | UUID): value to search for in `column`
            column (str, optional): the column name in which to search.
                Defaults to "uuid".

        Raises:
            NotFoundException: user not found in database given id_ and column
            IntegrityConflictException: update conflicts with existing data

        Returns:
            User: updated SQLAlchemy model
        """
        db_model = await cls.get_one_by_id(
            session, id_, column=column, with_for_update=True
        )
        if not db_model:
            raise NotFoundException(
                f"{User.__tablename__} id={id_} not found.",
            )

        values = data.model_dump(exclude_unset=True, exclude={"password"})
        for k, v in values.items():
            setattr(db_model, k, v)

        if data.password is not None:
            db_model.hashed_password = cls.get_password_hash(data.password)

        try:
            await session.commit()
            await session.refresh(db_model)
            return db_model
        except IntegrityError as e:
            raise IntegrityConflictException(
                f"{User.__tablename__} {column}={id_} conflict with existing data.",
            ) from e

In the create method override, we hash the password before calling the original create method to securely store the user in the database.

For the update_by_id method we rewrite the original method to include the logic to handle hashing the password if it exists in the update, maintaining the security of our stored data. The Factory and Repository Pattern’s power in action again! 👊

Again some simple unit tests to show the User repository in action:

import pytest
from snippets.crud import IntegrityConflictException, UserCrud
from snippets.models import UserInSchema, UserUpdateSchema


@pytest.mark.asyncio
class TestUserCrud:
    async def test_create_user(self, session):
        user_create = UserInSchema(
            username="test",
            email="test@test.com",
            password="password",
        )
        user = await UserCrud.create(session, user_create)
        assert user.uuid is not None
        assert user.created_at is not None
        assert user.updated_at is not None
        assert user.username == "test"
        assert user.email == "test@test.com"
        assert user.is_active is True
        assert user.hashed_password == "drowssap"

    async def test_create_user_conflict_username(self, session):
        user_create = UserInSchema(
            username="test",
            email="test@test.com",
            password="password",
        )
        _ = await UserCrud.create(session, user_create)

        user_conflict = UserInSchema(
            username="test",
            email="test1@test.com",
            password="password",
        )
        with pytest.raises(IntegrityConflictException):
            _ = await UserCrud.create(session, user_conflict)

    async def test_update_user(self, session):
        user_create = UserInSchema(
            username="test",
            email="test@test.com",
            password="password",
        )
        user = await UserCrud.create(session, user_create)

        user_update = await UserCrud.update_by_id(
            session, id_=user.uuid, data=UserUpdateSchema(password="new_password")
        )
        assert user_update.username == "test"
        assert user_update.email == "test@test.com"
        assert user_update.hashed_password == "drowssap_wen"

    async def test_update_user_conflict(self, session):
        _ = await UserCrud.create(
            session,
            UserInSchema(
                username="test",
                email="test@test.com",
                password="password",
            ),
        )
        _ = await UserCrud.create(
            session,
            UserInSchema(
                username="test1",
                email="test1@test.com",
                password="password",
            ),
        )

        with pytest.raises(IntegrityConflictException):
            _ = await UserCrud.update_by_id(
                session,
                id_="test1",
                column="username",
                data=UserUpdateSchema(email="test@test.com"),
            )

Conclusion

In our journey building a Python CRUD app with SQLAlchemy and Pydantic, the Factory and Repository Patterns stood out for their elegance and adaptability.

The Factory Pattern crafted a generic repository, easily extendable for any model. Meanwhile, the Repository Pattern abstracted data logic, providing flexibility in data source choices.

The Factory Pattern also showcased its flexibility as we effortlessly extended the repository, like adding password hashing to the User repository.

In essence, these patterns turned data management into an art, making our code not just functional but also effortlessly maintainable and extensible. Whether it’s a simple CRUD app or a larger project, consider these design patterns for a harmonious codebase. 🚀

If you enjoyed this article or have any feedback please feel free to clap/follow and check out the original source code GitHub.

Thanks for reading!