CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-psycopg

PostgreSQL database adapter for Python

—

Pending
Overview
Eval results
Files

sql-composition.mddocs/

Safe SQL Composition

SQL query building with automatic escaping, identifier quoting, and parameter placeholder management to prevent SQL injection while maintaining readability and flexibility.

Capabilities

SQL String Composition

Build complex SQL queries safely using template strings with format placeholders and automatic escaping.

class SQL:
    def __init__(self, template: str):
        """
        Create SQL template string.
        
        Args:
            template: SQL template with format placeholders like {table}, {field}
                     Must be a string literal for security
        """
        
    def format(self, *args, **kwargs) -> Composed:
        """
        Format SQL template with provided arguments.
        
        Args:
            *args: Positional format arguments
            **kwargs: Keyword format arguments
            
        Returns:
            Composed SQL object ready for execution
        """
        
    def join(self, seq: Iterable[Composable]) -> Composed:
        """
        Join sequence of composable objects with this SQL as separator.
        
        Args:
            seq: Sequence of Composable objects to join
            
        Returns:
            Composed SQL object with joined elements
        """
        
    def as_string(self, context: AdaptContext | None = None) -> str:
        """
        Convert to SQL string representation.
        
        Args:
            context: Adaptation context for custom formatting
            
        Returns:
            SQL string ready for database execution
        """

Identifier Management

Safely handle database identifiers (table names, column names, schema names) with automatic quoting and validation.

class Identifier:
    def __init__(self, *names: str):
        """
        Create database identifier with automatic quoting.
        
        Args:
            *names: Identifier parts (e.g., schema, table, column)
        """
        
    def as_string(self, context: AdaptContext | None = None) -> str:
        """
        Convert to properly quoted identifier string.
        
        Returns:
            Quoted identifier suitable for SQL queries
        """

Usage Examples

from psycopg import sql

# Simple identifier
table_name = sql.Identifier("users")
# Produces: "users"

# Qualified identifier
qualified = sql.Identifier("public", "users", "name")  
# Produces: "public"."users"."name"

# With special characters
special_table = sql.Identifier("user-data")
# Produces: "user-data"

Literal Value Handling

Convert Python objects to properly escaped SQL literals with automatic type detection and formatting.

class Literal:
    def __init__(self, obj: Any):
        """
        Create SQL literal from Python object.
        
        Args:
            obj: Python object to convert to SQL literal
        """
        
    def as_string(self, context: AdaptContext | None = None) -> str:
        """
        Convert to properly escaped SQL literal string.
        
        Args:
            context: Adaptation context for custom type handling
            
        Returns:
            Escaped SQL literal string
        """

Usage Examples

# String literal
name_literal = sql.Literal("John's Data")
# Produces: 'John''s Data'

# Number literal
age_literal = sql.Literal(25)
# Produces: 25

# Date literal
import datetime
date_literal = sql.Literal(datetime.date(2023, 12, 25))
# Produces: '2023-12-25'::date

# None/NULL literal
null_literal = sql.Literal(None)
# Produces: NULL

Parameter Placeholders

Create named and positional parameter placeholders for prepared statements and dynamic queries.

class Placeholder:
    def __init__(self, name: str | None = None):
        """
        Create parameter placeholder.
        
        Args:
            name: Named placeholder identifier (None for positional)
        """
        
    def as_string(self, context: AdaptContext | None = None) -> str:
        """
        Convert to parameter placeholder string.
        
        Returns:
            Parameter placeholder (%s, %(name)s, etc.)
        """

Usage Examples

# Positional placeholder
pos_placeholder = sql.Placeholder()
# Produces: %s

# Named placeholder  
named_placeholder = sql.Placeholder("user_id")
# Produces: %(user_id)s

Composed SQL Objects

Container for multiple SQL components that can be combined and manipulated as a single unit.

class Composed:
    def __init__(self, seq: Iterable[Composable]):
        """
        Create composed SQL object from sequence of components.
        
        Args:
            seq: Sequence of Composable SQL components
        """
        
    def join(self, joiner: Composable) -> Composed:
        """
        Join components with specified joiner.
        
        Args:
            joiner: Composable object to use as separator
            
        Returns:
            New Composed object with joined components
        """
        
    def as_string(self, context: AdaptContext | None = None) -> str:
        """Convert all components to single SQL string"""

Base Composable Interface

Abstract base class providing common functionality for all SQL composition objects.

from abc import ABC, abstractmethod

class Composable(ABC):
    @abstractmethod
    def as_string(self, context: AdaptContext | None = None) -> str:
        """Convert to SQL string representation"""
        
    def __add__(self, other: Composable) -> Composed:
        """Concatenate with another composable object (+)"""
        
    def __mul__(self, n: int) -> Composed:
        """Repeat this object n times (*)"""
        
    def __rmul__(self, n: int) -> Composed:
        """Repeat this object n times (n * obj)"""

Utility Functions

Standalone functions for common SQL composition tasks.

def quote(obj: Any, context: AdaptContext | None = None) -> str:
    """
    Convert Python object to quoted SQL string.
    
    Args:
        obj: Python object to quote
        context: Adaptation context for custom handling
        
    Returns:
        Quoted SQL string
        
    Note:
        This function is less efficient than using Literal objects
        as it doesn't cache adaptation rules.
    """

Advanced Usage Examples

Dynamic Query Building

from psycopg import sql

# Build SELECT query dynamically
def build_select(table, columns, where_conditions=None):
    query = sql.SQL("SELECT {fields} FROM {table}").format(
        fields=sql.SQL(", ").join(sql.Identifier(col) for col in columns),
        table=sql.Identifier(table)
    )
    
    if where_conditions:
        where_parts = []
        for column, value in where_conditions.items():
            where_parts.append(
                sql.SQL("{field} = {value}").format(
                    field=sql.Identifier(column),
                    value=sql.Placeholder(column)
                )
            )
        
        query += sql.SQL(" WHERE ") + sql.SQL(" AND ").join(where_parts)
    
    return query

