ffsshhttiikk PostgreSQL
What I do
I am a powerful, open-source relational database management system (RDBMS) known for its robustness, extensibility, and standards compliance. I support advanced features including ACID transactions, complex queries, foreign keys, triggers, updatable views, and stored procedures. I excel at handling complex data relationships, supporting JSON/JSONB for document-style data, full-text search, and advanced indexing strategies. I am widely used for mission-critical applications requiring data integrity and complex querying capabilities.
When to use me
- Building enterprise applications requiring ACID compliance and complex transactions
- Applications needing advanced SQL features (window functions, CTEs, recursive queries)
- Systems requiring robust data integrity and referential integrity constraints
- Applications combining relational and document-based data (JSON/JSONB)
- Projects needing strong extensibility with custom types, functions, and extensions
- Data warehousing and analytical workloads with complex aggregations
- Geographic information systems (GIS) with PostGIS extension
- Full-text search applications without external search engines
Core Concepts
- ACID Transactions: Atomic, Consistent, Isolated, Durable transactions ensure data integrity even during system failures
- MVCC (Multi-Version Concurrency Control): Allows concurrent reads and writes without locking by maintaining multiple versions of data
- Index Types: B-tree (default), Hash, GiST (geometric), SP-GiST, GIN (generalized inverted index), and BRIN (block range)
- JSON/JSONB: Native support for semi-structured data with full query capabilities on JSON documents
- PostgreSQL Extensions: Modular functionality including PostGIS, pg_cron, pg_partman, and custom extensions
- Query Optimization: Cost-based optimizer using statistics to choose optimal execution plans
- Connection Pooling: pgBouncer or built-in connection pooling for managing database connections
- Replication: Streaming replication, logical replication, and bidirectional replication for high availability
- Partitioning: Table partitioning for managing large datasets and improving query performance
- Window Functions: Advanced analytical queries with ROW_NUMBER, RANK, LAG, LEAD functions
Code Examples
Basic Connection and Query Execution
import psycopg2
from psycopg2 import sql
from contextlib import contextmanager
@contextmanager
def get_connection():
conn = psycopg2.connect(
host="localhost",
database="app_db",
user="app_user",
password="secure_password",
port=5432
)
try:
yield conn
finally:
conn.close()
def fetch_users_with_orders(limit=100):
query = """
SELECT u.id, u.email, u.created_at, COUNT(o.id) as order_count
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
WHERE u.created_at >= %s
GROUP BY u.id, u.email, u.created_at
ORDER BY order_count DESC
LIMIT %s
"""
with get_connection() as conn:
with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
cur.execute(query, ("2024-01-01", limit))
return cur.fetchall()
def insert_product_with_categories(product_data, category_ids):
with get_connection() as conn:
with conn.cursor() as cur:
cur.execute("""
INSERT INTO products (name, description, price, sku)
VALUES (%s, %s, %s, %s)
RETURNING id
""", (product_data["name"], product_data["description"],
product_data["price"], product_data["sku"]))
product_id = cur.fetchone()[0]
cur.executemany("""
INSERT INTO product_categories (product_id, category_id)
VALUES (%s, %s)
""", [(product_id, cat_id) for cat_id in category_ids])
conn.commit()
return product_id
Advanced Query with JSONB and Window Functions
import json
from psycopg2.extras import Json
def get_user_analytics():
query = """
WITH user_purchases AS (
SELECT
user_id,
created_at::date as purchase_date,
SUM(total_amount) as daily_total,
COUNT(*) as purchase_count
FROM orders
WHERE created_at >= CURRENT_DATE - INTERVAL '30 days'
GROUP BY user_id, created_at::date
)
SELECT
user_id,
purchase_date,
daily_total,
purchase_count,
SUM(daily_total) OVER (PARTITION BY user_id ORDER BY purchase_date
ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) as rolling_7day_total,
ROW_NUMBER() OVER (PARTITION BY purchase_date ORDER BY daily_total DESC) as daily_rank
FROM user_purchases
ORDER BY user_id, purchase_date
"""
with get_connection() as conn:
with conn.cursor() as cur:
cur.execute(query)
return cur.fetchall()
def search_products_by_attributes(filters):
where_clauses = []
params = []
if "category" in filters:
where_clauses.append("attributes->>'category' = %s")
params.append(filters["category"])
if "min_price" in filters:
where_clauses.append("(attributes->>'price')::decimal >= %s")
params.append(filters["min_price"])
if "tags" in filters:
where_clauses.append("attributes->'tags' @> %s::jsonb")
params.append(json.dumps(filters["tags"]))
query = f"""
SELECT id, name, attributes
FROM products
{'WHERE ' + ' AND '.join(where_clauses) if where_clauses else ''}
ORDER BY (attributes->>'popularity')::int DESC
LIMIT 50
"""
with get_connection() as conn:
with conn.cursor() as cur:
cur.execute(query, params)
return cur.fetchall()
Transaction Management with Savepoints
from psycopg2 import DatabaseError
def process_order_with_inventory(order_data, items):
with get_connection() as conn:
with conn.cursor() as cur:
try:
cur.execute("BEGIN")
cur.execute("""
INSERT INTO orders (user_id, total, status)
VALUES (%s, %s, 'pending')
RETURNING id
""", (order_data["user_id"], order_data["total"]))
order_id = cur.fetchone()[0]
cur.execute("SAVEPOINT before_items")
for item in items:
cur.execute("""
UPDATE inventory
SET quantity = quantity - %s
WHERE product_id = %s AND quantity >= %s
""", (item["quantity"], item["product_id"], item["quantity"]))
if cur.rowcount == 0:
cur.execute("ROLLBACK TO SAVEPOINT before_items")
raise ValueError(f"Insufficient inventory for product {item['product_id']}")
cur.execute("""
INSERT INTO order_items (order_id, product_id, quantity, price)
VALUES (%s, %s, %s, %s)
""", (order_id, item["product_id"], item["quantity"], item["price"]))
cur.execute("COMMIT")
return order_id
except (DatabaseError, ValueError) as e:
cur.execute("ROLLBACK")
raise e
Using COPY for Bulk Operations
import io
from psycopg2.extras import execute_values
def bulk_insert_products(products):
with get_connection() as conn:
with conn.cursor() as cur:
execute_values(cur, """
INSERT INTO products (name, sku, price, category, attributes, created_at)
VALUES %s
ON CONFLICT (sku) DO UPDATE SET
name = EXCLUDED.name,
price = EXCLUDED.price,
attributes = EXCLUDED.attributes,
updated_at = CURRENT_TIMESTAMP
""", products)
conn.commit()
def export_orders_to_csv(start_date, end_date):
query = """
COPY (
SELECT o.id, o.created_at, u.email, p.name, oi.quantity, oi.price
FROM orders o
JOIN users u ON o.user_id = u.id
JOIN order_items oi ON o.id = oi.order_id
JOIN products p ON oi.product_id = p.id
WHERE o.created_at BETWEEN %s AND %s
ORDER BY o.created_at
) TO STDOUT WITH CSV HEADER
"""
with get_connection() as conn:
with conn.cursor() as cur:
buffer = io.StringIO()
cur.copy_expert(query, buffer, size=8192)
return buffer.getvalue()
def import_products_from_csv(csv_file):
with get_connection() as conn:
with conn.cursor() as cur:
cur.execute("""
CREATE TEMP TABLE temp_products (
name TEXT, sku TEXT, price DECIMAL, category TEXT
) ON COMMIT DROP
""")
buffer = io.StringIO(csv_file)
cur.copy_from(buffer, "temp_products", columns=("name", "sku", "price", "category"), sep=",")
cur.execute("""
INSERT INTO products (name, sku, price, category)
SELECT name, sku, price, category FROM temp_products
ON CONFLICT (sku) DO NOTHING
""")
conn.commit()
Using Asynchronous Connections
import asyncio
import asyncpg
async def fetch_user_stats(user_id):
conn = await asyncpg.connect(
host="localhost",
database="app_db",
user="app_user",
password="secure_password"
)
try:
async with conn.transaction():
user = await conn.fetchrow("""
SELECT id, email, created_at FROM users WHERE id = $1
""", user_id)
orders = await conn.fetch("""
SELECT COUNT(*) as total_orders, SUM(total) as total_spent
FROM orders WHERE user_id = $1
""", user_id)
return {"user": dict(user), "orders": dict(orders[0])}
finally:
await conn.close()
async def bulk_insert_events(events):
conn = await asyncpg.connect(
host="localhost",
database="app_db",
user="app_user",
password="secure_password"
)
try:
await conn.executemany("""
INSERT INTO events (user_id, event_type, metadata, created_at)
VALUES ($1, $2, $3, $4)
""", events)
finally:
await conn.close()
async def get_dashboard_metrics():
conn = await asyncpg.connect(
host="localhost",
database="app_db",
user="app_user",
password="secure_password"
)
try:
queries = [
("SELECT COUNT(*) FROM users", "total_users"),
("SELECT COUNT(*) FROM orders WHERE created_at > CURRENT_DATE", "today_orders"),
("SELECT SUM(total) FROM orders WHERE created_at > CURRENT_DATE", "today_revenue"),
("SELECT COUNT(*) FROM products WHERE stock < 10", "low_stock_products")
]
results = await asyncio.gather(
*[conn.fetchval(query) for query, _ in queries]
)
return dict(zip([name for _, name in queries], results))
finally:
await conn.close()
Best Practices
- Use Connection Pooling: Implement connection pooling with pgBouncer to handle connection overhead and prevent connection exhaustion under load
- Optimize Indexes: Create appropriate indexes based on query patterns; use EXPLAIN ANALYZE to identify missing indexes
- Partition Large Tables: Use table partitioning for tables exceeding 10GB or with clear partition keys (time-based, range-based)
- Implement Proper Backups: Use pg_basebackup for physical backups and pg_dump for logical backups; test restoration regularly
- Configure Appropriate Checkpoint Intervals: Tune checkpoint_completion_target and checkpoint_segments for write-heavy workloads
- Use Prepared Statements: For frequently executed queries, use prepared statements to reduce planning overhead
- Monitor Query Performance: Track slow queries using pg_stat_statements and set appropriate log_min_duration_statement
- Implement Proper Vacuuming: Configure autovacuum appropriately and run manual VACUUM ANALYZE when needed
- Use Proper Data Types: Choose appropriate data types (INT vs BIGINT, VARCHAR limits, JSONB vs TEXT) for storage efficiency
- Secure Your Instance: Use SSL/TLS for connections, implement proper authentication (md5, scram-sha-256), and restrict network access