manutej/luxor-claude-marketplace/psycopg

# Basic installation (pure Python implementation)
pip install psycopg
 
# With binary package for better performance
pip install "psycopg[binary]"
 
# With connection pool support
pip install "psycopg[pool]"
 
# Complete installation with all extras
pip install "psycopg[binary,pool]"
 
# For C implementation (fastest, requires build tools)
pip install "psycopg[c]"

# Binary package (recommended for most cases)
pip install psycopg2-binary
 
# Source package (requires PostgreSQL dev libraries)
pip install psycopg2

# Production support database (read-only)
SUPPORT_DB_HOST=support-db.company.com
SUPPORT_DB_PORT=5432
SUPPORT_DB_NAME=support_analytics
SUPPORT_DB_USER=support_readonly
SUPPORT_DB_PASSWORD=secure_password_here
 
# Development/testing database
DEV_DB_HOST=localhost
DEV_DB_PORT=5432
DEV_DB_NAME=support_dev
DEV_DB_USER=dev_user
DEV_DB_PASSWORD=dev_password

from dotenv import load_dotenv
import os
load_dotenv()
DB_CONFIG = {
    'host': os.getenv('SUPPORT_DB_HOST'),
    'port': os.getenv('SUPPORT_DB_PORT'),
    'dbname': os.getenv('SUPPORT_DB_NAME'),
    'user': os.getenv('SUPPORT_DB_USER'),
    'password': os.getenv('SUPPORT_DB_PASSWORD'),
}

import psycopg
# Using connection string
conn = psycopg.connect(
    "postgresql://user:password@localhost:5432/support_db"
)
# Using parameters (recommended for flexibility)
conn = psycopg.connect(
    host="localhost",
    port=5432,
    dbname="support_db",
    user="support_user",
    password="secure_password",
    autocommit=False,  # Explicit transaction management
    connect_timeout=10  # Timeout in seconds
)
# Always use context manager for automatic cleanup
with psycopg.connect(**DB_CONFIG) as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT version()")
        print(cur.fetchone()[0])
# Connection automatically closed

import psycopg
import time
import logging
from typing import Optional
logger = logging.getLogger(__name__)
def connect_with_retry(
    max_attempts: int = 5,
    retry_delay: float = 2.0,
    backoff_factor: float = 2.0,
    **conn_params
) -> Optional[psycopg.Connection]:
    """
    Establish database connection with exponential backoff retry.
    Args:
        max_attempts: Maximum number of connection attempts
        retry_delay: Initial delay between retries (seconds)
        backoff_factor: Multiplier for exponential backoff
        **conn_params: Connection parameters for psycopg.connect()
    Returns:
        Connection object or None if all attempts fail
    """
    attempt = 0
    current_delay = retry_delay
    while attempt < max_attempts:
        try:
            conn = psycopg.connect(**conn_params)
            logger.info(f"Database connection established on attempt {attempt + 1}")
            return conn
        except psycopg.OperationalError as e:
            attempt += 1
            if attempt >= max_attempts:
                logger.error(f"Failed to connect after {max_attempts} attempts: {e}")
                raise
            logger.warning(
                f"Connection attempt {attempt} failed: {e}. "
                f"Retrying in {current_delay:.1f}s..."
            )
            time.sleep(current_delay)
            current_delay *= backoff_factor
    return None
# Usage in customer support tools
try:
    conn = connect_with_retry(
        host="support-db.company.com",
        dbname="support_analytics",
        user="support_user",
        password=os.getenv("DB_PASSWORD"),
        max_attempts=5
    )
    with conn:
        # Perform operations
        pass
except psycopg.OperationalError:
    logger.critical("Database unavailable - alerting on-call team")
    # Send alert to monitoring system

from psycopg_pool import ConnectionPool
import logging
logging.basicConfig(level=logging.INFO)
logging.getLogger("psycopg.pool").setLevel(logging.INFO)
# Create a global connection pool
pool = ConnectionPool(
    conninfo="postgresql://user:password@localhost/support_db",
    min_size=2,          # Minimum connections to maintain
    max_size=10,         # Maximum connections allowed
    max_idle=300,        # Close idle connections after 5 minutes
    max_lifetime=3600,   # Recycle connections after 1 hour
    timeout=30,          # Wait up to 30s for a connection
    check=ConnectionPool.check_connection,  # Validate connections
    open=True            # Open pool immediately
)
# Wait for minimum connections to be established
pool.wait()
# Use in application code
def get_customer_tickets(customer_id: int):
    """Fetch all tickets for a customer."""
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT ticket_id, subject, status, created_at "
                "FROM tickets WHERE customer_id = %s "
                "ORDER BY created_at DESC",
                (customer_id,)
            )
            return cur.fetchall()
