0
# Safe SQL Composition
1

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

4
## Capabilities
5

6
### SQL String Composition
7

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

10
```python { .api }
11
class SQL:
12
    def __init__(self, template: str):
13
        """
14
        Create SQL template string.
15
        
16
        Args:
17
            template: SQL template with format placeholders like {table}, {field}
18
                     Must be a string literal for security
19
        """
20
        
21
    def format(self, *args, **kwargs) -> Composed:
22
        """
23
        Format SQL template with provided arguments.
24
        
25
        Args:
26
            *args: Positional format arguments
27
            **kwargs: Keyword format arguments
28
            
29
        Returns:
30
            Composed SQL object ready for execution
31
        """
32
        
33
    def join(self, seq: Iterable[Composable]) -> Composed:
34
        """
35
        Join sequence of composable objects with this SQL as separator.
36
        
37
        Args:
38
            seq: Sequence of Composable objects to join
39
            
40
        Returns:
41
            Composed SQL object with joined elements
42
        """
43
        
44
    def as_string(self, context: AdaptContext | None = None) -> str:
45
        """
46
        Convert to SQL string representation.
47
        
48
        Args:
49
            context: Adaptation context for custom formatting
50
            
51
        Returns:
52
            SQL string ready for database execution
53
        """
54
```
55

56
### Identifier Management
57

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

60
```python { .api }
61
class Identifier:
62
    def __init__(self, *names: str):
63
        """
64
        Create database identifier with automatic quoting.
65
        
66
        Args:
67
            *names: Identifier parts (e.g., schema, table, column)
68
        """
69
        
70
    def as_string(self, context: AdaptContext | None = None) -> str:
71
        """
72
        Convert to properly quoted identifier string.
73
        
74
        Returns:
75
            Quoted identifier suitable for SQL queries
76
        """
77
```
78

79
#### Usage Examples
80

81
```python
82
from psycopg import sql
83

84
# Simple identifier
85
table_name = sql.Identifier("users")
86
# Produces: "users"
87

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

92
# With special characters
93
special_table = sql.Identifier("user-data")
94
# Produces: "user-data"
95
```
96

97
### Literal Value Handling
98

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

101
```python { .api }
102
class Literal:
103
    def __init__(self, obj: Any):
104
        """
105
        Create SQL literal from Python object.
106
        
107
        Args:
108
            obj: Python object to convert to SQL literal
109
        """
110
        
111
    def as_string(self, context: AdaptContext | None = None) -> str:
112
        """
113
        Convert to properly escaped SQL literal string.
114
        
115
        Args:
116
            context: Adaptation context for custom type handling
117
            
118
        Returns:
119
            Escaped SQL literal string
120
        """
121
```
122

123
#### Usage Examples
124

125
```python
126
# String literal
127
name_literal = sql.Literal("John's Data")
128
# Produces: 'John''s Data'
129

130
# Number literal
131
age_literal = sql.Literal(25)
132
# Produces: 25
133

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

139
# None/NULL literal
140
null_literal = sql.Literal(None)
141
# Produces: NULL
142
```
143

144
### Parameter Placeholders
145

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

148
```python { .api }
149
class Placeholder:
150
    def __init__(self, name: str | None = None):
151
        """
152
        Create parameter placeholder.
153
        
154
        Args:
155
            name: Named placeholder identifier (None for positional)
156
        """
157
        
158
    def as_string(self, context: AdaptContext | None = None) -> str:
159
        """
160
        Convert to parameter placeholder string.
161
        
162
        Returns:
163
            Parameter placeholder (%s, %(name)s, etc.)
164
        """
165
```
166

167
#### Usage Examples
168

169
```python
170
# Positional placeholder
171
pos_placeholder = sql.Placeholder()
172
# Produces: %s
173

174
# Named placeholder  
175
named_placeholder = sql.Placeholder("user_id")
176
# Produces: %(user_id)s
177
```
178

179
### Composed SQL Objects
180

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

