tessl/pypi-psycopg2

Python-PostgreSQL Database Adapter

—

Pending

Overview

Eval results

Files

SQL Composition

Name: tessl/pypi-psycopg2
Author: tessl

Safe SQL query construction with automatic quoting, parameter placeholders, and composable SQL elements to prevent SQL injection vulnerabilities while building dynamic queries programmatically.

Capabilities

SQL Building Classes

Composable SQL elements that can be combined to build complex queries safely with automatic escaping and quoting.

class Composable:
    """Abstract base for SQL composition objects."""
    
    def __init__(self, wrapped):
        """Initialize with wrapped object."""
    
    def as_string(self, context):
        """Get string representation (abstract method)."""
    
    def __add__(self, other):
        """Concatenate with another Composable."""
    
    def __mul__(self, n):
        """Repeat n times."""
    
    def __eq__(self, other):
        """Equality comparison."""
    
    def __ne__(self, other):
        """Inequality comparison."""

class Composed(Composable):
    """Sequence of Composable objects."""
    
    def __init__(self, seq):
        """Initialize with sequence of Composables."""
    
    @property
    def seq(self):
        """Property returning content list."""
    
    def as_string(self, context):
        """Join all elements."""
    
    def __iter__(self):
        """Iterator over elements."""
    
    def __add__(self, other):
        """Concatenation."""
    
    def join(self, joiner):
        """
        Join elements with separator.
        
        Parameters:
        - joiner (Composable): Separator to join with
        
        Returns:
        Composed: New composed object with joined elements
        """

class SQL(Composable):
    """Raw SQL snippet."""
    
    def __init__(self, string):
        """
        Initialize with SQL string.
        
        Parameters:
        - string (str): Raw SQL text
        """
    
    @property
    def string(self):
        """Property returning wrapped string."""
    
    def as_string(self, context):
        """Return string as-is."""
    
    def format(self, *args, **kwargs):
        """
        Format with placeholders.
        
        Parameters:
        - *args: Positional Composable arguments
        - **kwargs: Named Composable arguments
        
        Returns:
        Composed: Formatted SQL composition
        """
    
    def join(self, seq):
        """
        Join sequence with SQL string.
        
        Parameters:
        - seq (sequence): Sequence of Composables to join
        
        Returns:
        Composed: Joined composition
        """

class Identifier(Composable):
    """SQL identifier (quoted)."""
    
    def __init__(self, *strings):
        """
        Initialize with identifier parts.
        
        Parameters:
        - *strings: Identifier components (schema, table, column, etc.)
        """
    
    @property
    def strings(self):
        """Property returning tuple of strings."""
    
    @property
    def string(self):
        """Property for single string (raises if multiple)."""
    
    def as_string(self, context):
        """Return quoted identifier."""

class Literal(Composable):
    """SQL literal value."""
    
    @property
    def wrapped(self):
        """Property returning wrapped object."""
    
    def as_string(self, context):
        """Return adapted/quoted value."""

class Placeholder(Composable):
    """Parameter placeholder."""
    
    def __init__(self, name=None):
        """
        Initialize with optional name.
        
        Parameters:
        - name (str, optional): Placeholder name for named parameters
        """
    
    @property
    def name(self):
        """Property returning placeholder name."""
    
    def as_string(self, context):
        """Return %s or %(name)s."""

Usage Example:

from psycopg2 import sql

# Basic SQL composition
query = sql.SQL("SELECT {fields} FROM {table} WHERE {condition}").format(
    fields=sql.SQL(', ').join([
        sql.Identifier('id'),
        sql.Identifier('name'),
        sql.Identifier('email')
    ]),
    table=sql.Identifier('users'),
    condition=sql.SQL("{} > {}").format(
        sql.Identifier('age'),
        sql.Literal(25)
    )
)

print(query.as_string(conn))
# SELECT "id", "name", "email" FROM "users" WHERE "age" > 25

# Dynamic WHERE clauses
conditions = []
params = []

if name_filter:
    conditions.append(sql.SQL("{} ILIKE {}").format(
        sql.Identifier('name'),
        sql.Placeholder()
    ))
    params.append(f"%{name_filter}%")

if age_min:
    conditions.append(sql.SQL("{} >= {}").format(
        sql.Identifier('age'),
        sql.Placeholder()
    ))
    params.append(age_min)

if conditions:
    where_clause = sql.SQL(" WHERE ").join([
        sql.SQL(""),
        sql.SQL(" AND ").join(conditions)
    ])
else:
    where_clause = sql.SQL("")

