0
# Framework Integrations
1

2
Integrations with popular Python web frameworks for automatic session management and simplified database operations in web applications. These integrations handle session lifecycle automatically, reducing boilerplate code and preventing common session management errors.
3

4
## Capabilities
5

6
### Flask Integration
7

8
Flask extension for automatic database session management that integrates with Flask's request lifecycle.
9

10
```python { .api }
11
class Pony:
12
    def __init__(self, app=None):
13
        """Initialize Pony Flask extension.
14
        
15
        Args:
16
            app: Flask application instance (optional for factory pattern)
17
        """
18
        
19
    def init_app(self, app):
20
        """Initialize extension with Flask application.
21
        
22
        Args:
23
            app: Flask application instance
24
            
25
        Sets up before_request and teardown_request handlers for automatic
26
        db_session management throughout request lifecycle.
27
        """
28
```
29

30
## Usage Examples
31

32
### Basic Flask Integration
33

34
```python
35
from flask import Flask
36
from pony.flask import Pony
37
from pony.orm import *
38

39
# Create Flask app
40
app = Flask(__name__)
41

42
# Set up database
43
db = Database()
44

45
class User(db.Entity):
46
    name = Required(str)
47
    email = Required(str, unique=True)
48

49
db.bind('sqlite', filename='app.db')
50
db.generate_mapping(create_tables=True)
51

52
# Initialize Pony extension
53
pony = Pony(app)
54

55
# Now all Flask routes automatically run in db_session context
56
@app.route('/users')
57
def list_users():
58
    # No need for @db_session decorator or with db_session:
59
    users = select(u for u in User)[:]
60
    return {'users': [{'name': u.name, 'email': u.email} for u in users]}
61

62
@app.route('/users', methods=['POST'])
63
def create_user():
64
    from flask import request
65
    data = request.get_json()
66
    
67
    # Automatic session management - changes committed at request end
68
    user = User(name=data['name'], email=data['email'])
69
    return {'id': user.id, 'name': user.name, 'email': user.email}
70

71
if __name__ == '__main__':
72
    app.run(debug=True)
73
```
74

75
### Flask Factory Pattern
76

77
```python
78
from flask import Flask
79
from pony.flask import Pony
80
from pony.orm import Database
81

82
# Global instances
83
db = Database()
84
pony = Pony()
85

86
def create_app(config=None):
87
    """Application factory pattern with Pony integration."""
88
    app = Flask(__name__)
89
    
90
    # Configure app
91
    if config:
92
        app.config.update(config)
93
    
94
    # Initialize database
95
    db.bind('sqlite', filename=app.config.get('DATABASE_URL', 'app.db'))
96
    
97
    # Initialize Pony extension
98
    pony.init_app(app)
99
    
100
    # Register blueprints after Pony initialization
101
    from .routes import api_bp
102
    app.register_blueprint(api_bp)
103
    
104
    return app
105

106
# In routes.py
107
from flask import Blueprint, request, jsonify
108
from .models import User, db
109

110
api_bp = Blueprint('api', __name__)
111

112
@api_bp.route('/users/<int:user_id>')
113
def get_user(user_id):
114
    # Automatic db_session context
115
    try:
116
        user = User[user_id]
117
        return jsonify({'id': user.id, 'name': user.name})
118
    except ObjectNotFound:
119
        return jsonify({'error': 'User not found'}), 404
120

121
@api_bp.route('/users/<int:user_id>', methods=['PUT'])
122
def update_user(user_id):
123
    try:
124
        user = User[user_id]
125
        data = request.get_json()
126
        
127
        if 'name' in data:
128
            user.name = data['name']
129
        if 'email' in data:
130
            user.email = data['email']
131
            
132
        # Changes automatically committed at request end
133
        return jsonify({'id': user.id, 'name': user.name, 'email': user.email})
134
        
135
    except ObjectNotFound:
136
        return jsonify({'error': 'User not found'}), 404
137
    except IntegrityError:
138
        return jsonify({'error': 'Email already exists'}), 400
139
```
140

141
### Advanced Flask Integration with Error Handling
142

