-- Add a generated tsvector column (auto-updated)ALTER TABLE posts ADDCOLUMN search_vector tsvector
GENERATED ALWAYS AS (
setweight(to_tsvector('english', coalesce(title, '')), 'A') ||
setweight(to_tsvector('english', coalesce(content, '')), 'B')
) STORED;
-- GIN index for fast full-text searchCREATE INDEX idx_posts_search ON posts USING GIN(search_vector);
-- Search query with rankingSELECT
id, title,
ts_rank(search_vector, query) AS rank,
ts_headline('english', content, query, 'MaxFragments=2') AS excerpt
FROM posts, to_tsquery('english', 'typescript & patterns') AS query
WHERE search_vector @@ query
ORDERBY rank DESC
LIMIT 10;
From application code:
# asyncpgasyncdefsearch_posts(query: str) -> list[dict]:
# Convert user input to tsquery safely
tsquery = " & ".join(query.split())
rows = await pool.fetch("""
SELECT id, title,
ts_rank(search_vector, to_tsquery('english', $1)) AS rank
FROM posts
WHERE search_vector @@ to_tsquery('english', $1)
ORDER BY rank DESC
LIMIT 20
""", tsquery)
return [dict(r) for r in rows]
JSONB — flexible schema within PostgreSQL
-- JSONB columnALTER TABLE products ADDCOLUMN attributes JSONB DEFAULT'{}';
-- Insert with JSONBINSERT INTO products (name, attributes)
VALUES ('Laptop', '{"brand": "Dell", "ram_gb": 16, "tags": ["laptop", "portable"]}');
-- Query inside JSONBSELECT name, attributes->>'brand'AS brand
FROM products
WHERE attributes->>'brand'='Dell';
-- Query nested valuesSELECT name FROM products
WHERE (attributes->>'ram_gb')::int>=16;
-- Array containmentSELECT name FROM products
WHERE attributes->'tags' ? 'laptop';
-- Update a key inside JSONB (non-destructive)UPDATE products
SET attributes = attributes ||'{"ram_gb": 32}'::jsonb
WHERE id =1;
-- Remove a keyUPDATE products
SET attributes = attributes -'legacy_field'WHERE id =1;
JSONB indexes:
-- GIN index for @> (containment) and ? (key existence)CREATE INDEX idx_products_attributes ON products USING GIN(attributes);
-- btree index on a specific path (for equality/range)CREATE INDEX idx_products_brand ON products ((attributes->>'brand'));
Arrays
-- Array columnALTER TABLE users ADDCOLUMN tags TEXT[];
-- Insert with arrayINSERT INTO users (email, tags) VALUES ('alice@example.com', ARRAY['admin', 'beta']);
-- Append to arrayUPDATE users SET tags = array_append(tags, 'vip') WHERE id =1;
-- Remove from arrayUPDATE users SET tags = array_remove(tags, 'beta') WHERE id =1;
-- Check if array contains valueSELECT*FROM users WHERE'admin'=ANY(tags);
-- Overlap (any element in common)SELECT*FROM users WHERE tags &&ARRAY['admin', 'moderator'];
-- Containment (all elements present)SELECT*FROM users WHERE tags @>ARRAY['admin', 'vip'];
pg_notify — real-time with LISTEN/NOTIFY
-- Send a notification (from SQL or a trigger)
NOTIFY user_created, '{"userId": "123", "email": "alice@example.com"}';
-- Trigger-based notificationCREATEOR REPLACE FUNCTION notify_user_created()
RETURNStriggerAS $$
BEGIN
PERFORM pg_notify(
'user_created',
json_build_object('userId', NEW.id, 'email', NEW.email)::text
);
RETURNNEW;
END;
$$ LANGUAGE plpgsql;
CREATETRIGGER on_user_created
AFTER INSERTON users
FOREACHROWEXECUTEFUNCTION notify_user_created();
-- Function to safely decrement credits (atomic, no race condition)CREATEOR REPLACE FUNCTION spend_credits(
p_user_id UUID,
p_amount INTEGER
) RETURNSBOOLEANLANGUAGE plpgsql
AS $$
DECLARE
v_current INTEGER;
BEGIN-- Lock the row for this operationSELECT credits INTO v_current
FROM users
WHERE id = p_user_id
FORUPDATE;
IF v_current < p_amount THENRETURNFALSE; -- insufficient creditsEND IF;
UPDATE users
SET credits = credits - p_amount
WHERE id = p_user_id;
INSERT INTO credit_transactions (user_id, amount, type)
VALUES (p_user_id, p_amount, 'debit');
RETURNTRUE;
END;
$$;
-- Call from applicationSELECT spend_credits('user-uuid', 50);
-- Trigger function for updated_atCREATEOR REPLACE FUNCTION set_updated_at()
RETURNStriggerAS $$
BEGIN
NEW.updated_at = NOW();
RETURNNEW;
END;
$$ LANGUAGE plpgsql;
-- Apply to a tableCREATETRIGGER set_updated_at
BEFORE UPDATEON users
FOREACHROWEXECUTEFUNCTION set_updated_at();
Connection pooling with PgBouncer / Supavisor
For serverless environments, always use a pooler:
# .env# Direct connection (for migrations only)
DATABASE_URL_DIRECT="postgresql://user:pass@db.host:5432/mydb"# Pooled connection (for application queries)
DATABASE_URL="postgresql://user:pass@db.host:6543/mydb?pgbouncer=true"
# asyncpg with Supabase/Neon pooler — disable prepared statements
pool = await asyncpg.create_pool(
dsn=os.environ["DATABASE_URL"],
statement_cache_size=0, # required for PgBouncer transaction mode
)
Example
User: Add full-text search to a blog app (Python/asyncpg) with weighted title/content ranking, highlight excerpts, and a pg_notify trigger that pushes new post notifications to connected WebSocket clients.