Transaction Callbacks

def django_capture_on_commit_callbacks() -> DjangoCaptureOnCommitCallbacks:
    """
    Fixture for capturing Django on_commit callbacks.
    
    Returns a context manager that captures callbacks registered with
    transaction.on_commit() during test execution. Useful for testing
    code that defers operations until transaction commit.
    
    Returns:
        DjangoCaptureOnCommitCallbacks: Callback capture context manager
    """

class DjangoCaptureOnCommitCallbacks:
    """Context manager for capturing on_commit callbacks."""
    
    def __call__(
        self,
        *,
        using: str = "default",
        execute: bool = False,
    ) -> ContextManager[List[Callable[[], Any]]]:
        """
        Create context manager for capturing callbacks.
        
        Args:
            using: Database alias to capture callbacks for
            execute: Whether to execute captured callbacks automatically
        
        Returns:
            ContextManager that yields list of captured callback functions
        """

from django.db import transaction

def test_on_commit_callback(django_capture_on_commit_callbacks):
    """Test that on_commit callbacks are registered correctly."""
    
    def my_callback():
        print("Transaction committed!")
    
    # Capture callbacks registered during context
    with django_capture_on_commit_callbacks() as callbacks:
        # Register callback that should run on commit
        transaction.on_commit(my_callback)
        
        # Perform database operations
        from myapp.models import MyModel
        MyModel.objects.create(name="test")
    
    # Verify callback was captured
    assert len(callbacks) == 1
    assert callbacks[0] == my_callback

def test_multiple_callbacks(django_capture_on_commit_callbacks):
    """Test multiple on_commit callbacks."""
    
    callback_results = []
    
    def callback_1():
        callback_results.append("callback_1")
    
    def callback_2():
        callback_results.append("callback_2")
    
    with django_capture_on_commit_callbacks() as callbacks:
        transaction.on_commit(callback_1)
        transaction.on_commit(callback_2)
        
        # Some database operation
        from myapp.models import MyModel
        MyModel.objects.create(name="test")
    
    # Both callbacks captured
    assert len(callbacks) == 2
    assert callback_1 in callbacks
    assert callback_2 in callbacks
    
    # Manually execute callbacks to test behavior
    for callback in callbacks:
        callback()
    
    assert callback_results == ["callback_1", "callback_2"]

def test_conditional_callback(django_capture_on_commit_callbacks):
    """Test conditional on_commit callback registration."""
    
    def process_model(instance, send_email=False):
        instance.save()
        
        if send_email:
            def send_notification():
                # Email sending logic here
                pass
            transaction.on_commit(send_notification)
    
    from myapp.models import MyModel
    
    # Test without email
    with django_capture_on_commit_callbacks() as callbacks:
        obj = MyModel(name="test")
        process_model(obj, send_email=False)
    
    assert len(callbacks) == 0
    
    # Test with email
    with django_capture_on_commit_callbacks() as callbacks:
        obj = MyModel(name="test2")
        process_model(obj, send_email=True)
    
    assert len(callbacks) == 1

@pytest.mark.django_db(transaction=True)
def test_callback_with_rollback(django_capture_on_commit_callbacks):
    """Test that callbacks are not called on rollback."""
    
    callback_called = []
    
    def my_callback():
        callback_called.append(True)
    
    from myapp.models import MyModel
    
    try:
        with transaction.atomic():
            with django_capture_on_commit_callbacks() as callbacks:
                transaction.on_commit(my_callback)
                MyModel.objects.create(name="test")
                
                # Force rollback
                raise Exception("Force rollback")
    except Exception:
        pass
    
    # Callback was registered but not executed due to rollback
    assert len(callbacks) == 1
    assert len(callback_called) == 0

def test_nested_transactions(django_capture_on_commit_callbacks):
    """Test callbacks with nested transactions."""
    
    outer_callback_called = []
    inner_callback_called = []
    
    def outer_callback():
        outer_callback_called.append(True)
    
    def inner_callback():
        inner_callback_called.append(True)
    
    from myapp.models import MyModel
    
    with django_capture_on_commit_callbacks() as callbacks:
        # Outer transaction callback
        transaction.on_commit(outer_callback)
        
        with transaction.atomic():
            # Inner transaction callback
            transaction.on_commit(inner_callback)
            MyModel.objects.create(name="test")
    
    # Both callbacks should be captured
    assert len(callbacks) == 2
    assert outer_callback in callbacks
    assert inner_callback in callbacks

from typing import ContextManager, List, Callable, Any
from django.db import transaction

# Callback function type
CallbackFunction = Callable[[], Any]
CallbackList = List[CallbackFunction]

# Callback capture context manager
class DjangoCaptureOnCommitCallbacks:
    """Context manager for capturing Django on_commit callbacks."""
    
    def __call__(self) -> ContextManager[CallbackList]:
        """
        Create context manager that captures callbacks.
        
        Returns:
            ContextManager that yields list of callback functions
            registered with transaction.on_commit() during context
        """
    
    def __enter__(self) -> CallbackList:
        """Enter context and start capturing callbacks."""
    
    def __exit__(self, exc_type, exc_value, traceback) -> None:
        """Exit context and stop capturing callbacks."""

# Transaction state types
class TransactionState:
    """Represents transaction state during callback capture."""
    in_atomic_block: bool
    needs_rollback: bool
    savepoint_ids: List[str]

# Internal callback management
class CallbackManager:
    """Manages callback registration and execution."""
    callbacks: List[CallbackFunction]
    
    def register(self, callback: CallbackFunction) -> None: ...
    def clear(self) -> None: ...
    def execute_all(self) -> None: ...