Framework Integrations

class Pony:
    def __init__(self, app=None):
        """Initialize Pony Flask extension.
        
        Args:
            app: Flask application instance (optional for factory pattern)
        """
        
    def init_app(self, app):
        """Initialize extension with Flask application.
        
        Args:
            app: Flask application instance
            
        Sets up before_request and teardown_request handlers for automatic
        db_session management throughout request lifecycle.
        """

from flask import Flask
from pony.flask import Pony
from pony.orm import *

# Create Flask app
app = Flask(__name__)

# Set up database
db = Database()

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

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

# Initialize Pony extension
pony = Pony(app)

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

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

if __name__ == '__main__':
    app.run(debug=True)

from flask import Flask
from pony.flask import Pony
from pony.orm import Database

# Global instances
db = Database()
pony = Pony()

def create_app(config=None):
    """Application factory pattern with Pony integration."""
    app = Flask(__name__)
    
    # Configure app
    if config:
        app.config.update(config)
    
    # Initialize database
    db.bind('sqlite', filename=app.config.get('DATABASE_URL', 'app.db'))
    
    # Initialize Pony extension
    pony.init_app(app)
    
    # Register blueprints after Pony initialization
    from .routes import api_bp
    app.register_blueprint(api_bp)
    
    return app

# In routes.py
from flask import Blueprint, request, jsonify
from .models import User, db

api_bp = Blueprint('api', __name__)

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

@api_bp.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    try:
        user = User[user_id]
        data = request.get_json()
        
        if 'name' in data:
            user.name = data['name']
        if 'email' in data:
            user.email = data['email']
            
        # Changes automatically committed at request end
        return jsonify({'id': user.id, 'name': user.name, 'email': user.email})
        
    except ObjectNotFound:
        return jsonify({'error': 'User not found'}), 404
    except IntegrityError:
        return jsonify({'error': 'Email already exists'}), 400

from flask import Flask, request, jsonify, g
from pony.flask import Pony
from pony.orm import *
import logging

app = Flask(__name__)
pony = Pony(app)

# Database setup
db = Database()

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

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

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

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

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

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

# Routes with automatic session management
@app.route('/users', methods=['GET'])
def list_users():
    # Query parameters for filtering
    name_filter = request.args.get('name')
    
    if name_filter:
        users = select(u for u in User if name_filter in u.name)[:]
    else:
        users = User.select()[:]
    
    return jsonify({
        'users': [
            {'id': u.id, 'name': u.name, 'email': u.email, 'created_at': u.created_at.isoformat()}
            for u in users
        ]
    })

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    
    if not data or 'name' not in data or 'email' not in data:
        return jsonify({'error': 'Name and email required'}), 400
    
    # Validation
    if User.exists(email=data['email']):
        return jsonify({'error': 'Email already exists'}), 400
    
    user = User(name=data['name'], email=data['email'])
    
    return jsonify({
        'id': user.id,
        'name': user.name,
        'email': user.email,
        'created_at': user.created_at.isoformat()
    }), 201

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

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

@app.teardown_request
def teardown_request(exception):
    # Custom cleanup if needed
    if exception:
        logging.error(f"Request failed with exception: {exception}")

if __name__ == '__main__':
    app.run(debug=True)

from flask import Flask
from pony.flask import Pony
from pony.orm import Database

app = Flask(__name__)

# Multiple database instances
user_db = Database()
log_db = Database()

# User models
class User(user_db.Entity):
    name = Required(str)
    email = Required(str, unique=True)

# Logging models  
class AccessLog(log_db.Entity):
    user_id = Optional(int)
    endpoint = Required(str)
    timestamp = Required(datetime, default=datetime.now)

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

user_db.generate_mapping(create_tables=True)
log_db.generate_mapping(create_tables=True)

# Initialize Pony for main database
pony = Pony(app)

@app.route('/users/<int:user_id>')
def get_user(user_id):
    # Main database session handled automatically
    user = User[user_id]
    
    # Manual session for logging database
    with log_db.db_session:
        AccessLog(user_id=user_id, endpoint=request.endpoint)
    
    return jsonify({'id': user.id, 'name': user.name})

import pytest
from flask import Flask
from pony.flask import Pony
from pony.orm import *

def create_test_app():
    """Create Flask app for testing."""
    app = Flask(__name__)
    app.config['TESTING'] = True
    
    # Use in-memory SQLite for tests
    db = Database()
    
    class User(db.Entity):
        name = Required(str)
        email = Required(str, unique=True)
    
    db.bind('sqlite', ':memory:')
    db.generate_mapping(create_tables=True)
    
    pony = Pony(app)
    
    @app.route('/users', methods=['POST'])
    def create_user():
        from flask import request
        data = request.get_json()
        user = User(name=data['name'], email=data['email'])
        return {'id': user.id}
    
    return app, db

@pytest.fixture
def app():
    app, db = create_test_app()
    yield app

@pytest.fixture
def client(app):
    return app.test_client()

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

def test_automatic_rollback_on_error(client):
    """Test that errors trigger automatic rollback."""
    # First request should succeed
    response1 = client.post('/users',
                           json={'name': 'User1', 'email': 'user1@example.com'})
    assert response1.status_code == 200
    
    # Second request with same email should fail and rollback
    response2 = client.post('/users',
                           json={'name': 'User2', 'email': 'user1@example.com'})
    assert response2.status_code != 200
    
    # Database should remain consistent
    # (Additional verification would require access to db instance)