0
# Development Tools
1

2
Optional features for development including query recording, modification tracking, and Flask shell integration for enhanced debugging capabilities.
3

4
## Capabilities
5

6
### Query Recording
7

8
Tools for recording and analyzing database queries during development and testing.
9

10
```python { .api }
11
def get_recorded_queries() -> list[_QueryInfo]:
12
    """
13
    Get list of recorded query information for current session.
14
    
15
    Only available when SQLALCHEMY_RECORD_QUERIES config is enabled.
16
    Returns empty list if recording is disabled or no queries recorded.
17
    
18
    Returns:
19
    List of _QueryInfo objects with query details and timing
20
    """
21

22
@dataclasses.dataclass
23
class _QueryInfo:
24
    """
25
    Information about an executed database query.
26
    
27
    Contains timing, statement, and location information for debugging
28
    slow queries and understanding application database usage patterns.
29
    """
30
    
31
    statement: str | None  # SQL statement with parameter placeholders
32
    parameters: Any  # Parameters sent with the SQL statement
33
    start_time: float  # Query start timestamp
34
    end_time: float  # Query completion timestamp  
35
    location: str  # Code location where query was executed
36
    
37
    @property
38
    def duration(self) -> float:
39
        """Query execution duration in seconds."""
40
```
41

42
### Model Change Tracking
43

44
Signals for tracking model changes during database sessions.
45

46
```python { .api }
47
from flask.signals import Namespace
48

49
_signals: Namespace
50

51
models_committed: Signal
52
"""
53
Blinker signal sent after session commit with model changes.
54

55
The sender is the Flask application. The receiver gets a 'changes' 
56
argument with list of (instance, operation) tuples where operation
57
is 'insert', 'update', or 'delete'.
58
"""
59

60
before_models_committed: Signal  
61
"""
62
Blinker signal sent before session commit with model changes.
63

64
Works exactly like models_committed but fires before the commit
65
takes place, allowing for last-minute modifications or validations.
66
"""
67
```
68

69
### Flask Shell Integration
70

71
Functions for adding database context to Flask shell sessions.
72

73
```python { .api }
74
def add_models_to_shell() -> dict[str, Any]:
75
    """
76
    Add db instance and all model classes to Flask shell context.
77
    
78
    Automatically registered as shell context processor when
79
    add_models_to_shell=True in SQLAlchemy initialization.
80
    
81
    Returns:
82
    Dictionary mapping model class names to classes plus 'db' key
83
    """
84
```
85

86
## Usage Examples
87

88
### Enabling Query Recording
89

90
```python
91
# Enable query recording in development
92
app.config['SQLALCHEMY_RECORD_QUERIES'] = True
93

94
db = SQLAlchemy(app)
95

96
@app.after_request
97
def log_queries(response):
98
    if app.debug:
99
        queries = get_recorded_queries()
100
        for query in queries:
101
            app.logger.debug(
102
                f"Query: {query.statement} "
103
                f"Duration: {query.duration:.3f}s "
104
                f"Location: {query.location}"
105
            )
106
    return response
107
```
108

109
### Query Performance Analysis
110

111
```python
112
@app.route('/debug/queries')
113
def debug_queries():
114
    """Debug endpoint to analyze recent queries."""
115
    if not app.debug:
116
        abort(404)
117
    
118
    queries = get_recorded_queries()
119
    
120
    # Find slow queries
121
    slow_queries = [q for q in queries if q.duration > 0.1]
122
    
123
    # Query statistics
124
    total_time = sum(q.duration for q in queries)
125
    avg_time = total_time / len(queries) if queries else 0
126
    
127
    return {
128
        'total_queries': len(queries),
129
        'total_time': total_time,
130
        'average_time': avg_time,
131
        'slow_queries': len(slow_queries),
132
        'queries': [
133
            {
134
                'statement': q.statement,
135
                'duration': q.duration,
136
                'location': q.location,
137
                'parameters': str(q.parameters)
138
            }
139
            for q in queries
140
        ]
141
    }
142
```
143

144
### Model Change Tracking
145

146
```python
147
# Enable modification tracking
148
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
149

150
db = SQLAlchemy(app)
151

152
@models_committed.connect
153
def log_model_changes(sender, changes):
154
    """Log all model changes after commit."""
155
    for instance, operation in changes:
156
        app.logger.info(
157
            f"{operation.title()}: {instance.__class__.__name__} "
158
            f"ID={getattr(instance, 'id', 'unknown')}"
159
        )
160

161
@before_models_committed.connect  
162
def validate_changes(sender, changes):
163
    """Perform validation before committing changes."""
164
    for instance, operation in changes:
165
        if operation == 'delete' and hasattr(instance, 'protected'):
166
            if instance.protected:
167
                raise ValueError(f"Cannot delete protected {instance}")
168
```
169

170
### Custom Change Tracking
171

172
```python
173
class AuditMixin:
174
    """Mixin to add audit fields to models."""
175
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
176
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
177
    created_by = db.Column(db.String(100))
178
    updated_by = db.Column(db.String(100))
179

180
class User(AuditMixin, db.Model):
181
    id = db.Column(db.Integer, primary_key=True)
182
    username = db.Column(db.String(80), unique=True)
183

184
@before_models_committed.connect
185
def set_audit_fields(sender, changes):
186
    """Automatically set audit fields on model changes."""
187
    current_user = get_current_user()  # Your auth function
188
    
189
    for instance, operation in changes:
190
        if isinstance(instance, AuditMixin):
191
            if operation == 'insert':
192
                instance.created_by = current_user.username
193
            elif operation == 'update':
194
                instance.updated_by = current_user.username
195
```
196

