tessl/pypi-psycopg2-binary

PostgreSQL database adapter for Python with thread-safe connection pooling and SQL operations

—

Pending

Overview

Eval results

Files

SQL Composition and Query Building

Name: tessl/pypi-psycopg2-binary
Author: tessl

Safe SQL query construction using composable objects for identifiers, literals, and SQL fragments. The sql module prevents SQL injection while enabling dynamic query building through a flexible composition system that properly handles identifiers, literals, and SQL fragments.

Capabilities

Composable Base Classes

Foundation classes for building SQL queries through composition with proper escaping and formatting.

class Composable:
    """Abstract base class for SQL composable objects."""
    
    def __init__(self, wrapped):
        """Initialize composable with wrapped content."""
    
    def as_string(self, context):
        """
        Return string representation for SQL context.
        
        Parameters:
        - context (connection/cursor): SQL context for rendering
        
        Returns:
        str: SQL string representation
        """
    
    def __add__(self, other):
        """
        Concatenate with another composable.
        
        Parameters:
        - other (Composable): Object to concatenate
        
        Returns:
        Composed: Combined composable object
        """
    
    def __mul__(self, n):
        """
        Repeat composable n times.
        
        Parameters:
        - n (int): Number of repetitions
        
        Returns:
        Composed: Repeated composable object
        """

class Composed(Composable):
    """Sequence of composable objects."""
    
    def __init__(self, seq):
        """
        Initialize from sequence of composables.
        
        Parameters:
        - seq (sequence): Sequence of Composable objects
        """
    
    @property
    def seq(self):
        """
        Access to composable sequence.
        
        Returns:
        list: List of composable objects
        """
    
    def join(self, joiner):
        """
        Join composables with separator.
        
        Parameters:
        - joiner (str/SQL): Separator string or SQL object
        
        Returns:
        Composed: Joined composable object
        """

SQL Fragments

Raw SQL string fragments that are not escaped, for static SQL content.

class SQL(Composable):
    """SQL string fragment (not escaped)."""
    
    def __init__(self, string):
        """
        Initialize SQL fragment.
        
        Parameters:
        - string (str): SQL string content
        
        Raises:
        TypeError: If string is not str type
        """
    
    @property
    def string(self):
        """
        Access to wrapped string.
        
        Returns:
        str: SQL string content
        """
    
    def format(self, *args, **kwargs):
        """
        Format SQL with composable arguments.
        
        Parameters:
        - *args: Positional composable arguments
        - **kwargs: Named composable arguments
        
        Returns:
        Composed: Formatted SQL object
        """
    
    def join(self, seq):
        """
        Join sequence with this SQL as separator.
        
        Parameters:
        - seq (iterable): Sequence of composables
        
        Returns:
        Composed: Joined composable object
        """

Usage examples:

from psycopg2 import sql

# Basic SQL fragments
select_sql = sql.SQL("SELECT * FROM users")
where_sql = sql.SQL("WHERE active = %s")

# Combining SQL fragments
query = select_sql + sql.SQL(" ") + where_sql
print(query.as_string(conn))  # "SELECT * FROM users WHERE active = %s"

# SQL formatting with placeholders
template = sql.SQL("SELECT {fields} FROM {table}")
query = template.format(
    fields=sql.SQL("id, name, email"),
    table=sql.SQL("users")
)

# Joining SQL fragments
fields = [sql.SQL("id"), sql.SQL("name"), sql.SQL("email")]
field_list = sql.SQL(", ").join(fields)

Identifiers

SQL identifiers (table names, column names) with proper quoting and escaping.

class Identifier(Composable):
    """SQL identifier with proper quoting."""
    
    def __init__(self, *strings):
        """
        Initialize identifier.
        
        Parameters:
        - *strings: Identifier components (e.g., schema, table, column)
        """
    
    @property
    def strings(self):
        """
        Access to identifier components.
        
        Returns:
        tuple: Identifier string components
        """

Usage examples:

from psycopg2 import sql

