> ## Documentation Index > Fetch the complete documentation index at: https://notes.kodekloud.com/llms.txt > Use this file to discover all available pages before exploring further. # Joins Sqlalchemy > Explains building SQLAlchemy joins in FastAPI to fetch posts with vote counts, handle aggregation, pagination, and resolve Pydantic response model mismatches. This guide demonstrates how to combine data from two tables using SQLAlchemy joins within a FastAPI posts router. We'll build the query incrementally: start from a basic posts query, add a join to the votes table, aggregate (count) votes per post, and then apply filters, limit, and offset. Finally, we'll cover response-model mismatches and two strategies to resolve them. Why this matters * Efficiently fetch posts with vote counts in a single query. * Avoid N+1 query problems by leveraging SQL joins and aggregation. * Ensure FastAPI/Pydantic response models match the returned data shape. Table of contents * Basic posts query * Adding a JOIN * Counting votes and grouping * Complete query with filters, limit, and offset * Response model mismatches and solutions * Final router implementation * Alternative: flattening results * Summary and references ## Basic posts query (fetch posts only) This is the starting `get_posts` implementation that retrieves posts only: ```python theme={null} from fastapi import Depends, APIRouter, status from sqlalchemy.orm import Session from typing import List, Optional from .. import models, schemas, oauth2 from ..database import get_db router = APIRouter( prefix="/posts", tags=["Posts"], ) @router.get("/", response_model=List[schemas.Post]) def get_posts( db: Session = Depends(get_db), current_user: int = Depends(oauth2.get_current_user), limit: int = 10, skip: int = 0, search: Optional[str] = "" ): posts = ( db.query(models.Post) .filter(models.Post.title.contains(search)) .limit(limit) .offset(skip) .all() ) return posts ``` Note: the core of the query is `db.query(models.Post)` — this returns mapped Post model instances (one per row). If you remove `.all()` you get a SQLAlchemy Query object; inspect it with `str(query)` or `query.statement` to view the generated SQL for debugging. ## Adding a JOIN To include vote information, add a `join(...)` to the query. This demonstrates joining the `votes` table on the foreign key: ```python theme={null} results = db.query(models.Post).join( models.Vote, models.Vote.post_id == models.Post.id ) ``` By default SQLAlchemy produces an inner join. Use `isouter=True` to create a LEFT OUTER JOIN so posts with zero votes are included. If you want to include posts without any votes (i.e., zero votes), use a left outer join: ```python theme={null} results = db.query(models.Post).join( models.Vote, models.Vote.post_id == models.Post.id, isouter=True ) ``` ## Counting votes and grouping To compute vote counts per post, import SQLAlchemy's `func` and apply `func.count(...)` along with `.group_by(...)`. Use `.label(...)` to name the aggregated column: ```python theme={null} from sqlalchemy import func results = ( db.query(models.Post, func.count(models.Vote.post_id).label("votes")) .join(models.Vote, models.Vote.post_id == models.Post.id, isouter=True) .group_by(models.Post.id) ) ``` The generated SQL is equivalent to: SELECT posts.\*, count(votes.post\_id) AS votes\ FROM posts LEFT OUTER JOIN votes ON votes.post\_id = posts.id\ GROUP BY posts.id; Naming the label (we used "votes") makes it easy to read the results and include the count in the API response. ## Complete query with filters, limit, and offset Re-apply filtering, pagination, and execute the query: ```python theme={null} results = ( db.query(models.Post, func.count(models.Vote.post_id).label("votes")) .join(models.Vote, models.Vote.post_id == models.Post.id, isouter=True) .group_by(models.Post.id) .filter(models.Post.title.contains(search)) .limit(limit) .offset(skip) .all() ) ``` This returns a list where each row contains two elements (commonly as a tuple): `(PostInstance, votes_count)`. ## Pydantic response model mismatch and solution When your endpoint's `response_model` is `List[schemas.Post]` but you return `(PostInstance, votes_count)` rows, FastAPI/Pydantic will raise validation errors because the returned structure doesn't match the expected schema shape. Two approaches to fix this: 1. Create a response schema that matches the returned tuple/nested shape. 2. Flatten/transform each row into a dict matching an existing schema. Below is an example response schema approach. Example schemas to match the returned structure ```python theme={null} from pydantic import BaseModel, EmailStr from datetime import datetime from typing import Optional class PostBase(BaseModel): title: str content: str published: bool = True class UserOut(BaseModel): id: int email: EmailStr created_at: datetime class Config: orm_mode = True class Post(PostBase): id: int created_at: datetime owner_id: int owner: UserOut class Config: orm_mode = True class PostOut(BaseModel): Post: Post # capital "Post" matches how the query result may be serialized votes: int class Config: orm_mode = True ``` Important: If your query serialization nests the Post under a capitalized `Post` key, your Pydantic model must match that key exactly. Otherwise set a different returned shape or adjust your schema accordingly. If your response model does not match the exact shape (keys, nesting, capitalization) of the returned data, FastAPI/Pydantic will raise validation errors. Update the returned data shape or the response model to match. ## Final router implementation (response model matches joined results) Set the endpoint's `response_model` to `List[schemas.PostOut]` and return the query results directly: ```python theme={null} from fastapi import Depends, APIRouter, status from sqlalchemy.orm import Session from sqlalchemy import func from typing import List, Optional from .. import models, schemas, oauth2 from ..database import get_db router = APIRouter(prefix="/posts", tags=["Posts"]) @router.get("/", response_model=List[schemas.PostOut]) def get_posts( db: Session = Depends(get_db), current_user: int = Depends(oauth2.get_current_user), limit: int = 10, skip: int = 0, search: Optional[str] = "" ): results = ( db.query(models.Post, func.count(models.Vote.post_id).label("votes")) .join(models.Vote, models.Vote.post_id == models.Post.id, isouter=True) .group_by(models.Post.id) .filter(models.Post.title.contains(search)) .limit(limit) .offset(skip) .all() ) return results ``` Returned JSON shape (example): ```json theme={null} [ { "Post": { "title": "something something beaches hello", "content": "something something beaches", "published": true, "id": 19, "created_at": "2021-08-28T22:38:44.511524-04:00", "owner_id": 21, "owner": { "id": 21, "email": "sanjeev1@gmail.com", "created_at": "2021-08-28T21:09:08.032365-04:00" } }, "votes": 0 } ] ``` ## Alternative: flattening results into a single dict per post If you prefer a top-level post object that includes `votes` (no nested `Post` key), transform the query rows before returning: ```python theme={null} rows = ( db.query(models.Post, func.count(models.Vote.post_id).label("votes")) .join(models.Vote, models.Vote.post_id == models.Post.id, isouter=True) .group_by(models.Post.id) .filter(models.Post.title.contains(search)) .limit(limit) .offset(skip) .all() ) flattened = [] for post, votes in rows: post_dict = dict(post.__dict__) # remove SQLAlchemy protected attribute post_dict.pop("_sa_instance_state", None) post_dict["votes"] = votes flattened.append(post_dict) # Then return flattened with a response_model that matches this structure return flattened ``` This approach lets you keep an existing `schemas.PostWithVotes` (or similar) with `votes` as a top-level field. ## Quick reference table — strategies and use cases | Strategy | When to use | Resulting response shape | | -------------------------------------: | -------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- | | Create a new response schema (PostOut) | You want to return joined entities exactly as the ORM provides (possibly nested under model names) | Each item is a structured object matching the join (e.g., `{"Post": {...}, "votes": 3}`) | | Flatten results to dicts | You prefer a single-level object containing post fields + aggregated columns | Each item is a dict with post fields and top-level `votes` integer | | LEFT OUTER JOIN (isouter=True) | Include posts with zero votes | Zero-count rows are included | | INNER JOIN (default) | Only include posts that have at least one matching vote row | Rows without votes excluded | ## Best practices and tips * Always inspect generated SQL when debugging: `str(query)` or `query.statement`. * Use `isouter=True` for left outer joins to include items without matches. * Set `orm_mode = True` on Pydantic models that accept ORM objects. * Ensure your response\_model exactly matches the keys, nesting, and capitalization of the returned data. * If returning ORM instances directly, prefer response models designed for ORM dataclasses; when returning dicts, use plain Pydantic models. ## Links and references * [SQLAlchemy ORM Query API](https://docs.sqlalchemy.org/en/14/orm/query.html) * [SQLAlchemy func and aggregates](https://docs.sqlalchemy.org/en/14/core/functions.html) * [FastAPI response models and Pydantic](https://fastapi.tiangolo.com/tutorial/response-model/) * [Pydantic orm\_mode documentation](https://pydantic-docs.helpmanual.io/usage/models/#orm-mode) Summary * Build the SQLAlchemy query step-by-step: select entities, join (use `isouter=True` for LEFT JOIN), aggregate with `func`, `group_by`, then apply filters, limit, and offset. * When returning results from joined queries, choose between updating/creating response models or flattening the result to match an existing schema. * Always ensure Pydantic models match the exact structure returned by your endpoints to avoid validation errors.