PostgreSQL — Code
Practical Python from Basics to Production
Real code patterns for every database operation you will use. Built around SQLAlchemy
async + asyncpg and raw SQL.
Ringlet Dev Series • Basics → Advanced • 2025
Setup and Connection
Install
pip install sqlalchemy asyncpg alembic psycopg2-binary
Async engine (Ringlet uses this)
from [Link] import create_async_engine, AsyncSession
from [Link] import sessionmaker, DeclarativeBase
import os
DATABASE_URL = [Link]('DATABASE_URL')
# Must use postgresql+asyncpg:// for async
engine = create_async_engine(
DATABASE_URL,
echo=False, # Set True to log all SQL (dev only)
pool_size=10, # Max persistent connections
max_overflow=20, # Extra connections under load
pool_pre_ping=True, # Test connections before using
AsyncSessionLocal = sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
class Base(DeclarativeBase):
pass
Dependency — get a session per request
from contextlib import asynccontextmanager
@asynccontextmanager
async def get_db():
async with AsyncSessionLocal() as session:
try:
yield session
await [Link]()
except Exception:
await [Link]()
raise
Defining Models — Ringlet Schema
from sqlalchemy import Column, String, Integer, DateTime, Boolean, ForeignKey,
Text
from [Link] import UUID
from [Link] import relationship
from datetime import datetime, timezone
import uuid
class Business(Base):
__tablename__ = 'businesses'
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
name = Column(String(255), nullable=False)
phone = Column(String(20), unique=True, nullable=False)
timezone = Column(String(50), default='Asia/Kolkata')
created_at = Column(DateTime(timezone=True), default=lambda:
[Link]([Link]))
appointments = relationship('Appointment', back_populates='business')
class Appointment(Base):
__tablename__ = 'appointments'
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
business_id = Column(UUID(as_uuid=True), ForeignKey('[Link]'),
nullable=False)
caller_name = Column(String(255), nullable=False)
caller_phone = Column(String(20), nullable=False)
service = Column(String(100), nullable=False)
appointment_dt = Column(DateTime(timezone=True), nullable=False)
duration_mins = Column(Integer, nullable=False)
status = Column(String(20), default='confirmed') # confirmed/cancelled
reminder_sent = Column(Boolean, default=False)
created_at = Column(DateTime(timezone=True), default=lambda:
[Link]([Link]))
business = relationship('Business', back_populates='appointments')
class CallLog(Base):
__tablename__ = 'call_logs'
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
call_sid = Column(String(100), unique=True, nullable=False)
business_id = Column(UUID(as_uuid=True), ForeignKey('[Link]'))
caller = Column(String(20))
duration_s = Column(Integer)
outcome = Column(String(50)) # booked/faq/cancelled/failed
transcript = Column(Text)
created_at = Column(DateTime(timezone=True), default=lambda:
[Link]([Link]))
Migrations with Alembic
Setup — run once
alembic init alembic
# Then edit alembic/[Link] to point at your models
alembic/[Link] — key changes
from [Link] import Base # import your models
target_metadata = [Link]
# And set the URL:
config.set_main_option('[Link]',
[Link]('DATABASE_URL').replace('+asyncpg', ''))
# Alembic uses sync psycopg2 — strip the asyncpg driver
Daily workflow
# After changing a model, generate a migration
alembic revision --autogenerate -m 'add reminder_sent to appointments'
# Apply all pending migrations
alembic upgrade head
# Roll back one migration
alembic downgrade -1
# See current state
alembic current
# See migration history
alembic history
CRUD Operations
INSERT — Create an appointment
from sqlalchemy import select
async def create_appointment(db: AsyncSession, data: dict) -> Appointment:
appt = Appointment(
business_id = data['business_id'],
caller_name = data['caller_name'],
caller_phone = data['caller_phone'],
service = data['service'],
appointment_dt = data['appointment_dt'],
duration_mins = data['duration_mins'],
[Link](appt)
await [Link]() # Gets the ID without committing
await [Link](appt) # Reload from DB
return appt
SELECT — Query appointments
from datetime import datetime, timezone, timedelta
# Get one by ID
async def get_appointment(db: AsyncSession, appt_id: str) -> Appointment | None:
result = await [Link](select(Appointment).where([Link] == appt_id))
return result.scalar_one_or_none()
# Get upcoming appointments needing reminders
async def get_reminder_targets(db: AsyncSession) -> list[Appointment]:
now = [Link]([Link])
in_24h = now + timedelta(hours=24)
result = await [Link](
select(Appointment)
.where(
Appointment.appointment_dt.between(now, in_24h),
[Link] == 'confirmed',
Appointment.reminder_sent == False
.order_by(Appointment.appointment_dt)
return [Link]().all()
# Check availability — no confirmed appts in this slot
async def is_slot_available(
db: AsyncSession, business_id: str,
start: datetime, duration_mins: int
) -> bool:
end = start + timedelta(minutes=duration_mins)
result = await [Link](
select(Appointment)
.where(
Appointment.business_id == business_id,
[Link] == 'confirmed',
Appointment.appointment_dt < end,
(Appointment.appointment_dt +
timedelta(minutes=duration_mins)) > start
return result.scalar_one_or_none() is None
UPDATE and DELETE
from sqlalchemy import update, delete
# UPDATE — mark reminder as sent
async def mark_reminder_sent(db: AsyncSession, appt_id: str):
await [Link](
update(Appointment)
.where([Link] == appt_id)
.values(reminder_sent=True)
# UPDATE — cancel appointment
async def cancel_appointment(db: AsyncSession, appt_id: str):
await [Link](
update(Appointment)
.where([Link] == appt_id)
.values(status='cancelled')
# DELETE — hard delete (rarely used, prefer status='cancelled')
async def delete_appointment(db: AsyncSession, appt_id: str):
await [Link](
delete(Appointment).where([Link] == appt_id)
)
Raw SQL — When You Need Full Power
Use raw SQL for complex queries SQLAlchemy ORM makes ugly:
from sqlalchemy import text
# Available slots for a given day
async def get_available_slots(db: AsyncSession, business_id: str, date: str) ->
list:
result = await [Link](text('''
SELECT
generate_series(
:date::date + time '09:00',
:date::date + time '18:00',
interval '45 minutes'
) AS slot
EXCEPT
SELECT appointment_dt
FROM appointments
WHERE
business_id = :business_id AND
DATE(appointment_dt) = :date AND
status = 'confirmed'
ORDER BY slot
'''), {'business_id': business_id, 'date': date})
return [[Link] for row in result]
# Daily summary for the business
async def daily_summary(db: AsyncSession, business_id: str, date: str) -> dict:
result = await [Link](text('''
SELECT
COUNT(*) AS total,
COUNT(*) FILTER (WHERE status = 'confirmed') AS confirmed,
COUNT(*) FILTER (WHERE status = 'cancelled') AS cancelled,
STRING_AGG(service, ', ') AS services
FROM appointments
WHERE business_id = :business_id
AND DATE(appointment_dt) = :date
'''), {'business_id': business_id, 'date': date})
return dict([Link]().one())
Transactions — Explicit Control
# The get_db() context manager commits on exit, rolls back on error.
# For more control, use explicit transactions:
async def book_with_lock(db: AsyncSession, data: dict):
# SELECT FOR UPDATE — lock the slot row while we work
result = await [Link](
select(Appointment)
.where(
Appointment.business_id == data['business_id'],
Appointment.appointment_dt == data['appointment_dt'],
[Link] == 'confirmed'
.with_for_update() # Locks the row
existing = result.scalar_one_or_none()
if existing:
raise ValueError('Slot already booked')
appt = Appointment(**data)
[Link](appt)
# Session commits when get_db() context exits
return appt
# SAVEPOINT — partial rollback within a transaction
async def with_savepoint(db: AsyncSession):
async with db.begin_nested() as savepoint:
try:
[Link](SomeModel(...))
await [Link]()
except Exception:
await [Link]() # Only undoes this nested block
raise
Indexes in Practice
Add these indexes to Ringlet's schema for production performance:
# In your Alembic migration or model definition
from sqlalchemy import Index
# Find appointments by caller phone (cancel/lookup flow)
Index('idx_appt_caller_phone', Appointment.caller_phone)
# Find upcoming appointments for reminders (runs every hour)
Index('idx_appt_dt_status', Appointment.appointment_dt, [Link])
# Partial index — only index confirmed appointments
# (Much smaller, reminder job only cares about confirmed)
Index(
'idx_appt_confirmed_reminder',
Appointment.appointment_dt,
Appointment.reminder_sent,
postgresql_where=([Link] == 'confirmed')
# Business lookup by phone (inbound call matching)
Index('idx_business_phone', [Link], unique=True)
# Call log lookup by call_sid (real-time call handling)
Index('idx_calllog_sid', CallLog.call_sid, unique=True)
EXPLAIN — Debug Slow Queries
Run this in psql to see the query plan:
-- See the plan (no execution)
EXPLAIN
SELECT * FROM appointments
WHERE business_id = 'uuid-here'
AND status = 'confirmed'
AND appointment_dt > NOW();
-- See the plan AND run it (real timings)
EXPLAIN (ANALYZE, BUFFERS)
SELECT * FROM appointments
WHERE business_id = 'uuid-here'
AND status = 'confirmed'
AND appointment_dt > NOW();
What to look for:
• Seq Scan on a large table — you need an index
• rows=10000 but actual rows=1 — stale statistics, run ANALYZE appointments
• actual time much higher than cost estimate — something unexpected is happening
• Hash Join on a small table — might be faster with Nested Loop + index
Useful diagnostic queries
-- See all active queries
SELECT pid, now() - pg_stat_activity.query_start AS duration,
query, state
FROM pg_stat_activity
WHERE state != 'idle'
ORDER BY duration DESC;
-- Find missing indexes (sequential scans on large tables)
SELECT schemaname, tablename, seq_scan, seq_tup_read,
idx_scan, idx_tup_fetch
FROM pg_stat_user_tables
ORDER BY seq_scan DESC;
-- Table sizes
SELECT tablename,
pg_size_pretty(pg_total_relation_size(tablename::regclass)) AS size
FROM pg_tables
WHERE schemaname = 'public'
ORDER BY pg_total_relation_size(tablename::regclass) DESC;
-- Longest running queries
SELECT pid, now() - query_start AS age, query
FROM pg_stat_activity
WHERE state = 'active'
ORDER BY age DESC LIMIT 10;