FastAPI not full utilization

Question

I have Django development in background for several years. Now I am practicing more framework this time is FastAPI.

What program does:
It is todo tasks with undo feature. I split the model into 2 parts: TaskContent and CurrentTaskContent.

I use CurrentTaskContent as a revision part. I store update, delete history in this model.

And the TaskContent is actual content in the database.

I got feedback that my code is under utilizing the framework. Here is my repository: https://github.com/elcolie/crud-taskado

Question: What are the features I missed?

Here is my sample.

import logging
import typing as typ

from datetime import date

from fastapi import status, Response

from core.common.serializers import ListTaskSchemaOutput, get_user
from core.methods.crud import TaskRepository
from core.methods.get_list_method.pagination_gadgets import generate_query_params
from core.models.models import StatusEnum, User
from core.common.validate_input import ResponsePayload, ErrorDetail, validate_due_date, validate_status, validate_username, \
    SummaryTask
logger = logging.getLogger(__name__)


def list_tasks(  # pylint: disable=too-many-locals
    response: Response,
    due_date: str | None = None,
    task_status: str | None = None,
    created_by__username: str | None = None,
    updated_by__username: str | None = None,
    _page_number: int = 1,  # Page number
    _per_page: int = 10,  # Number of items per page
) -> typ.Union[ResponsePayload, typ.Dict[str, typ.List[ErrorDetail]]]:
    """Endpoint to list all tasks."""
    errors: typ.List[ErrorDetail] = []

    due_date_instance: typ.Optional[date] = None
    status_instance: typ.Optional[StatusEnum] = None
    user_instance: typ.Optional[User] = None
    updated_user_instance: typ.Optional[User] = None
    try:
        due_date_instance = validate_due_date(due_date) if due_date else None
    except ValueError as e:
        logger.info(f"Due date validation failed. {e}")  # pylint: disable=logging-fstring-interpolation
        errors.append(ErrorDetail(loc=['due_date'], msg=str(e), type='ValueError'))
    try:
        status_instance = validate_status(task_status) if task_status else None
    except ValueError as e:
        logger.info(f"Status validation failed. {e}")  # pylint: disable=logging-fstring-interpolation
        errors.append(ErrorDetail(loc=['status'], msg=str(e), type='ValueError'))
    try:
        user_instance = (
            validate_username(created_by__username) if created_by__username else None
        )
    except ValueError as e:
        logger.info(f"Username validation failed. {e}")  # pylint: disable=logging-fstring-interpolation
        errors.append(
            ErrorDetail(loc=['created_by__username'], msg=str(e), type='ValueError')
        )
    try:
        updated_user_instance = (
            validate_username(updated_by__username) if updated_by__username else None
        )
    except ValueError as e:
        logger.info(f"Updated username validation failed. {e}")  # pylint: disable=logging-fstring-interpolation
        errors.append(
            ErrorDetail(loc=['updated_by__username'], msg=str(e), type='ValueError')
        )
    if len(errors) > 0:
        response.status_code = status.HTTP_406_NOT_ACCEPTABLE
        return {'message': errors}

    task_repository = TaskRepository()
    tasks_results = task_repository.list_tasks(
        due_date_instance, status_instance, user_instance, updated_user_instance
    )
    list_task_schema_output = ListTaskSchemaOutput()

    # Stick with class not using dictionary.

    _list_tasks = []
    for _task in tasks_results:
        created_by: User = get_user(_task[0].created_by)
        updated_by: User = get_user(_task[0].updated_by)
        _list_tasks.append(
            {
                'id': _task[1].id,
                'title': _task[1].title,
                'description': _task[1].description,
                'due_date': _task[1].due_date,
                'status': _task[1].status,
                'created_by': _task[0].created_by,
                'updated_by': _task[0].updated_by,
                # Should wrap into single query.
                'created_by__username': created_by.username
                if created_by is not None
                else None,
                'updated_by__username': updated_by.username
                if updated_by is not None
                else None,
            }
        )
    serialized_tasks = list_task_schema_output.dump(_list_tasks, many=True)
    start = (_page_number - 1) * _per_page
    end = start + _per_page
    data_length = len(serialized_tasks)

    if end >= data_length:
        if _page_number > 1:
            previous = generate_query_params(
                due_date=due_date,
                task_status=task_status,
                created_by__username=created_by__username,
                updated_by__username=updated_by__username,
                _page_number=_page_number - 1,
                _per_page=_per_page,
            )
        else:
            previous = None
        return ResponsePayload(
            count=data_length,
            tasks=[SummaryTask(**i) for i in serialized_tasks[start:end]],
            next=None,
            previous=previous,
        )

    if _page_number > 1:
        previous = generate_query_params(
            due_date=due_date,
            task_status=task_status,
            created_by__username=created_by__username,
            updated_by__username=updated_by__username,
            _page_number=_page_number - 1,
            _per_page=_per_page,
        )
    else:
        previous = None
    _next = generate_query_params(
        due_date=due_date,
        task_status=task_status,
        created_by__username=created_by__username,
        updated_by__username=updated_by__username,
        _page_number=_page_number + 1,
        _per_page=_per_page,
    )
    return ResponsePayload(
        count=data_length,
        tasks=[SummaryTask(**i) for i in serialized_tasks[start:end]],
        next=_next,
        previous=previous,
    )