# Simple identifiers
table = sql.Identifier('users')
column = sql.Identifier('user_name')

# Qualified identifiers (schema.table, table.column)
qualified_table = sql.Identifier('public', 'users')
qualified_column = sql.Identifier('users', 'id')

# Safe dynamic table/column selection
def get_user_data(table_name, column_name):
    query = sql.SQL("SELECT {column} FROM {table}").format(
        column=sql.Identifier(column_name),
        table=sql.Identifier(table_name)
    )
    return query

# Usage - properly quoted regardless of input
query = get_user_data('user_profiles', 'first_name')
# Result: SELECT "first_name" FROM "user_profiles"

query = get_user_data('special-table', 'column with spaces')
# Result: SELECT "column with spaces" FROM "special-table"

Literals

SQL literal values with proper escaping and type conversion.

class Literal(Composable):
    """SQL literal value with proper escaping."""
    
    def __init__(self, wrapped):
        """
        Initialize literal value.
        
        Parameters:
        - wrapped: Python value to convert to SQL literal
        """
    
    @property
    def wrapped(self):
        """
        Access to wrapped value.
        
        Returns:
        Wrapped Python value
        """

Usage examples:

from psycopg2 import sql

# Basic literals
name_literal = sql.Literal('Alice')
age_literal = sql.Literal(30)
active_literal = sql.Literal(True)

# Dynamic WHERE clauses
def build_where_clause(conditions):
    clauses = []
    for column, value in conditions.items():
        clause = sql.SQL("{} = {}").format(
            sql.Identifier(column),
            sql.Literal(value)
        )
        clauses.append(clause)
    return sql.SQL(" AND ").join(clauses)

# Usage
conditions = {'name': 'Alice', 'age': 30, 'active': True}
where_clause = build_where_clause(conditions)
# Result: "name" = 'Alice' AND "age" = 30 AND "active" = true

query = sql.SQL("SELECT * FROM users WHERE {}").format(where_clause)

Placeholders

SQL parameter placeholders for prepared statement usage.

class Placeholder(Composable):
    """SQL parameter placeholder."""
    
    def __init__(self, name=None):
        """
        Initialize placeholder.
        
        Parameters:
        - name (str, optional): Named parameter name
        """
    
    @property
    def name(self):
        """
        Access to placeholder name.
        
        Returns:
        str/None: Parameter name or None for positional
        """

Usage examples:

from psycopg2 import sql

# Named placeholders
name_placeholder = sql.Placeholder('name')
age_placeholder = sql.Placeholder('age')

query = sql.SQL("SELECT * FROM users WHERE name = {} AND age > {}").format(
    name_placeholder, age_placeholder
)
# Result: "SELECT * FROM users WHERE name = %(name)s AND age > %(age)s"

# Execute with named parameters
cur.execute(query, {'name': 'Alice', 'age': 25})

# Positional placeholders
pos_placeholder = sql.Placeholder()
query = sql.SQL("INSERT INTO users (name, email) VALUES ({}, {})").format(
    pos_placeholder, pos_placeholder
)
# Result: "INSERT INTO users (name, email) VALUES (%s, %s)"

cur.execute(query, ('Alice', 'alice@example.com'))

Advanced Query Building

Complex query construction patterns using composition methods.

# Dynamic SELECT with optional fields
def build_select_query(table, fields=None, conditions=None, order_by=None):
    # Base query
    if fields:
        field_list = sql.SQL(", ").join(sql.Identifier(f) for f in fields)
    else:
        field_list = sql.SQL("*")
    
    query = sql.SQL("SELECT {} FROM {}").format(
        field_list, 
        sql.Identifier(table)
    )
    
    # Add WHERE clause
    if conditions:
        where_parts = []
        for column, value in conditions.items():
            where_parts.append(
                sql.SQL("{} = {}").format(
                    sql.Identifier(column),
                    sql.Placeholder(column)
                )
            )
        where_clause = sql.SQL(" AND ").join(where_parts)
        query = query + sql.SQL(" WHERE ") + where_clause
    
    # Add ORDER BY
    if order_by:
        order_fields = sql.SQL(", ").join(
            sql.Identifier(field) for field in order_by
        )
        query = query + sql.SQL(" ORDER BY ") + order_fields
    
    return query