query = sql.SQL("SELECT * FROM {table}{where}").format(
    table=sql.Identifier('users'),
    where=where_clause
)

# Execute with parameters
cur.execute(query, params)

SQL Constants

Pre-defined SQL constants for common values.

NULL = SQL("NULL")      # SQL NULL constant
DEFAULT = SQL("DEFAULT")  # SQL DEFAULT constant

Usage Example:

from psycopg2 import sql

# Using SQL constants
insert_query = sql.SQL("INSERT INTO {table} ({columns}) VALUES ({values})").format(
    table=sql.Identifier('users'),
    columns=sql.SQL(', ').join([
        sql.Identifier('name'),
        sql.Identifier('email'),
        sql.Identifier('created_at')
    ]),
    values=sql.SQL(', ').join([
        sql.Placeholder('name'),
        sql.Placeholder('email'), 
        sql.DEFAULT  # Use DEFAULT for created_at timestamp
    ])
)

cur.execute(insert_query, {'name': 'Alice', 'email': 'alice@example.com'})

Safe Dynamic Queries

Building queries dynamically while preventing SQL injection attacks.

Usage Example:

from psycopg2 import sql
import psycopg2

def build_select_query(table, columns=None, conditions=None, order_by=None, limit=None):
    """Build a safe SELECT query dynamically."""
    
    # Default to all columns
    if columns:
        fields = sql.SQL(', ').join(sql.Identifier(col) for col in columns)
    else:
        fields = sql.SQL('*')
    
    # Base query
    query_parts = [
        sql.SQL("SELECT"),
        fields,
        sql.SQL("FROM"),
        sql.Identifier(table)
    ]
    params = []
    
    # Add WHERE clause
    if conditions:
        where_conditions = []
        for column, operator, value in conditions:
            where_conditions.append(
                sql.SQL("{} {} {}").format(
                    sql.Identifier(column),
                    sql.SQL(operator),
                    sql.Placeholder()
                )
            )
            params.append(value)
        
        query_parts.extend([
            sql.SQL("WHERE"),
            sql.SQL(" AND ").join(where_conditions)
        ])
    
    # Add ORDER BY
    if order_by:
        order_fields = []
        for col_spec in order_by:
            if isinstance(col_spec, tuple):
                col, direction = col_spec
                order_fields.append(
                    sql.SQL("{} {}").format(
                        sql.Identifier(col),
                        sql.SQL(direction)
                    )
                )
            else:
                order_fields.append(sql.Identifier(col_spec))
        
        query_parts.extend([
            sql.SQL("ORDER BY"),
            sql.SQL(", ").join(order_fields)
        ])
    
    # Add LIMIT
    if limit:
        query_parts.extend([
            sql.SQL("LIMIT"),
            sql.Placeholder()
        ])
        params.append(limit)
    
    return sql.SQL(" ").join(query_parts), params

# Usage
conn = psycopg2.connect("host=localhost dbname=mydb user=myuser")

query, params = build_select_query(
    table='users',
    columns=['id', 'name', 'email'],
    conditions=[
        ('age', '>=', 18),
        ('status', '=', 'active')
    ],
    order_by=[('name', 'ASC'), ('id', 'DESC')],
    limit=50
)

with conn.cursor() as cur:
    cur.execute(query, params)
    results = cur.fetchall()

conn.close()

Table and Column Name Safety

Safe handling of dynamic table and column names that come from user input or configuration.

Usage Example:

from psycopg2 import sql

def safe_table_query(table_name, user_columns, user_input_table):
    """Safely handle user-provided table and column names."""
    
    # Validate table name (you should have your own validation)
    allowed_tables = ['users', 'products', 'orders']
    if user_input_table not in allowed_tables:
        raise ValueError(f"Table {user_input_table} not allowed")
    
    # Validate column names
    allowed_columns = ['id', 'name', 'email', 'created_at']
    for col in user_columns:
        if col not in allowed_columns:
            raise ValueError(f"Column {col} not allowed")
    
    # Build safe query
    query = sql.SQL("SELECT {columns} FROM {table}").format(
        columns=sql.SQL(', ').join(sql.Identifier(col) for col in user_columns),
        table=sql.Identifier(user_input_table)
    )
    
    return query

# Usage with validation
try:
    query = safe_table_query('users', ['id', 'name'], 'users')
    cur.execute(query)
    results = cur.fetchall()
except ValueError as e:
    print(f"Security error: {e}")

Advanced Composition Patterns

Complex query building patterns for advanced use cases.

Usage Example:

from psycopg2 import sql

