The next step in our development is to connect users and todos

To achieve this, we'll need to modify our models to create a relationship between users and todos. We'll also need to update our current services to handle this new relationship, and add a new service and controller to allow users to create, update, and delete their todos

This article is a follow-up to the previous one. If you haven't read it yet, I highly recommend doing so for better context and understanding

You can find the previous article here

The first thing we need to do to connect users and todos is update our models. Currently, the User model is not connected with the Todo model. To establish this relationship, we need to make some modifications to our existing models

One of the modifications we need to make is to remove the unique constraint from the value column of the Todo model

Next, we'll create a new unique constraint called a composite unique constraint on the value and user_id columns of the Todo model. This will ensure that each user can only have one todo with a specific value

Todo.py

import datetime

from sqlmodel import SQLModel, Field, Relationship, UniqueConstraint
from typing import Optional

import sqlalchemy as sa

from app.models.User import User


class Todo(SQLModel, table=True):
    __tablename__ = "todos"

    id: Optional[int] = Field(default=None, primary_key=True)
    value: str = Field(sa_column=sa.Column(sa.TEXT, nullable=False, unique=True))
    user_id: int = Field(sa_column=sa.Column(sa.Integer, sa.ForeignKey(User.user_id, ondelete="CASCADE"), nullable=False))
    created_at: datetime.datetime = Field(sa_column=sa.Column(sa.DateTime(timezone=True), default=datetime.datetime.now))

    user: User = Relationship(back_populates="todos", sa_relationship_kwargs={'lazy': 'selectin'})

    __table_args__ = (
        UniqueConstraint("value", "user_id", name="unique_todos_user_id_value"),
    )

User.py

import datetime

from sqlmodel import SQLModel, Field, Relationship

import sqlalchemy as sa


class User(SQLModel, table=True):
    __tablename__ = "users"

    user_id: int = Field(default=None, primary_key=True)
    username: str = Field(sa_column=sa.Column(sa.TEXT, nullable=False, unique=True))
    first_name: str = Field(sa_column=sa.Column(sa.TEXT, nullable=False))
    last_name: str = Field(sa_column=sa.Column(sa.TEXT, nullable=False))
    email: str = Field(sa_column=sa.Column(sa.TEXT, nullable=False, unique=True))
    birthdate: datetime.date = Field(nullable=False)
    created_at: str = Field(sa_column=sa.Column(sa.DateTime(timezone=True), default=datetime.datetime.now))

    passwords: list["UserPassword"] = Relationship(back_populates="user", sa_relationship_kwargs={'lazy': 'joined'})
    todos: list["Todo"] = Relationship(back_populates="user", sa_relationship_kwargs={'lazy': 'selectin'})

Now that we have updated our models to connect users and todos, the next step is to remove the create_todo method from UserService.py. This is because it doesn't make sense to create a todo without associating it with a user

 async def create_todo(self, data: CreateTodoRequest):
        new_todo = Todo(value=data.value)

        self.session.add(new_todo)
        await self.session.commit()

        return new_todo

…and related endpoint in TodoController.py

@todo_router.post("/", response_model=Todo)
async def create_todo(*, data: CreateTodoRequest = Body(), todo_service: TodoService = Depends(get_todo_service)):
    new_todo = await todo_service.create_todo(data)

    return new_todo

To connect users and todos and allow for the creation and management of user todos, we will create a new service called UserTodoService.py

This service will handle the CRUD operations for user todos

UserTodoService.py

import asyncpg
from sqlalchemy import exc
from sqlalchemy.cimmutabledict import immutabledict
from sqlmodel import select, and_, delete, update

from requests import Session

from app.exceptions.TodoDuplicationException import TodoDuplicationException
from app.models.Todo import Todo
from app.models.User import User
from app.requests.CreateTodoRequest import CreateTodoRequest
from app.requests.UpdateTodoRequest import UpdateTodoRequest


class UserTodoService:
    def __init__(self, session: Session):
        self.session = session

    async def create_todo(self, data: CreateTodoRequest, user_id: str):
        try:
            new_todo = Todo(value=data.value, user_id=user_id)

            self.session.add(new_todo)
            await self.session.commit()

            return new_todo
        except (exc.IntegrityError, asyncpg.exceptions.UniqueViolationError):
            raise TodoDuplicationException(f"Todo [{data.value}] already exists")
        except Exception as e:
            raise Exception(e)

    async def get_todos(self, user_id: int, offset: int, limit: int):
        query = (
            select(Todo)
            .where(Todo.user_id == user_id)
            .offset(offset)
            .limit(limit)
        )

        result = await self.session.execute(query)
        return result.scalars().all()

    async def get_todo(self, user_id: int, todo_id: int):
        query = (
            select(Todo)
            .where(and_(Todo.user_id == user_id, Todo.id == todo_id))
        )

        result = await self.session.execute(query)
        return result.scalars().one()

    async def delete_todo(self, user_id: int, todo_id: int) -> None:
        query = (
            delete(Todo)
            .where(and_(Todo.id == todo_id, User.user_id == user_id))
        )

        await self.session.execute(query, execution_options=immutabledict({"synchronize_session": 'fetch'}))
        await self.session.commit()

    async def update_todo(self, user_id: int, new_todo: UpdateTodoRequest) -> None:
        query = (
            update(Todo)
            .values(value=new_todo.value)
            .where(and_(Todo.id == new_todo.id, User.user_id == user_id))
        )

        await self.session.execute(query, execution_options=immutabledict({"synchronize_session": 'fetch'}))
        await self.session.commit()