197
### Flask Shell Integration
198

199
```python
200
# Automatic shell integration (enabled by default)
201
db = SQLAlchemy(app, add_models_to_shell=True)
202

203
# Manual shell integration
204
def make_shell_context():
205
    """Add custom items to shell context."""
206
    context = add_models_to_shell()  # Get db and models
207
    context.update({
208
        'datetime': datetime,
209
        'timedelta': timedelta,
210
    })
211
    return context
212

213
app.shell_context_processor(make_shell_context)
214
```
215

216
Using the enhanced shell:
217

218
```bash
219
$ flask shell
220
>>> db
221
<SQLAlchemy postgresql://...>
222
>>> User
223
<class 'app.models.User'>
224
>>> users = User.query.all()
225
>>> len(users)
226
42
227
```
228

229
### Development Middleware
230

231
```python
232
class QueryAnalysisMiddleware:
233
    """WSGI middleware to analyze database queries."""
234
    
235
    def __init__(self, app):
236
        self.app = app
237
    
238
    def __call__(self, environ, start_response):
239
        def new_start_response(status, headers):
240
            # Log query stats after request
241
            queries = get_recorded_queries()
242
            if queries:
243
                total_time = sum(q.duration for q in queries)
244
                print(f"Request used {len(queries)} queries in {total_time:.3f}s")
245
            return start_response(status, headers)
246
        
247
        return self.app(environ, new_start_response)
248

249
# Apply middleware in development
250
if app.debug:
251
    app.wsgi_app = QueryAnalysisMiddleware(app.wsgi_app)
252
```
253

254
### Query Profiling
255

256
```python
257
import cProfile
258
import pstats
259
from io import StringIO
260

261
@app.route('/profile/<path:endpoint>')
262
def profile_endpoint(endpoint):
263
    """Profile database queries for a specific endpoint."""
264
    if not app.debug:
265
        abort(404)
266
    
267
    # Clear previous queries
268
    g._sqlalchemy_queries = []
269
    
270
    # Profile the endpoint
271
    pr = cProfile.Profile()
272
    pr.enable()
273
    
274
    # Make request to endpoint
275
    with app.test_client() as client:
276
        response = client.get(f'/{endpoint}')
277
    
278
    pr.disable()
279
    
280
    # Get query information
281
    queries = get_recorded_queries()
282
    
283
    # Get profiling stats
284
    s = StringIO()
285
    ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
286
    ps.print_stats()
287
    
288
    return {
289
        'endpoint': endpoint,
290
        'status_code': response.status_code,
291
        'query_count': len(queries),
292
        'total_query_time': sum(q.duration for q in queries),
293
        'queries': [
294
            {
295
                'statement': q.statement,
296
                'duration': q.duration,
297
                'location': q.location
298
            }
299
            for q in queries
300
        ],
301
        'profile_stats': s.getvalue()
302
    }
303
```
304

305
### Testing with Query Recording
306

307
```python
308
def test_user_list_queries(client, app):
309
    """Test that user list endpoint doesn't have N+1 queries."""
310
    with app.app_context():
311
        # Clear any existing queries
312
        g._sqlalchemy_queries = []
313
        
314
        # Make request
315
        response = client.get('/users')
316
        
317
        # Check query count
318
        queries = get_recorded_queries()
319
        assert len(queries) <= 2  # Should be 1-2 queries max
320
        
321
        # Check for N+1 patterns
322
        select_queries = [q for q in queries if 'SELECT' in q.statement]
323
        assert len(select_queries) <= 1, "Possible N+1 query detected"
324
```
325

326
## Configuration
327

328
### Query Recording Configuration
329

330
```python
331
# Enable query recording (disabled by default)
332
SQLALCHEMY_RECORD_QUERIES = True
333

334
# Only record in development
335
SQLALCHEMY_RECORD_QUERIES = app.debug
336
```
337

338
### Modification Tracking Configuration  
339

340
```python
341
# Enable modification tracking (disabled by default)
342
SQLALCHEMY_TRACK_MODIFICATIONS = True
343

344
# Can be expensive in production, consider carefully
345
SQLALCHEMY_TRACK_MODIFICATIONS = app.debug
346
```
347

348
### Shell Integration Configuration
349

350
```python
351
# Shell integration (enabled by default)
352
db = SQLAlchemy(app, add_models_to_shell=True)
353

354
# Disable shell integration
355
db = SQLAlchemy(app, add_models_to_shell=False)
356
```
357

358
## Performance Considerations
359

360
- **Query Recording**: Adds overhead to every query, only enable in development
361
- **Modification Tracking**: Tracks changes in memory, can use significant memory with large change sets
362
- **Production Usage**: Both features are primarily for development and should generally be disabled in production
363
- **Memory Usage**: Recorded queries accumulate in memory during request lifetime
364
- **Signal Overhead**: Model change signals add processing overhead to commits