# Monitor pool health
def check_pool_health():
    """Get pool statistics for monitoring."""
    stats = pool.get_stats()
    return {
        'pool_size': stats.get('pool_size', 0),
        'available': stats.get('pool_available', 0),
        'waiting': stats.get('requests_waiting', 0),
        'requests_total': stats.get('requests_num', 0),
        'requests_queued': stats.get('requests_queued', 0),
        'connection_errors': stats.get('connections_errors', 0)
    }
# Cleanup on application shutdown
import atexit
atexit.register(pool.close)

from psycopg_pool import AsyncConnectionPool
from contextlib import asynccontextmanager
from fastapi import FastAPI
import logging
logger = logging.getLogger(__name__)
# Create async pool
async_pool = AsyncConnectionPool(
    conninfo="postgresql://user:password@localhost/support_db",
    min_size=5,
    max_size=20,
    open=False  # Open during application startup
)
# FastAPI integration with lifespan
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage async pool lifecycle with FastAPI."""
    await async_pool.open()
    logger.info("Async connection pool opened")
    yield
    await async_pool.close()
    logger.info("Async connection pool closed")
app = FastAPI(lifespan=lifespan)
# Async endpoint using the pool
@app.get("/api/customers/{customer_id}/tickets")
async def get_customer_tickets(customer_id: int):
    """Fetch customer tickets asynchronously."""
    async with async_pool.connection() as conn:
        async with conn.cursor() as cur:
            await cur.execute(
                "SELECT ticket_id, subject, status, priority, created_at "
                "FROM tickets "
                "WHERE customer_id = %s "
                "ORDER BY created_at DESC "
                "LIMIT 100",
                (customer_id,)
            )
            tickets = await cur.fetchall()
            return [
                {
                    'ticket_id': t[0],
                    'subject': t[1],
                    'status': t[2],
                    'priority': t[3],
                    'created_at': t[4].isoformat()
                }
                for t in tickets
            ]
# Health check endpoint
@app.get("/health/database")
async def database_health():
    """Check database and pool health."""
    try:
        async with async_pool.connection() as conn:
            async with conn.cursor() as cur:
                await cur.execute("SELECT 1")
                result = await cur.fetchone()
        stats = async_pool.get_stats()
        return {
            'status': 'healthy',
            'database_responsive': result[0] == 1,
            'pool_stats': {
                'size': stats.get('pool_size'),
                'available': stats.get('pool_available'),
                'waiting': stats.get('requests_waiting')
            }
        }
    except Exception as e:
        logger.error(f"Database health check failed: {e}")
        return {'status': 'unhealthy', 'error': str(e)}

import asyncio
import psycopg
async def fetch_support_metrics():
    """Fetch support metrics asynchronously."""
    async with await psycopg.AsyncConnection.connect(
        "postgresql://user:password@localhost/support_db"
    ) as conn:
        async with conn.cursor() as cur:
            # Execute query asynchronously
            await cur.execute(
                """
                SELECT
                    status,
                    COUNT(*) as count,
                    AVG(response_time_minutes) as avg_response
                FROM tickets
                WHERE created_at > CURRENT_DATE - INTERVAL '7 days'
                GROUP BY status
                """
            )
            # Fetch results
            results = await cur.fetchall()
            # Process asynchronously
            metrics = {}
            async for row in cur:
                metrics[row[0]] = {
                    'count': row[1],
                    'avg_response': float(row[2]) if row[2] else 0
                }
            return metrics
# Run async function
if __name__ == "__main__":
    metrics = asyncio.run(fetch_support_metrics())
    print(metrics)

import asyncio
import psycopg
from typing import List, Tuple
async def fetch_customer_data(customer_id: int, conn: psycopg.AsyncConnection):
    """Fetch all data for a customer."""
    async with conn.cursor() as cur:
        await cur.execute(
            "SELECT * FROM customers WHERE customer_id = %s",
            (customer_id,)
        )
        return await cur.fetchone()