Skip to content

Reference

shared.dry_run.context

Dry-run context manager implementation using contextvars for thread-safe operation.

This module provides a context manager to set dry-run mode and a helper function to check if the current execution context is in dry-run mode.

Key behaviors: - Once dry-run is enabled, it cannot be disabled in nested contexts - Context manager is safely nestable

Limitations: - Dry-run is not passed down to child threads, if you do multi-threading, ensure you pass the current dry-run state to child threads

dry_run

dry_run()

Dry-run context manager.

Once enabled, dry-run mode cannot be disabled in nested contexts.

Dry-run is not passed down to child threads, if you do multi-threading, ensure you pass the current dry-run state to child threads manually.

Example

with dry_run(): # code runs in dry-run mode assert is_dry_run()

with dry_run():
    # still in dry run in inner contexts
    assert is_dry_run()

# even when exiting the inner context
assert is_dry_run()

disabled only when exiting the outermost context

assert not is_dry_run()

Source code in shared/dry_run/context.py
@contextmanager
def dry_run() -> Iterator[None]:
    """Dry-run context manager.

    Once enabled, dry-run mode cannot be disabled in nested contexts.

    Dry-run is not passed down to child threads, if you do multi-threading,
    ensure you pass the current dry-run state to child threads manually.

    Example:
        with dry_run():
            # code runs in dry-run mode
            assert is_dry_run()

            with dry_run():
                # still in dry run in inner contexts
                assert is_dry_run()

            # even when exiting the inner context
            assert is_dry_run()

        # disabled only when exiting the outermost context
        assert not is_dry_run()
    """
    is_root = not _DRY_RUN_CONTEXT.get()
    token = _DRY_RUN_CONTEXT.set(True)

    # TODO @antoine.bertin 2025-10-01 remove this when current_session points to
    # transaction as well as the db fixture from the tests
    has_db = "sqlalchemy" in current_app.extensions
    if has_db and is_root and current_app.db.session:  # type: ignore[attr-defined]
        event.listen(
            current_app.db.session,  # type: ignore[attr-defined]
            "before_commit",
            _raise_commit_warning,
        )
    try:
        yield
    finally:
        _DRY_RUN_CONTEXT.reset(token)

        if has_db and is_root and current_app.db.session:  # type: ignore[attr-defined]
            event.remove(
                current_app.db.session,  # type: ignore[attr-defined]
                "before_commit",
                _raise_commit_warning,
            )

is_dry_run

is_dry_run()

Check if the current execution context is in dry-run mode.

Returns:

Name Type Description
bool bool

True if currently in dry-run mode, False otherwise.

Example

with dry_run(): assert is_dry_run()

assert not is_dry_run()

Source code in shared/dry_run/context.py
def is_dry_run() -> bool:
    """Check if the current execution context is in dry-run mode.

    Returns:
        bool: True if currently in dry-run mode, False otherwise.

    Example:
        with dry_run():
            assert is_dry_run()

        assert not is_dry_run()
    """
    return _DRY_RUN_CONTEXT.get()

shared.dry_run.dry_run

dry_run()

Dry-run context manager.

Once enabled, dry-run mode cannot be disabled in nested contexts.

Dry-run is not passed down to child threads, if you do multi-threading, ensure you pass the current dry-run state to child threads manually.

Example

with dry_run(): # code runs in dry-run mode assert is_dry_run()

with dry_run():
    # still in dry run in inner contexts
    assert is_dry_run()

# even when exiting the inner context
assert is_dry_run()

disabled only when exiting the outermost context

assert not is_dry_run()

Source code in shared/dry_run/context.py
@contextmanager
def dry_run() -> Iterator[None]:
    """Dry-run context manager.

    Once enabled, dry-run mode cannot be disabled in nested contexts.

    Dry-run is not passed down to child threads, if you do multi-threading,
    ensure you pass the current dry-run state to child threads manually.

    Example:
        with dry_run():
            # code runs in dry-run mode
            assert is_dry_run()

            with dry_run():
                # still in dry run in inner contexts
                assert is_dry_run()

            # even when exiting the inner context
            assert is_dry_run()

        # disabled only when exiting the outermost context
        assert not is_dry_run()
    """
    is_root = not _DRY_RUN_CONTEXT.get()
    token = _DRY_RUN_CONTEXT.set(True)

    # TODO @antoine.bertin 2025-10-01 remove this when current_session points to
    # transaction as well as the db fixture from the tests
    has_db = "sqlalchemy" in current_app.extensions
    if has_db and is_root and current_app.db.session:  # type: ignore[attr-defined]
        event.listen(
            current_app.db.session,  # type: ignore[attr-defined]
            "before_commit",
            _raise_commit_warning,
        )
    try:
        yield
    finally:
        _DRY_RUN_CONTEXT.reset(token)

        if has_db and is_root and current_app.db.session:  # type: ignore[attr-defined]
            event.remove(
                current_app.db.session,  # type: ignore[attr-defined]
                "before_commit",
                _raise_commit_warning,
            )

shared.dry_run.is_dry_run

is_dry_run()

Check if the current execution context is in dry-run mode.

Returns:

Name Type Description
bool bool

True if currently in dry-run mode, False otherwise.

Example

with dry_run(): assert is_dry_run()

assert not is_dry_run()

Source code in shared/dry_run/context.py
def is_dry_run() -> bool:
    """Check if the current execution context is in dry-run mode.

    Returns:
        bool: True if currently in dry-run mode, False otherwise.

    Example:
        with dry_run():
            assert is_dry_run()

        assert not is_dry_run()
    """
    return _DRY_RUN_CONTEXT.get()