# Usage
query = build_select_query(
    table='users',
    fields=['id', 'name', 'email'],
    conditions={'active': True, 'role': 'admin'},
    order_by=['name', 'created_at']
)

cur.execute(query, {'active': True, 'role': 'admin'})

Bulk Operations with SQL Composition

Using SQL composition with bulk operations for dynamic queries.

from psycopg2.extras import execute_values

# Dynamic bulk insert
def bulk_insert(table, columns, data):
    # Build column list
    column_list = sql.SQL(", ").join(sql.Identifier(col) for col in columns)
    
    # Build VALUES template  
    value_template = sql.SQL("({})").format(
        sql.SQL(", ").join(sql.Placeholder() for _ in columns)
    )
    
    # Complete query
    query = sql.SQL("INSERT INTO {} ({}) VALUES {}").format(
        sql.Identifier(table),
        column_list,
        value_template
    )
    
    # Execute bulk insert
    execute_values(cur, query, data, template=None, page_size=100)

# Usage
columns = ['name', 'email', 'age']
data = [
    ('Alice', 'alice@example.com', 30),
    ('Bob', 'bob@example.com', 25),
    ('Charlie', 'charlie@example.com', 35)
]

bulk_insert('users', columns, data)

SQL Injection Prevention

SQL composition provides strong protection against SQL injection attacks.

# UNSAFE - vulnerable to SQL injection  
user_input = "'; DROP TABLE users; --"
unsafe_query = f"SELECT * FROM users WHERE name = '{user_input}'"
# DON'T DO THIS!

# SAFE - using SQL composition
safe_query = sql.SQL("SELECT * FROM users WHERE name = {}").format(
    sql.Literal(user_input)
)
# Properly escaped: SELECT * FROM users WHERE name = '''; DROP TABLE users; --'

# SAFE - using placeholders (preferred for values)
param_query = sql.SQL("SELECT * FROM users WHERE name = {}").format(
    sql.Placeholder('name')
)
cur.execute(param_query, {'name': user_input})

# SAFE - identifiers properly quoted
table_name = "user-profiles"  # Contains special characters
safe_table_query = sql.SQL("SELECT * FROM {}").format(
    sql.Identifier(table_name)
)
# Result: SELECT * FROM "user-profiles"

Types

Composition Operators

# Concatenation
composable1 + composable2  # Returns Composed object

# Repetition  
composable * n  # Returns Composed with n repetitions

# Equality
composable1 == composable2  # True if same type and content

Context Objects

SqlContext = connection | cursor  # Valid contexts for as_string()

String Formatting

# Format placeholders
"{}"           # Positional placeholder
"{0}", "{1}"   # Numbered placeholders  
"{name}"       # Named placeholder

# Format arguments must be Composable objects
FormatArgs = Composable | tuple[Composable, ...] | dict[str, Composable]

Common Patterns

# Table and column name building
schema_table = sql.Identifier('schema_name', 'table_name')
table_column = sql.Identifier('table_name', 'column_name')

# IN clause construction
values = [sql.Literal(v) for v in [1, 2, 3, 4]]
in_clause = sql.SQL("IN ({})").format(sql.SQL(", ").join(values))

# Dynamic ORDER BY
order_fields = ['name', 'created_at']
order_clause = sql.SQL("ORDER BY {}").format(
    sql.SQL(", ").join(sql.Identifier(f) for f in order_fields)
)

# CASE expressions
case_expr = sql.SQL("""
    CASE 
        WHEN {} > {} THEN {}
        ELSE {}
    END
""").format(
    sql.Identifier('age'),
    sql.Literal(18),
    sql.Literal('adult'),
    sql.Literal('minor')
)

Install with Tessl CLI