0
# Row Factories and Result Processing
1
2
Flexible result formatting with built-in row factories for tuples, dictionaries, named tuples, and custom objects, plus protocols for creating custom result processors.
3
4
## Capabilities
5
6
### Built-in Row Factories
7
8
Pre-defined row factories for common result formatting needs, providing different ways to access query results.
9
10
```python { .api }
11
def tuple_row(cursor) -> RowMaker[tuple]:
12
"""
13
Default row factory returning tuples.
14
15
Args:
16
cursor: Database cursor with result metadata
17
18
Returns:
19
RowMaker that creates tuples from result rows
20
"""
21
22
def dict_row(cursor) -> RowMaker[dict]:
23
"""
24
Row factory returning dictionaries with column names as keys.
25
26
Args:
27
cursor: Database cursor with result metadata
28
29
Returns:
30
RowMaker that creates dicts with column names as keys
31
"""
32
33
def namedtuple_row(cursor) -> RowMaker[NamedTuple]:
34
"""
35
Row factory returning named tuples with column names as attributes.
36
37
Args:
38
cursor: Database cursor with result metadata
39
40
Returns:
41
RowMaker that creates named tuples with column attributes
42
"""
43
44
def scalar_row(cursor) -> RowMaker[Any]:
45
"""
46
Row factory for single-column results returning the value directly.
47
48
Args:
49
cursor: Database cursor with result metadata
50
51
Returns:
52
RowMaker that returns single column values
53
54
Raises:
55
ProgrammingError: If result has more than one column
56
"""
57
```
58
59
### Custom Object Row Factories
60
61
Create row factories that instantiate custom classes or call functions with result data.
62
63
```python { .api }
64
def class_row(cls: type) -> RowFactory:
65
"""
66
Row factory that creates instances of specified class.
67
68
Args:
69
cls: Class to instantiate with row values
70
71
Returns:
72
RowFactory that creates instances of cls
73
74
Note:
75
Class constructor must accept row values as positional arguments
76
"""
77
78
def args_row(func: Callable) -> RowFactory:
79
"""
80
Row factory that calls function with row values as positional arguments.
81
82
Args:
83
func: Function to call with row values
84
85
Returns:
86
RowFactory that calls func(*row_values)
87
"""
88
89
def kwargs_row(func: Callable) -> RowFactory:
90
"""
91
Row factory that calls function with column names as keyword arguments.
92
93
Args:
94
func: Function to call with column name/value pairs
95
96
Returns:
97
RowFactory that calls func(**{col_name: col_value, ...})
98
"""
99
```
100
101
### Row Factory Protocols
102
103
Type protocols defining the interfaces for row factories and row makers.
104
105
```python { .api }
106
from typing import Protocol, Sequence, Any, TypeVar
107
108
Row = TypeVar("Row", covariant=True)
109
110
class RowMaker(Protocol[Row]):
111
"""
112
Protocol for callable that converts sequence of values to row object.
113
114
The sequence contains database values already adapted to Python types.
115
Return value is the object your application receives.
116
"""
117
118
def __call__(self, values: Sequence[Any]) -> Row:
119
"""
120
Convert sequence of column values to row object.
121
122
Args:
123
values: Sequence of column values from database
124
125
Returns:
126
Row object of type Row
127
"""
128
129
class RowFactory(Protocol):
130
"""
131
Protocol for callable that creates RowMaker from cursor.
132
133
Row factories inspect cursor metadata to determine how to format results.
134
"""
135
136
def __call__(self, cursor) -> RowMaker:
137
"""
138
Create RowMaker for given cursor.
139
140
Args:
141
cursor: Database cursor with result metadata
142
143
Returns:
144
RowMaker configured for cursor's result structure
145
"""
146
```
147
148
### Row Types
149
150
Type definitions for common row formats returned by built-in factories.
151
152
```python { .api }
153
from typing import Any, Dict
154
155
# Type aliases for documentation
156
TupleRow = tuple[Any, ...]
157
DictRow = Dict[str, Any]
158
159
# Row type is parameterized for custom row factories
160
Row = TypeVar("Row", covariant=True, default=TupleRow)
161
```
162
163
## Usage Examples
164
165
### Basic Row Factory Usage
166
167
```python
168
from psycopg import rows
169
170
# Default tuple rows
171
with conn.cursor() as cur:
172
cur.execute("SELECT id, name, email FROM users")
173
for row in cur:
174
id, name, email = row # Tuple unpacking
175
print(f"User {id}: {name} <{email}>")
176
177
# Dictionary rows
178
with conn.cursor(row_factory=rows.dict_row) as cur:
179
cur.execute("SELECT id, name, email FROM users")
180
for row in cur:
181
print(f"User {row['id']}: {row['name']} <{row['email']}>")
182
183
# Named tuple rows
184
with conn.cursor(row_factory=rows.namedtuple_row) as cur:
185
cur.execute("SELECT id, name, email FROM users")
186
for row in cur:
187
print(f"User {row.id}: {row.name} <{row.email}>")
188
189
# Scalar results
190
with conn.cursor(row_factory=rows.scalar_row) as cur:
191
cur.execute("SELECT COUNT(*) FROM users")
192
count = cur.fetchone() # Returns integer directly, not tuple
193
print(f"Total users: {count}")
194
```
195
196
### Custom Class Integration
197
198
```python
199
from dataclasses import dataclass
200
from psycopg import rows
201
202
@dataclass
203
class User:
204
id: int
205
name: str
206
email: str
207
created_at: datetime
208
209
# Using class_row factory
210
def get_users():
211
with conn.cursor(row_factory=rows.class_row(User)) as cur:
212
cur.execute("SELECT id, name, email, created_at FROM users")
213
return cur.fetchall() # Returns list of User objects
214
215
users = get_users()
216
for user in users:
217
print(f"{user.name} ({user.email}) created on {user.created_at}")
218
```
219
220
### Function-Based Processing
221
222
```python
223
# Process results with function
224
def process_user_data(id, name, email, created_at):
225
return {
226
'user_id': id,
227
'display_name': name.title(),
228
'contact': email.lower(),
229
'age_days': (datetime.now() - created_at).days
230
}
231
232
# Using args_row factory
233
with conn.cursor(row_factory=rows.args_row(process_user_data)) as cur:
234
cur.execute("SELECT id, name, email, created_at FROM users")
235
processed_users = cur.fetchall()
236
237
# Using kwargs_row factory
238
def format_user(**kwargs):
239
return f"{kwargs['name']} <{kwargs['email']}> (ID: {kwargs['id']})"
240
241
with conn.cursor(row_factory=rows.kwargs_row(format_user)) as cur:
242
cur.execute("SELECT id, name, email FROM users")
243
formatted = cur.fetchall() # Returns list of formatted strings
244
```
245
246
### Custom Row Factory Creation
247
248
```python
249
from psycopg.rows import RowFactory, RowMaker
250
from typing import Any, Sequence
251
252
def json_row(cursor) -> RowMaker[str]:
253
"""Row factory that returns each row as JSON string"""
254
import json
255
256
# Get column names from cursor
257
if cursor.description is None:
258
raise ProgrammingError("No result to format")
259
260
column_names = [col.name for col in cursor.description]
261
262
def make_json_row(values: Sequence[Any]) -> str:
263
row_dict = dict(zip(column_names, values))
264
return json.dumps(row_dict, default=str)
265
266
return make_json_row
267
268
# Usage
269
with conn.cursor(row_factory=json_row) as cur:
270
cur.execute("SELECT id, name, email FROM users LIMIT 1")
271
json_result = cur.fetchone()
272
print(json_result) # {"id": 1, "name": "Alice", "email": "alice@example.com"}
273
```
274
275
### Row Factory with Validation
276
277
```python
278
from typing import Any, Sequence
279
import re
280
281
def validated_email_row(cursor) -> RowMaker[dict]:
282
"""Row factory that validates email format"""
283
284
if cursor.description is None:
285
raise ProgrammingError("No result to format")
286
287
column_names = [col.name for col in cursor.description]
288
email_pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
289
290
def make_validated_row(values: Sequence[Any]) -> dict:
291
row = dict(zip(column_names, values))
292
293
# Validate email if present
294
if 'email' in row and row['email']:
295
if not email_pattern.match(row['email']):
296
row['email_valid'] = False
297
row['email_error'] = 'Invalid email format'
298
else:
299
row['email_valid'] = True
300
row['email_error'] = None
301
302
return row
303
304
return make_validated_row
305
306
# Usage
307
with conn.cursor(row_factory=validated_email_row) as cur:
308
cur.execute("SELECT id, name, email FROM users")
309
for row in cur:
310
if not row.get('email_valid', True):
311
print(f"Warning: {row['name']} has invalid email: {row['email']}")
312
```
313
314
### Conditional Row Factories
315
316
```python
317
def smart_row_factory(cursor) -> RowMaker:
318
"""Choose row factory based on result structure"""
319
320
if cursor.description is None:
321
raise ProgrammingError("No result to format")
322
323
column_count = len(cursor.description)
324
325
if column_count == 1:
326
# Single column - return scalar
327
return rows.scalar_row(cursor)
328
elif column_count <= 3:
329
# Few columns - use named tuple for attribute access
330
return rows.namedtuple_row(cursor)
331
else:
332
# Many columns - use dictionary for flexible access
333
return rows.dict_row(cursor)
334
335
# Usage
336
with conn.cursor(row_factory=smart_row_factory) as cur:
337
# Single column query -> scalar result
338
cur.execute("SELECT COUNT(*) FROM users")
339
count = cur.fetchone() # Returns int
340
341
# Two column query -> named tuple
342
cur.execute("SELECT name, email FROM users LIMIT 1")
343
user = cur.fetchone() # Returns named tuple with .name, .email
344
345
# Many columns -> dictionary
346
cur.execute("SELECT * FROM users LIMIT 1")
347
full_user = cur.fetchone() # Returns dict with all columns
348
```
349
350
### Performance-Optimized Row Factory
351
352
```python
353
def fast_dict_row(cursor) -> RowMaker[dict]:
354
"""Optimized dictionary row factory for performance-critical code"""
355
356
if cursor.description is None:
357
raise ProgrammingError("No result to format")
358
359
# Pre-compute column names to avoid repeated attribute access
360
column_names = tuple(col.name for col in cursor.description)
361
362
# Use dict constructor with zip for better performance than dict comprehension
363
def make_fast_dict(values: Sequence[Any]) -> dict:
364
return dict(zip(column_names, values))
365
366
return make_fast_dict
367
```
368
369
### Row Factory for Complex Aggregations
370
371
```python
372
def grouped_row_factory(group_by_column: str):
373
"""Row factory that groups results by specified column"""
374
375
def factory(cursor) -> RowMaker[dict]:
376
if cursor.description is None:
377
raise ProgrammingError("No result to format")
378
379
column_names = [col.name for col in cursor.description]
380
381
if group_by_column not in column_names:
382
raise ValueError(f"Group column '{group_by_column}' not in result")
383
384
group_index = column_names.index(group_by_column)
385
results_by_group = {}
386
387
def make_grouped_row(values: Sequence[Any]) -> dict:
388
group_value = values[group_index]
389
row_dict = dict(zip(column_names, values))
390
391
if group_value not in results_by_group:
392
results_by_group[group_value] = []
393
394
results_by_group[group_value].append(row_dict)
395
396
# Return current group state
397
return {
398
'current_row': row_dict,
399
'group_value': group_value,
400
'group_data': results_by_group[group_value]
401
}
402
403
return make_grouped_row
404
405
return factory
406
407
# Usage
408
with conn.cursor(row_factory=grouped_row_factory('department')) as cur:
409
cur.execute("SELECT name, department, salary FROM employees ORDER BY department")
410
for row in cur:
411
print(f"Employee: {row['current_row']['name']}")
412
print(f"Department: {row['group_value']}")
413
print(f"Colleagues in dept: {len(row['group_data'])}")
414
```
415
416
## Setting Row Factories
417
418
Row factories can be set at different levels:
419
420
### Connection Level
421
422
```python
423
# Set default row factory for all cursors on connection
424
conn.row_factory = rows.dict_row
425
426
with conn.cursor() as cur:
427
cur.execute("SELECT * FROM users")
428
# Results are dictionaries
429
```
430
431
### Cursor Level
432
433
```python
434
# Set row factory for specific cursor
435
with conn.cursor(row_factory=rows.namedtuple_row) as cur:
436
cur.execute("SELECT * FROM users")
437
# Results are named tuples
438
439
# Change row factory on existing cursor
440
cur.row_factory = rows.dict_row
441
cur.execute("SELECT * FROM products")
442
# Results are now dictionaries
443
```
444
445
### Per-Query Basis
446
447
```python
448
# Different row factories for different queries on same cursor
449
with conn.cursor() as cur:
450
# Scalar results for aggregates
451
cur.row_factory = rows.scalar_row
452
cur.execute("SELECT COUNT(*) FROM users")
453
count = cur.fetchone()
454
455
# Dictionary results for detailed data
456
cur.row_factory = rows.dict_row
457
cur.execute("SELECT * FROM users LIMIT 10")
458
users = cur.fetchall()
459
```
460
461
## Type Hints and Integration
462
463
```python
464
from typing import TypeVar, Generic
465
from psycopg.rows import RowMaker
466
467
T = TypeVar('T')
468
469
class TypedCursor(Generic[T]):
470
"""Type-safe cursor wrapper"""
471
472
def __init__(self, cursor, row_factory: RowMaker[T]):
473
self.cursor = cursor
474
self.cursor.row_factory = row_factory
475
476
def fetchone(self) -> T | None:
477
return self.cursor.fetchone()
478
479
def fetchall(self) -> list[T]:
480
return self.cursor.fetchall()
481
482
# Usage with type safety
483
@dataclass
484
class User:
485
id: int
486
name: str
487
email: str
488
489
typed_cursor = TypedCursor(conn.cursor(), rows.class_row(User))
490
users: list[User] = typed_cursor.fetchall() # Type checker knows this is list[User]
491
```