Notice there is a new class — TodoDuplicationException

This exception will be used to handle cases where a user attempts to create a todo with a value that already exists for their account

TodoDuplicationException.py

class TodoDuplicationException(Exception):
    pass

Don't forget to register our service in deps.py to make it available to our controllers

from app.services.UserTodoService import UserTodoService
from database import async_session
from app.services.TodoService import TodoService
from app.services.UserService import UserService


# ... 


async def get_user_todo_service():
    async with async_session() as session:
        async with session.begin():
            yield UserTodoService(session)

UserTodoController.py

from typing import Annotated

from fastapi import APIRouter, status, Body, Depends, HTTPException, Response, Query
from fastapi.responses import JSONResponse

from fastapi.encoders import jsonable_encoder
from app.auth.user import get_current_user
from app.deps import get_user_todo_service
from app.exceptions.TodoDuplicationException import TodoDuplicationException
from app.responses.GetByUsernameResponse import GetByUsernameResponse
from app.services.UserTodoService import UserTodoService
from app.requests.CreateTodoRequest import CreateTodoRequest
from app.requests.UpdateTodoRequest import UpdateTodoRequest

user_todo_router = APIRouter(
    prefix="/users/todos",
    tags=["User todos"]
)


@user_todo_router.post("/", status_code=status.HTTP_201_CREATED)
async def create_todo(*, todo: CreateTodoRequest = Body(),
                      todo_service: UserTodoService = Depends(get_user_todo_service),
                      current_user: GetByUsernameResponse = Depends(get_current_user),
                      ):
    try:
        new_todo = await todo_service.create_todo(todo, current_user.user_id)
        created_at = new_todo.created_at
        todo_id = new_todo.id
        value = new_todo.value

        return JSONResponse(status_code=status.HTTP_201_CREATED,
                            content=jsonable_encoder(
                                {
                                    "todo_id": todo_id,
                                    "value": value,
                                    "created_at": created_at,
                                }
                            ))
    except TodoDuplicationException as e:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail={
            "message": str(e),
            "code": "EMAIL_DUPLICATION",
            "status_code": status.HTTP_409_CONFLICT,
        })
    except Exception as e:
        print(e)
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={
            "message": "Something went wrong",
            "code": "INTERNAL_SERVER_ERROR",
            "status_code": status.HTTP_500_INTERNAL_SERVER_ERROR,
        })


@user_todo_router.get("/")
async def get_todos(*,
                    todo_service: UserTodoService = Depends(get_user_todo_service),
                    offset: Annotated[int | None, Query()] = 0,
                    limit: Annotated[int | None, Query()] = 10,
                    current_user: GetByUsernameResponse = Depends(get_current_user),
                    ):
    todo = await todo_service.get_todos(current_user.user_id, offset, limit)

    return todo


@user_todo_router.get("/{todo_id}")
async def get_todo(*,
                   todo_id: int,
                   todo_service: UserTodoService = Depends(get_user_todo_service),
                   current_user: GetByUsernameResponse = Depends(get_current_user),
                   ):
    todo = await todo_service.get_todo(current_user.user_id, todo_id)

    return todo


@user_todo_router.patch("/", status_code=status.HTTP_204_NO_CONTENT)
async def update_todo(*,
                      todo: UpdateTodoRequest = Body(),
                      todo_service: UserTodoService = Depends(get_user_todo_service),
                      current_user: GetByUsernameResponse = Depends(get_current_user),
                      ):
    await todo_service.update_todo(current_user.user_id, todo)
    return Response(status_code=status.HTTP_204_NO_CONTENT)


@user_todo_router.delete("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(*,
                      todo_id: int,
                      todo_service: UserTodoService = Depends(get_user_todo_service),
                      current_user: GetByUsernameResponse = Depends(get_current_user),
                      ):
    await todo_service.delete_todo(current_user.user_id, todo_id)
    return Response(status_code=status.HTTP_204_NO_CONTENT)

… and register in main.py

from fastapi import FastAPI, APIRouter

from app.controllers.AuthController import auth_router
from app.controllers.TodoController import todo_router
from app.controllers.UserTodoController import user_todo_router
from database import init_db

# ...

app.include_router(user_todo_router, prefix="/api/v1")

Create a todo

Don't forget to add Bearer Token

Get all todos

I created more than one todo to see if it works

Update a todo

Get a todo

Delete a todo

Thank you for reading this series! I hope you enjoyed it. I'll share three bonus articles to help you improve our REST API:

1. Add Swagger documentation to our API to make it easier to understand and use.
2. Deploy our REST API
3. Handle exceptions properly to ensure that your API returns the appropriate response instead of an internal server error

So, stay tuned!

If you enjoyed the information and found it helpful, you can show your support by buying me a coffee! Your contribution helps me continue creating valuable content and resources