# Usage
query = build_select(
    "users", 
    ["id", "name", "email"],
    {"active": True, "age": 25}
)
# Produces: SELECT "id", "name", "email" FROM "users" WHERE "active" = %(active)s AND "age" = %(age)s

cursor.execute(query, {"active": True, "age": 25})

Complex INSERT Statements

# Dynamic INSERT with multiple rows
def build_insert(table, columns, num_rows):
    placeholders = sql.SQL("({})").format(
        sql.SQL(", ").join(sql.Placeholder() for _ in columns)
    )
    
    values_list = sql.SQL(", ").join(placeholders for _ in range(num_rows))
    
    query = sql.SQL("INSERT INTO {table} ({fields}) VALUES {values}").format(
        table=sql.Identifier(table),
        fields=sql.SQL(", ").join(sql.Identifier(col) for col in columns),
        values=values_list
    )
    return query

# Insert multiple users
columns = ["name", "email", "age"]
query = build_insert("users", columns, 3)
cursor.execute(query, ("Alice", "alice@example.com", 30, 
                      "Bob", "bob@example.com", 25,
                      "Charlie", "charlie@example.com", 35))

UPDATE with Dynamic SET Clause

def build_update(table, updates, where_conditions):
    set_parts = []
    for column, value in updates.items():
        set_parts.append(
            sql.SQL("{field} = {placeholder}").format(
                field=sql.Identifier(column),
                placeholder=sql.Placeholder(column)
            )
        )
    
    query = sql.SQL("UPDATE {table} SET {sets}").format(
        table=sql.Identifier(table),
        sets=sql.SQL(", ").join(set_parts)
    )
    
    if where_conditions:
        where_parts = []
        for column, value in where_conditions.items():
            where_key = f"where_{column}"
            where_parts.append(
                sql.SQL("{field} = {placeholder}").format(
                    field=sql.Identifier(column),
                    placeholder=sql.Placeholder(where_key)
                )
            )
        query += sql.SQL(" WHERE ") + sql.SQL(" AND ").join(where_parts)
    
    return query

# Update user data
updates = {"name": "John Smith", "email": "john.smith@example.com"}
where = {"id": 123}
query = build_update("users", updates, where)

params = {**updates, "where_id": 123}
cursor.execute(query, params)

Schema and Table Management

# Create table with dynamic schema
def create_table_query(schema, table, columns):
    """
    columns: dict of {column_name: column_definition}
    """
    table_ident = sql.Identifier(schema, table) if schema else sql.Identifier(table)
    
    column_defs = []
    for name, definition in columns.items():
        column_defs.append(
            sql.SQL("{name} {definition}").format(
                name=sql.Identifier(name),
                definition=sql.SQL(definition)
            )
        )
    
    query = sql.SQL("CREATE TABLE {table} ({columns})").format(
        table=table_ident,
        columns=sql.SQL(", ").join(column_defs)
    )
    return query

# Usage
columns = {
    "id": "SERIAL PRIMARY KEY",
    "name": "VARCHAR(100) NOT NULL",
    "email": "VARCHAR(255) UNIQUE",
    "created_at": "TIMESTAMP DEFAULT CURRENT_TIMESTAMP"
}

query = create_table_query("public", "users", columns)
cursor.execute(query)

Handling Complex WHERE Clauses

def build_complex_where(conditions):
    """
    conditions: dict with operators
    {
        "name": ("LIKE", "%john%"),
        "age": (">=", 18),
        "status": ("IN", ["active", "pending"])
    }
    """
    where_parts = []
    params = {}
    
    for i, (column, (operator, value)) in enumerate(conditions.items()):
        param_name = f"param_{i}"
        
        if operator.upper() == "IN":
            # Handle IN clause with list
            placeholders = sql.SQL("({})").format(
                sql.SQL(", ").join(sql.Placeholder(f"{param_name}_{j}") 
                                 for j in range(len(value)))
            )
            where_parts.append(
                sql.SQL("{field} IN {placeholders}").format(
                    field=sql.Identifier(column),
                    placeholders=placeholders
                )
            )
            for j, item in enumerate(value):
                params[f"{param_name}_{j}"] = item
        else:
            where_parts.append(
                sql.SQL("{field} {op} {placeholder}").format(
                    field=sql.Identifier(column),
                    op=sql.SQL(operator),
                    placeholder=sql.Placeholder(param_name)
                )
            )
            params[param_name] = value
    
    where_clause = sql.SQL(" AND ").join(where_parts)
    return where_clause, params

# Usage
conditions = {
    "name": ("LIKE", "%john%"),
    "age": (">=", 18),
    "status": ("IN", ["active", "pending"])
}

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

cursor.execute(query, params)

Security Benefits

The SQL composition system provides several security advantages:

  1. SQL Injection Prevention: All user input is properly escaped and parameterized
  2. Identifier Safety: Table and column names are automatically quoted to handle special characters
  3. Type Safety: Automatic type detection and proper formatting for different data types
  4. Validation: Invalid SQL constructs are caught at composition time rather than execution time

Performance Considerations

  • SQL composition objects are lightweight and can be reused
  • Complex queries should be built once and reused with different parameters
  • Use sql.Placeholder() for repeated parameter patterns
  • Consider caching composed queries for frequently used patterns

Install with Tessl CLI

npx tessl i tessl/pypi-psycopg

docs

advanced-operations.md

connections.md

cursors.md

error-handling.md

index.md

row-factories.md

sql-composition.md

type-system.md

tile.json