Step 3
Step 3 — PostgreSQL connection
30 min
Step 3 — PostgreSQL connection
If your API needs to remember, you need a DB. PostgreSQL is the consensus pick.
Connection pool — opening fresh is expensive
A new DB connection takes tens of milliseconds. Pre-open N connections and lend them out.
import os
from psycopg2 import pool
_pg_pool = None
def get_pool():
global _pg_pool
if _pg_pool is None:
_pg_pool = pool.SimpleConnectionPool(
minconn=1, maxconn=10,
host=os.getenv("DB_HOST", "localhost"),
port=int(os.getenv("DB_PORT", "5432")),
database=os.getenv("DB_NAME", "mydb"),
user=os.getenv("DB_USER", "user"),
password=os.getenv("DB_PASSWORD", "secret"),
)
return _pg_pool
Context manager for safety
from contextlib import contextmanager
@contextmanager
def get_conn():
p = get_pool()
conn = p.getconn()
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
p.putconn(conn)
with get_conn() as conn: — auto rollback on error, commit on success, always returns to the pool.
First query
def find_recent(limit: int = 20):
with get_conn() as conn, conn.cursor() as cur:
cur.execute(
"SELECT id, title, body, created_at FROM posts "
"ORDER BY created_at DESC LIMIT %s",
(limit,),
)
return [
{"id": r[0], "title": r[1], "body": r[2], "created_at": r[3]}
for r in cur.fetchall()
]
%s placeholder — never build SQL with f-strings (injection risk).
.env for secrets
DB_HOST=localhost
DB_PORT=5432
DB_NAME=mydb
DB_USER=user
DB_PASSWORD=secret
from dotenv import load_dotenv
load_dotenv()
Add .env to .gitignore.
Try it
Spin up Postgres with docker run -d -p 5432:5432 -e POSTGRES_PASSWORD=secret postgres:17, create a posts table, and call find_recent.
Next
Step 4 schedules recurring jobs.