def build_upsert_query(table, conflict_columns, update_columns, insert_columns=None):
    """Build PostgreSQL UPSERT query using ON CONFLICT."""
    
    if insert_columns is None:
        insert_columns = update_columns
    
    # Build INSERT part
    insert_query = sql.SQL("INSERT INTO {table} ({columns}) VALUES ({placeholders})").format(
        table=sql.Identifier(table),
        columns=sql.SQL(', ').join(sql.Identifier(col) for col in insert_columns),
        placeholders=sql.SQL(', ').join(sql.Placeholder() for _ in insert_columns)
    )
    
    # Build ON CONFLICT part
    conflict_clause = sql.SQL("ON CONFLICT ({conflict_cols}) DO UPDATE SET {updates}").format(
        conflict_cols=sql.SQL(', ').join(sql.Identifier(col) for col in conflict_columns),
        updates=sql.SQL(', ').join(
            sql.SQL("{col} = EXCLUDED.{col}").format(col=sql.Identifier(col))
            for col in update_columns
        )
    )
    
    # Combine
    return sql.SQL(" ").join([insert_query, conflict_clause])

# Usage
upsert_query = build_upsert_query(
    table='users',
    conflict_columns=['email'],
    update_columns=['name', 'updated_at'],
    insert_columns=['email', 'name', 'created_at', 'updated_at']
)

params = ['user@example.com', 'Updated Name', '2023-01-01', '2023-01-02']
cur.execute(upsert_query, params)

# Complex joins with dynamic conditions
def build_join_query(base_table, joins, conditions=None):
    """Build query with multiple joins."""
    
    query_parts = [
        sql.SQL("SELECT * FROM"),
        sql.Identifier(base_table)
    ]
    
    # Add joins
    for join_type, join_table, on_condition in joins:
        query_parts.extend([
            sql.SQL(join_type),
            sql.Identifier(join_table),
            sql.SQL("ON"),
            sql.SQL(on_condition)
        ])
    
    # Add WHERE conditions
    if conditions:
        query_parts.extend([
            sql.SQL("WHERE"),
            sql.SQL(" AND ").join(sql.SQL(cond) for cond in conditions)
        ])
    
    return sql.SQL(" ").join(query_parts)

# Usage
join_query = build_join_query(
    base_table='users',
    joins=[
        ('LEFT JOIN', 'profiles', 'users.id = profiles.user_id'),
        ('INNER JOIN', 'organizations', 'users.org_id = organizations.id')
    ],
    conditions=['users.active = true', 'organizations.status = \'active\'']
)

Types

Composable Base Types

class Composable:
    """Base class for all SQL composition objects."""
    
    def as_string(self, context) -> str:
        """Get SQL string representation."""
    
    def __add__(self, other: 'Composable') -> 'Composed':
        """Concatenate composables."""
    
    def __mul__(self, n: int) -> 'Composed':
        """Repeat composable n times."""
    
    def __eq__(self, other) -> bool:
        """Equality comparison."""

class Composed(Composable):
    """Sequence of multiple composables."""
    
    seq: list  # List of composable elements
    
    def join(self, joiner: Composable) -> 'Composed':
        """Join elements with separator."""

class SQL(Composable):
    """Raw SQL text."""
    
    string: str  # The SQL string
    
    def format(self, *args, **kwargs) -> 'Composed':
        """Format with placeholders."""
    
    def join(self, seq) -> 'Composed':
        """Join sequence with this SQL."""

class Identifier(Composable):
    """Quoted SQL identifier."""
    
    strings: tuple[str, ...]  # Identifier components
    string: str  # Single string (if only one component)

class Literal(Composable):
    """SQL literal value."""
    
    wrapped: Any  # The wrapped Python value

class Placeholder(Composable):
    """Parameter placeholder."""
    
    name: str | None  # Placeholder name (None for positional)

SQL Constants

NULL: SQL       # SQL("NULL")
DEFAULT: SQL    # SQL("DEFAULT")

Install with Tessl CLI

npx tessl i tessl/pypi-psycopg2

docs

advanced-cursors.md

batch-operations.md

connection-pooling.md

connections-cursors.md

tessl/pypi-psycopg2

sql-composition.md.css-3qkkll{font-size:var(--chakra-font-sizes-sm);font-weight:var(--chakra-font-weights-normal);color:var(--chakra-colors-gray-300);}docs/

SQL Composition

Capabilities

SQL Building Classes

SQL Constants

Safe Dynamic Queries

Table and Column Name Safety

Advanced Composition Patterns

Types

Composable Base Types

SQL Constants

sql-composition.mddocs/