Sql Retrieve Posts

In this lesson you’ll learn how to retrieve posts from a PostgreSQL database using psycopg2 inside a FastAPI application. We’ll cover a resilient connection pattern that retries while the database starts, how to execute a SELECT query, and how to fetch results correctly so FastAPI can return JSON responses.

1. Open a resilient database connection

Create a dedicated module (e.g., connection.py) that attempts to connect in a retry loop. This pattern is helpful when your database may still be initializing (for example in containers or managed services).

# connection.py
import time
import psycopg2
from psycopg2.extras import RealDictCursor

while True:
    try:
        conn = psycopg2.connect(
            host="localhost",
            database="fastapi",
            user="postgres",
            password="password123",
            cursor_factory=RealDictCursor
        )
        cursor = conn.cursor()
        print("Database connection was successful!")
        break
    except Exception as error:
        print("Connecting to database failed")
        print("Error:", error)
        time.sleep(2)

Make this module importable by your FastAPI app so route handlers can from connection import conn, cursor and execute queries.

Common connection errors and quick fixes

Error (example)	Meaning	Quick remediation
`FATAL: password authentication failed for user "postgres"`	Bad credentials	Verify username/password or use environment variables for credentials
`could not connect to server: Connection refused`	Database not reachable	Ensure DB is running and host/port are correct
`timeout expired`	Network or startup delay	Use retry loop (as above) or increase timeout

2. Understand cursor.execute() vs fetch methods

Important: cursor.execute() runs the SQL statement but returns None. To retrieve rows from a SELECT query you must call one of the fetch methods.

Remember: always invoke the fetch method with parentheses — e.g. cursor.fetchall() — otherwise you only have a reference to the function, not the query results.

Use the method that matches your expected result size:

Method	Use case	Example
`fetchall()`	Multiple rows (e.g., `SELECT * FROM posts`)	`posts = cursor.fetchall()`
`fetchone()`	Single row (e.g., unique `id`)	`post = cursor.fetchone()`
`fetchmany(n)`	Controlled batch size for large result sets	`rows = cursor.fetchmany(100)`

Always use parameterized queries to avoid SQL injection, passing parameters as the second argument to cursor.execute():

cursor.execute("SELECT * FROM posts WHERE id = %s", (post_id,))
post = cursor.fetchone()

For write operations (INSERT, UPDATE, DELETE), call conn.commit() to persist changes, or set conn.autocommit = True.

3. Example in-memory posts (for comparison / simple create route)

my_posts = [
    {"title": "title of post 1", "content": "content of post 1", "id": 1},
    {"title": "favorite foods", "content": "I like pizza", "id": 2}
]

4. Route handlers (main.py)

Below is an example FastAPI app that imports the database connection and demonstrates reading rows from PostgreSQL for the /posts endpoint. It also includes a simple in-memory create route to illustrate POST handling.

# main.py
from fastapi import FastAPI, status
from pydantic import BaseModel
from random import randrange

# Import the connection and cursor from the connection module
from connection import conn, cursor

app = FastAPI()

# Pydantic model for incoming post creation
class Post(BaseModel):
    title: str
    content: str
    published: bool = True

# Root endpoint
@app.get("/")
def root():
    return {"message": "Hello World"}

# Retrieve all posts from the database
@app.get("/posts")
def get_posts():
    cursor.execute("""SELECT * FROM posts""")
    posts = cursor.fetchall()  # returns a list of RealDictRow (dict-like) rows
    return {"data": posts}

# Create a post (still using in-memory list in this example)
@app.post("/posts", status_code=status.HTTP_201_CREATED)
def create_posts(post: Post):
    post_dict = post.dict()
    post_dict["id"] = randrange(0, 1000000)
    my_posts.append(post_dict)
    return {"data": post_dict}

5. Example JSON response

Example response returned by the /posts endpoint (FastAPI serializes RealDictRow objects naturally as JSON):

[
  {
    "id": 1,
    "title": "first post",
    "content": "some interesting stuff",
    "published": true,
    "created_at": "2021-08-21T23:08:39.633120-04:00"
  },
  {
    "id": 2,
    "title": "second post",
    "content": "yadayadayda",
    "published": true,
    "created_at": "2021-08-21T23:08:39.633120-04:00"
  }
]

Notes on types and formatting:

RealDictCursor yields RealDictRow objects that behave like dictionaries and serialize to JSON correctly.
PostgreSQL may add metadata columns such as created_at with timezone-aware timestamps.

Do not keep plaintext database credentials in source code for production. Use environment variables, configuration systems, or secret management (e.g., Vault, AWS Secrets Manager) and rotate credentials regularly.

6. Quick checklist and best practices

Use a retry loop for resilient connections when databases may be slow to start.
Call cursor.execute() to run SQL, then cursor.fetchall() / cursor.fetchone() / cursor.fetchmany() to retrieve results.
Use fetchall() for queries returning multiple rows and fetchone() for a single-row lookup by unique key.
For writes, call conn.commit() or enable conn.autocommit = True.
Always use parameterized queries to prevent SQL injection (e.g., cursor.execute(sql, params)).
Keep credentials out of code (use environment variables or secret stores).

Introduction

Python Setup

FastAPI Basics

Working with Databases

Databases with Python

Advanced FastAPI

Database Migration

Deployment

Testing

CI/CD

Sql Retrieve Posts

1. Open a resilient database connection

Common connection errors and quick fixes

2. Understand cursor.execute() vs fetch methods

3. Example in-memory posts (for comparison / simple create route)

4. Route handlers (main.py)

5. Example JSON response

6. Quick checklist and best practices

Links and references

Watch Video

​1. Open a resilient database connection

​Common connection errors and quick fixes

​2. Understand cursor.execute() vs fetch methods

​3. Example in-memory posts (for comparison / simple create route)

​4. Route handlers (main.py)

​5. Example JSON response

​6. Quick checklist and best practices

​Links and references

Watch Video

1. Open a resilient database connection

Common connection errors and quick fixes

2. Understand cursor.execute() vs fetch methods

3. Example in-memory posts (for comparison / simple create route)

4. Route handlers (main.py)

5. Example JSON response

6. Quick checklist and best practices

Links and references