183
```python { .api }
184
class Composed:
185
    def __init__(self, seq: Iterable[Composable]):
186
        """
187
        Create composed SQL object from sequence of components.
188
        
189
        Args:
190
            seq: Sequence of Composable SQL components
191
        """
192
        
193
    def join(self, joiner: Composable) -> Composed:
194
        """
195
        Join components with specified joiner.
196
        
197
        Args:
198
            joiner: Composable object to use as separator
199
            
200
        Returns:
201
            New Composed object with joined components
202
        """
203
        
204
    def as_string(self, context: AdaptContext | None = None) -> str:
205
        """Convert all components to single SQL string"""
206
```
207

208
### Base Composable Interface
209

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

212
```python { .api }
213
from abc import ABC, abstractmethod
214

215
class Composable(ABC):
216
    @abstractmethod
217
    def as_string(self, context: AdaptContext | None = None) -> str:
218
        """Convert to SQL string representation"""
219
        
220
    def __add__(self, other: Composable) -> Composed:
221
        """Concatenate with another composable object (+)"""
222
        
223
    def __mul__(self, n: int) -> Composed:
224
        """Repeat this object n times (*)"""
225
        
226
    def __rmul__(self, n: int) -> Composed:
227
        """Repeat this object n times (n * obj)"""
228
```
229

230
### Utility Functions
231

232
Standalone functions for common SQL composition tasks.
233

234
```python { .api }
235
def quote(obj: Any, context: AdaptContext | None = None) -> str:
236
    """
237
    Convert Python object to quoted SQL string.
238
    
239
    Args:
240
        obj: Python object to quote
241
        context: Adaptation context for custom handling
242
        
243
    Returns:
244
        Quoted SQL string
245
        
246
    Note:
247
        This function is less efficient than using Literal objects
248
        as it doesn't cache adaptation rules.
249
    """
250
```
251

252
## Advanced Usage Examples
253

254
### Dynamic Query Building
255

256
```python
257
from psycopg import sql
258

259
# Build SELECT query dynamically
260
def build_select(table, columns, where_conditions=None):
261
    query = sql.SQL("SELECT {fields} FROM {table}").format(
262
        fields=sql.SQL(", ").join(sql.Identifier(col) for col in columns),
263
        table=sql.Identifier(table)
264
    )
265
    
266
    if where_conditions:
267
        where_parts = []
268
        for column, value in where_conditions.items():
269
            where_parts.append(
270
                sql.SQL("{field} = {value}").format(
271
                    field=sql.Identifier(column),
272
                    value=sql.Placeholder(column)
273
                )
274
            )
275
        
276
        query += sql.SQL(" WHERE ") + sql.SQL(" AND ").join(where_parts)
277
    
278
    return query
279

280
# Usage
281
query = build_select(
282
    "users", 
283
    ["id", "name", "email"],
284
    {"active": True, "age": 25}
285
)
286
# Produces: SELECT "id", "name", "email" FROM "users" WHERE "active" = %(active)s AND "age" = %(age)s
287

288
cursor.execute(query, {"active": True, "age": 25})
289
```
290

291
### Complex INSERT Statements
292

293
```python
294
# Dynamic INSERT with multiple rows
295
def build_insert(table, columns, num_rows):
296
    placeholders = sql.SQL("({})").format(
297
        sql.SQL(", ").join(sql.Placeholder() for _ in columns)
298
    )
299
    
300
    values_list = sql.SQL(", ").join(placeholders for _ in range(num_rows))
301
    
302
    query = sql.SQL("INSERT INTO {table} ({fields}) VALUES {values}").format(
303
        table=sql.Identifier(table),
304
        fields=sql.SQL(", ").join(sql.Identifier(col) for col in columns),
305
        values=values_list
306
    )
307
    return query
308

309
# Insert multiple users
310
columns = ["name", "email", "age"]
311
query = build_insert("users", columns, 3)
312
cursor.execute(query, ("Alice", "alice@example.com", 30, 
313
                      "Bob", "bob@example.com", 25,
314
                      "Charlie", "charlie@example.com", 35))
315
```
316

317
### UPDATE with Dynamic SET Clause
318