Answer 1

"todo with undo"

Ok, I just kind of love that!

modern type annotations

def list_tasks(  ...
) -> typ.Union[ResponsePayload, typ.Dict[str, typ.List[ErrorDetail]]]:

Prefer the | syntax for these, e.g.
-> ResponsePayload | dict[str, list[ErrorDetail]]:

The notation you're using is certainly valid, but it is rather dated, consistent with cPython interpreters of prior years, maybe IDK 3.7 or so.

parameters are part of the Public API

def list_tasks(  ...
    ...
    _page_number: int = 1,  # Page number
    _per_page: int = 10,  # Number of items per page
    ...

I don't understand what meaning the _ underscore _local prefix could possibly indicate in this context. Callers will dance to the tune you call, they will conform to the Public API you are designing. To put foo in the public signature and then change your mind and label it _foo doesn't make sense. Make these local variables if you believe caller should never specify their value.

BTW, thank you for the very nice type annotation on errors. Prefer to spell it errors: list[ErrorDetail] = [] since the rest of your code indicates that we're targeting sufficiently modern interpreters which understand such annotations.

And rather than e.g. Optional[date], prefer date | None.

dealing with errors

    except ValueError as e:
    ...
    except ValueError as e:
    ...
    except ValueError as e:
    ...
    except ValueError as e:

IDK, maybe we need all these?

Maybe better to attempt four actions within a single try?

The big thing I'm not yet clear on is, what is the Single Responsibility of this method, what promise does it make to the caller, what post-condition does it guarantee?

We attempt a first, second, and third action which could potentially fail. In the event of failure, it's not obvious to me that we are required to soldier on and attempt the third or fourth action. Would bailing out early be acceptable? The """docstring""" doesn't offer any advice on this question, since it is missing. Please write one. In the coming months, maintenance engineers will want to know what the various responsibilities are. They could wing it, making random choices, but you might not be happy with what they choose. Offer some guidance.

local variable

    # Stick with class not using dictionary.
    _list_tasks = []

I don't understand what that means.

You're defining a local variable. It is fundamentally "private"; we don't need a _private prefix on it.

It's not like we're assigning self._list_tasks, which caller might possibly mess with. This local variable is guaranteed to go out-of-scope and be destroyed by the time caller gets to interact with its results. That's why we create small functions in the first place: so local vars are created and destroyed by the time we've computed some useful result and returned it.

double dunder

What is going on here?

        user_instance = (
            validate_username(created_by__username) if ...
        )

That created_by__username parameter was part of the signature -- I didn't even notice that craziness.

Please__don't__use__double__dunder__scores__in__variable__names. It's just weird, and doesn't have any meaning. It's not like you wrote __username__ or __str__, which describe protocols that are special for the cPython interpreter. Pick normal names, please.