143
```python
144
from flask import Flask, request, jsonify, g
145
from pony.flask import Pony
146
from pony.orm import *
147
import logging
148

149
app = Flask(__name__)
150
pony = Pony(app)
151

152
# Database setup
153
db = Database()
154

155
class User(db.Entity):
156
    name = Required(str)
157
    email = Required(str, unique=True)
158
    created_at = Required(datetime, default=datetime.now)
159

160
db.bind('sqlite', filename='app.db')
161
db.generate_mapping(create_tables=True)
162

163
# Custom error handlers that work with Pony sessions
164
@app.errorhandler(ObjectNotFound)
165
def handle_not_found(e):
166
    return jsonify({'error': 'Resource not found'}), 404
167

168
@app.errorhandler(MultipleObjectsFoundError)
169
def handle_multiple_found(e):
170
    return jsonify({'error': 'Multiple resources found'}), 500
171

172
@app.errorhandler(IntegrityError)
173
def handle_integrity_error(e):
174
    return jsonify({'error': 'Data integrity violation'}), 400
175

176
@app.errorhandler(TransactionError)
177
def handle_transaction_error(e):
178
    logging.error(f"Transaction error: {e}")
179
    return jsonify({'error': 'Transaction failed'}), 500
180

181
# Routes with automatic session management
182
@app.route('/users', methods=['GET'])
183
def list_users():
184
    # Query parameters for filtering
185
    name_filter = request.args.get('name')
186
    
187
    if name_filter:
188
        users = select(u for u in User if name_filter in u.name)[:]
189
    else:
190
        users = User.select()[:]
191
    
192
    return jsonify({
193
        'users': [
194
            {'id': u.id, 'name': u.name, 'email': u.email, 'created_at': u.created_at.isoformat()}
195
            for u in users
196
        ]
197
    })
198

199
@app.route('/users', methods=['POST'])
200
def create_user():
201
    data = request.get_json()
202
    
203
    if not data or 'name' not in data or 'email' not in data:
204
        return jsonify({'error': 'Name and email required'}), 400
205
    
206
    # Validation
207
    if User.exists(email=data['email']):
208
        return jsonify({'error': 'Email already exists'}), 400
209
    
210
    user = User(name=data['name'], email=data['email'])
211
    
212
    return jsonify({
213
        'id': user.id,
214
        'name': user.name,
215
        'email': user.email,
216
        'created_at': user.created_at.isoformat()
217
    }), 201
218

219
@app.route('/users/<int:user_id>', methods=['DELETE'])
220
def delete_user(user_id):
221
    user = User[user_id]  # Raises ObjectNotFound if not found
222
    user.delete()
223
    return '', 204
224

225
# Request middleware for additional session configuration
226
@app.before_request
227
def before_request():
228
    # Access to request-specific session configuration
229
    if request.endpoint and request.endpoint.startswith('admin'):
230
        # Enable SQL debugging for admin routes
231
        set_sql_debug(True)
232

233
@app.teardown_request
234
def teardown_request(exception):
235
    # Custom cleanup if needed
236
    if exception:
237
        logging.error(f"Request failed with exception: {exception}")
238

239
if __name__ == '__main__':
240
    app.run(debug=True)
241
```
242

243
### Flask with Multiple Databases
244

245
```python
246
from flask import Flask
247
from pony.flask import Pony
248
from pony.orm import Database
249

250
app = Flask(__name__)
251

252
# Multiple database instances
253
user_db = Database()
254
log_db = Database()
255

256
# User models
257
class User(user_db.Entity):
258
    name = Required(str)
259
    email = Required(str, unique=True)
260

261
# Logging models  
262
class AccessLog(log_db.Entity):
263
    user_id = Optional(int)
264
    endpoint = Required(str)
265
    timestamp = Required(datetime, default=datetime.now)
266

267
# Bind databases
268
user_db.bind('postgresql', host='localhost', user='app', password='secret', database='users')
269
log_db.bind('sqlite', filename='logs.db')
270

271
user_db.generate_mapping(create_tables=True)
272
log_db.generate_mapping(create_tables=True)
273

274
# Initialize Pony for main database
275
pony = Pony(app)
276

277
@app.route('/users/<int:user_id>')
278
def get_user(user_id):
279
    # Main database session handled automatically
280
    user = User[user_id]
281
    
282
    # Manual session for logging database
283
    with log_db.db_session:
284
        AccessLog(user_id=user_id, endpoint=request.endpoint)
285
    
286
    return jsonify({'id': user.id, 'name': user.name})
287
```
288

289
### Testing with Flask Integration
290

291
```python
292
import pytest
293
from flask import Flask
294
from pony.flask import Pony
295
from pony.orm import *
296

297
def create_test_app():
298
    """Create Flask app for testing."""
299
    app = Flask(__name__)
300
    app.config['TESTING'] = True
301
    
302
    # Use in-memory SQLite for tests
303
    db = Database()
304
    
305
    class User(db.Entity):
306
        name = Required(str)
307
        email = Required(str, unique=True)
308
    
309
    db.bind('sqlite', ':memory:')
310
    db.generate_mapping(create_tables=True)
311
    
312
    pony = Pony(app)
313
    
314
    @app.route('/users', methods=['POST'])
315
    def create_user():
316
        from flask import request
317
        data = request.get_json()
318
        user = User(name=data['name'], email=data['email'])
319
        return {'id': user.id}
320
    
321
    return app, db
322

323
@pytest.fixture
324
def app():
325
    app, db = create_test_app()
326
    yield app
327

328
@pytest.fixture
329
def client(app):
330
    return app.test_client()
331

332
def test_user_creation(client):
333
    """Test user creation through Flask integration."""
334
    response = client.post('/users', 
335
                          json={'name': 'Test User', 'email': 'test@example.com'})
336
    
337
    assert response.status_code == 200
338
    data = response.get_json()
339
    assert 'id' in data
340
    assert data['id'] > 0
341

342
def test_automatic_rollback_on_error(client):
343
    """Test that errors trigger automatic rollback."""
344
    # First request should succeed
345
    response1 = client.post('/users',
346
                           json={'name': 'User1', 'email': 'user1@example.com'})
347
    assert response1.status_code == 200
348
    
349
    # Second request with same email should fail and rollback
350
    response2 = client.post('/users',
351
                           json={'name': 'User2', 'email': 'user1@example.com'})
352
    assert response2.status_code != 200
353
    
354
    # Database should remain consistent
355
    # (Additional verification would require access to db instance)
356
```