319
```python
320
def build_update(table, updates, where_conditions):
321
    set_parts = []
322
    for column, value in updates.items():
323
        set_parts.append(
324
            sql.SQL("{field} = {placeholder}").format(
325
                field=sql.Identifier(column),
326
                placeholder=sql.Placeholder(column)
327
            )
328
        )
329
    
330
    query = sql.SQL("UPDATE {table} SET {sets}").format(
331
        table=sql.Identifier(table),
332
        sets=sql.SQL(", ").join(set_parts)
333
    )
334
    
335
    if where_conditions:
336
        where_parts = []
337
        for column, value in where_conditions.items():
338
            where_key = f"where_{column}"
339
            where_parts.append(
340
                sql.SQL("{field} = {placeholder}").format(
341
                    field=sql.Identifier(column),
342
                    placeholder=sql.Placeholder(where_key)
343
                )
344
            )
345
        query += sql.SQL(" WHERE ") + sql.SQL(" AND ").join(where_parts)
346
    
347
    return query
348

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

354
params = {**updates, "where_id": 123}
355
cursor.execute(query, params)
356
```
357

358
### Schema and Table Management
359

360
```python
361
# Create table with dynamic schema
362
def create_table_query(schema, table, columns):
363
    """
364
    columns: dict of {column_name: column_definition}
365
    """
366
    table_ident = sql.Identifier(schema, table) if schema else sql.Identifier(table)
367
    
368
    column_defs = []
369
    for name, definition in columns.items():
370
        column_defs.append(
371
            sql.SQL("{name} {definition}").format(
372
                name=sql.Identifier(name),
373
                definition=sql.SQL(definition)
374
            )
375
        )
376
    
377
    query = sql.SQL("CREATE TABLE {table} ({columns})").format(
378
        table=table_ident,
379
        columns=sql.SQL(", ").join(column_defs)
380
    )
381
    return query
382

383
# Usage
384
columns = {
385
    "id": "SERIAL PRIMARY KEY",
386
    "name": "VARCHAR(100) NOT NULL",
387
    "email": "VARCHAR(255) UNIQUE",
388
    "created_at": "TIMESTAMP DEFAULT CURRENT_TIMESTAMP"
389
}
390

391
query = create_table_query("public", "users", columns)
392
cursor.execute(query)
393
```
394

395
### Handling Complex WHERE Clauses
396

397
```python
398
def build_complex_where(conditions):
399
    """
400
    conditions: dict with operators
401
    {
402
        "name": ("LIKE", "%john%"),
403
        "age": (">=", 18),
404
        "status": ("IN", ["active", "pending"])
405
    }
406
    """
407
    where_parts = []
408
    params = {}
409
    
410
    for i, (column, (operator, value)) in enumerate(conditions.items()):
411
        param_name = f"param_{i}"
412
        
413
        if operator.upper() == "IN":
414
            # Handle IN clause with list
415
            placeholders = sql.SQL("({})").format(
416
                sql.SQL(", ").join(sql.Placeholder(f"{param_name}_{j}") 
417
                                 for j in range(len(value)))
418
            )
419
            where_parts.append(
420
                sql.SQL("{field} IN {placeholders}").format(
421
                    field=sql.Identifier(column),
422
                    placeholders=placeholders
423
                )
424
            )
425
            for j, item in enumerate(value):
426
                params[f"{param_name}_{j}"] = item
427
        else:
428
            where_parts.append(
429
                sql.SQL("{field} {op} {placeholder}").format(
430
                    field=sql.Identifier(column),
431
                    op=sql.SQL(operator),
432
                    placeholder=sql.Placeholder(param_name)
433
                )
434
            )
435
            params[param_name] = value
436
    
437
    where_clause = sql.SQL(" AND ").join(where_parts)
438
    return where_clause, params
439

440
# Usage
441
conditions = {
442
    "name": ("LIKE", "%john%"),
443
    "age": (">=", 18),
444
    "status": ("IN", ["active", "pending"])
445
}
446

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

453
cursor.execute(query, params)
454
```
455

456
## Security Benefits
457

458
The SQL composition system provides several security advantages:
459

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

465
## Performance Considerations
466

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