Flask + PostgreSQL CRUD Operations

Learn how to build CRUD operations the Flask Vibe way: with raw SQL and psycopg2.

Why No ORM?

ORMs hide what's happening. When you write:

# SQLAlchemy (ORM)
user = User.query.filter_by(email=email).first()

You don't know: - What SQL is being generated - If indexes are being used - How many queries are running (N+1 problem)

Compare with Flask Vibe approach:

# psycopg2 (raw SQL)
with get_db() as conn:
    with conn.cursor() as cur:
        cur.execute(
            "SELECT * FROM users WHERE email = %s LIMIT 1",
            (email,)
        )
        user = cur.fetchone()

Crystal clear. You see exactly what's happening.

Database Connection Helper

# utils/db.py
import psycopg2
from psycopg2.extras import RealDictCursor
from contextlib import contextmanager
from config import Config

@contextmanager
def get_db():
    """Get a database connection, commit on success, rollback and close on error."""
    conn = psycopg2.connect(
        host=Config.DB_HOST,
        port=Config.DB_PORT,
        database=Config.DB_NAME,
        user=Config.DB_USER,
        password=Config.DB_PASSWORD
    )
    conn.cursor_factory = RealDictCursor
    try:
        yield conn
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()  # Always close — use connection pooling in production

For production, replace this with the pooled version from Connection Pooling with psycopg2. Pooling reuses connections instead of opening a new one per request.

CREATE: Adding Records

from flask import Flask, request, redirect
from utils.db import get_db

@app.route("/users/create", methods=["POST"])
def create_user():
    name = request.form.get("name")
    email = request.form.get("email")

    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id",
                (name, email)
            )
            user_id = cur.fetchone()['id']
            conn.commit()

    return redirect(f"/users/{user_id}")

READ: Fetching Records

@app.route("/users")
def list_users():
    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT * FROM users ORDER BY created_at DESC")
            users = cur.fetchall()
    return render_template("users.html", users=users)

@app.route("/users/<int:user_id>")
def show_user(user_id):
    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))
            user = cur.fetchone()

    if not user:
        return "User not found", 404

    return render_template("user.html", user=user)

UPDATE: Modifying Records

@app.route("/users/<int:user_id>/edit", methods=["POST"])
def update_user(user_id):
    name = request.form.get("name")
    email = request.form.get("email")

    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE users SET name = %s, email = %s, updated_at = NOW() WHERE id = %s",
                (name, email, user_id)
            )
            conn.commit()

    return redirect(f"/users/{user_id}")

DELETE: Removing Records

@app.route("/users/<int:user_id>/delete", methods=["POST"])
def delete_user(user_id):
    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute("DELETE FROM users WHERE id = %s", (user_id,))
            conn.commit()

    return redirect("/users")

AI Prompt That Generated This

"Create Flask CRUD routes for a users table with columns: id, name, email, created_at, updated_at. Use psycopg2 with parameterized queries for SQL injection protection. Include error handling for database connections."

AI loves simple, direct code. No ORM confusion, no magic methods to hallucinate.

Next Steps

• Learn about Connection Pooling with psycopg2
• Explore CSRF Protection
• Build Full-Text Search with PostgreSQL