[Python] Optimizing My FastAPI queries with joins for large kanban board ( freezes on 5 tasks...

Discussão em 'Python' iniciado por Stack, Setembro 12, 2025.

  1. Stack

    Stack Membro Participativo

    I'm building a project management app and on my backend I have a Task model with many-to-many relationships to Users (assignees) and Tags. I want to fetch tasks with their assignees and tags efficiently.

    Currently, my API endpoint fetches tasks with multiple joins and eager loads:

    @router.get("/", response_model=List[TaskOut])
    def get_tasks(
    project_id: Optional[int] = Query(None),
    archived: Optional[bool] = Query(None),
    open_only: bool = Query(False),
    created_by_me: bool = Query(False),
    assigned_to_me: bool = Query(False),
    due_before: Optional[date] = Query(None),
    due_after: Optional[date] = Query(None),
    tags: Optional[List[str]] = Query(None),
    order_by: Optional[str] = Query(
    None, description="Sort field, prefix with '-' for descending"
    ),
    skip: int = Query(0, ge=0),
    limit: int = Query(10, le=100),
    db: Session = Depends(get_db),
    current_user: Users = Depends(get_current_user),
    ):
    query = db.query(Task)

    if project_id:
    require_project_membership(db, project_id, current_user.id)
    query = query.filter(Task.project_id == project_id)
    else:
    query = (
    query.join(Project)
    .outerjoin(ProjectMember, ProjectMember.project_id == Task.project_id)
    .filter(
    (Project.owner_id == current_user.id)
    | (ProjectMember.user_id == current_user.id)
    )
    )

    if archived is True:
    query = query.filter(Task.active == False)
    elif archived is False:
    query = query.filter(Task.active == True)

    if open_only:
    query = query.filter(
    Task.active == True,
    Task.status.in_(
    [
    TaskStatusEnum.in_progress,
    TaskStatusEnum.changes_requested,
    TaskStatusEnum.approved,
    ]
    ),
    )

    if created_by_me:
    query = query.filter(Task.creator_id == current_user.id)

    if assigned_to_me:
    query = query.join(Task.assignees).filter(Users.id == current_user.id)

    # for my due dates
    if due_before:
    query = query.filter(Task.due_date <= due_before)
    if due_after:
    query = query.filter(Task.due_date >= due_after)

    # for the taags
    if tags:
    query = query.join(Task.tags).filter(Tag.name.in_(tags)).distinct()

    # my query for sorting
    if order_by:
    desc_order = order_by.startswith("-")
    field_name = order_by.lstrip("-")

    sort_map = {
    "due_date": Task.due_date,
    "priority": Task.priority,
    "created_at": Task.created_at,
    "updated_at": Task.updated_at,
    "name": Task.name,
    }

    if field_name in sort_map:
    query = query.order_by(
    desc(sort_map[field_name]) if desc_order else asc(sort_map[field_name])
    )

    tasks = query.offset(skip).limit(limit).all()

    return [TaskOut.from_orm_with_assignees(task) for task in tasks]


    Then I return them using a Pydantic schema:

    class TaskOut(TaskBase):
    id: int
    project_id: int
    stage_id: Optional[int] = None
    assignees: List[AssigneeOut] = []
    milestone_id: Optional[int] = None
    creator_id: Optional[int] = None
    created_at: datetime
    updated_at: Optional[datetime] = None

    class Config:
    from_attributes = True

    @classmethod
    def from_orm_with_assignees(cls, task):
    return cls(
    id=task.id,
    name=task.name,
    description=task.description,
    due_date=task.due_date,
    status=task.status.value if task.status else None,
    active=task.active,
    priority=task.priority,
    project_id=task.project_id,
    stage_id=task.stage_id,
    milestone_id=task.milestone_id,
    creator_id=task.creator_id,
    created_at=task.created_at,
    updated_at=task.updated_at,
    assignees=[AssigneeOut.model_validate(u) for u in task.assignees],
    )



    Problem: Even with only 5 tasks, the kanban frontend becomes very slow and sometimes freezes. I suspect the joins and eager loading are not optimized enough.

    What I've considered:

    Stripping down queries and fetching assignees and tags in separate API calls.

    Using Redis for caching.

    Adding indexes on join and filter columns.

    Questions:


    • Is it better to fetch related data (assignees, tags, messages) in separate API calls instead of doing joins?


    • How can I optimize SQLAlchemy queries with many-to-many relationships for large datasets?


    • Are there specific techniques for FastAPI + SQLAlchemy to make kanban-style task loading performant?


    • I want to keep the API simple, avoid hundreds of tiny DB calls, and make the frontend responsive.

    Any guidance on best practices for SQLAlchemy queries with joins, indexing, and eager loading would be appreciated! My frontend is in Nextjs btw.

    Continue reading...

Compartilhe esta Página