Building a List Endpoint

1. Anatomy of the "List" Endpoint

To create an endpoint that returns a list of items, we need to combine several concepts we've already touched upon:

• Path Operation Decorator: We'll use @router.get("/posts") to define the HTTP method (GET) and the path (/posts).
• Dependency Injection: We will use db: AsyncSession = Depends(get_async_session) in our function signature to get a database session.
• Response Model: This is a key part. Since our endpoint will return multiple posts, not just one, we need to tell FastAPI that the response will be a list of our Pydantic PostRead schema. We do this with response_model=list[PostRead].

The video below, although using a synchronous approach, clearly demonstrates setting the response_model to a list of Pydantic models. The concept is identical in our asynchronous implementation.

2. Asynchronous Querying with SQLAlchemy 2.0

Now for the core logic: how do we actually ask the database for all the posts? With SQLAlchemy's modern async support, this involves a clear, three-step process inside our endpoint function.

1. select(Post): We use the select function from SQLAlchemy to create a Select statement object. This object describes our query—"I want to select all columns from the Post table"—but it doesn't actually run it yet.
2. await db.execute(...): We pass the Select object to the db.execute() method. Since this is a network operation (talking to the database), we await it. This is the point where the query is sent to the database and executed.
3. .scalars().all(): The result of execute is a Result object that contains the rows returned by the database.
   • .scalars() extracts the first item from each row, which in our case is the full Post ORM object that SQLAlchemy constructs.
   • .all() fetches all of these scalar results and puts them into a standard Python list.

The following video is an excellent, concise demonstration of this exact process. It shows how to build an async "list" endpoint from scratch using this modern syntax.

This pattern is the standard for async queries in SQLAlchemy 2.0. To reinforce this, the following article shows the exact same select -> execute -> scalars pattern in a text-based format.

3. Hands-On: Implementing the GET /posts Endpoint

Let's put this all together. We'll create a new router file for our post-related endpoints.

1. If you haven't already, create a directory named routers in your project's root.
2. Inside the routers directory, create a new file named posts.py.
3. Add the following code to routers/posts.py:

# routers/posts.py

from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession




# Import our models and schemas
from models import Post
from schemas import PostRead
from database import get_async_session




# Create a new router
router = APIRouter()


@router.get("/posts", response_model=list[PostRead])
async def get_all_posts(db: AsyncSession = Depends(get_async_session)):
    """
    Get all posts from the database.
    """



    # Create a select statement
    query = select(Post)
    



    # Execute the query
    result = await db.execute(query)
    



    # Get the scalar results (the Post objects) and convert to a list
    posts = result.scalars().all()
    
    return posts

4. Now, you need to include this new router in your main application file (main.py).

# main.py

from fastapi import FastAPI
from contextlib import asynccontextmanager

from database import create_db_and_tables
from routers import posts  # Import the posts router




# Lifespan function to create DB tables on startup
@asynccontextmanager
async def lifespan(app: FastAPI):
    print("Creating database and tables...")
    await create_db_and_tables()
    yield
    print("Shutting down...")

app = FastAPI(lifespan=lifespan)




# Include the posts router
app.include_router(posts.router)

@app.get("/")
def read_root():
    return {"message": "Welcome to the Blog API"}

Testing your Endpoint

Start your application with uvicorn:
uvicorn main:app --reload

Navigate to http://127.0.0.1:8000/docs. You should now see your new /posts endpoint listed.

Try it out! Click "Execute". The response will likely be an empty array [], because our database is empty. That's perfectly fine! We've successfully built the endpoint to read the data; we just haven't added the endpoint to create data yet. This empty response proves our code is working correctly.

The JSON data you see is what a frontend application would use to display a list of blog posts, perhaps in a table like the one shown below.

Conclusion

Excellent work! You have successfully created your first database-driven API endpoint. You've learned how to combine path operations, dependency injection, and Pydantic response models with the modern, asynchronous querying style of SQLAlchemy 2.0.

Key Takeaways:

• To return a list of items, set the response_model to list[YourPydanticSchema].
• The asynchronous query pattern is:
  1. Create a query with select().
  2. Execute it with await db.execute().
  3. Extract the ORM objects with .scalars().all().
• Routers (APIRouter) help organize your endpoints into logical modules (e.g., all post-related endpoints go in routers/posts.py).

In the next lesson, we'll build on this by implementing the second fundamental "read" operation: retrieving a single record by its ID. This will introduce us to path parameters and how to add a WHERE clause to our database queries.