Skip to content

Reference

shared.dry_run.bypass_dry_run 

bypass_dry_run()

Force-disable dry-run for the wrapped block.

Counterpart of dry_run(). Lets a nested transaction() actually commit while everything outside the block keeps rolling back. Intended for audit/trace persistence (e.g. EngineRun rows the inspection UI relies on) where the dry-run consumer explicitly wants the trace preserved.

Do NOT use this to escape dry-run for business writes — it defeats the dry-run contract.

Example

with dry_run(): with bypass_dry_run(): # transactions in here commit normally assert not is_dry_run() # dry-run is restored on exit assert is_dry_run()

Source code in shared/dry_run/context.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@contextmanager
def bypass_dry_run() -> Iterator[None]:
    """Force-disable dry-run for the wrapped block.

    Counterpart of `dry_run()`. Lets a nested `transaction()` actually commit
    while everything outside the block keeps rolling back. Intended for
    audit/trace persistence (e.g. EngineRun rows the inspection UI relies
    on) where the dry-run consumer explicitly wants the trace preserved.

    Do NOT use this to escape dry-run for business writes — it defeats the
    dry-run contract.

    Example:
        with dry_run():
            with bypass_dry_run():
                # transactions in here commit normally
                assert not is_dry_run()
            # dry-run is restored on exit
            assert is_dry_run()
    """
    token = _DRY_RUN_CONTEXT.set(False)
    try:
        yield
    finally:
        _DRY_RUN_CONTEXT.reset(token)

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

bypass_dry_run 

bypass_dry_run()

Force-disable dry-run for the wrapped block.

Counterpart of dry_run(). Lets a nested transaction() actually commit while everything outside the block keeps rolling back. Intended for audit/trace persistence (e.g. EngineRun rows the inspection UI relies on) where the dry-run consumer explicitly wants the trace preserved.

Do NOT use this to escape dry-run for business writes — it defeats the dry-run contract.

Example

with dry_run(): with bypass_dry_run(): # transactions in here commit normally assert not is_dry_run() # dry-run is restored on exit assert is_dry_run()

Source code in shared/dry_run/context.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@contextmanager
def bypass_dry_run() -> Iterator[None]:
    """Force-disable dry-run for the wrapped block.

    Counterpart of `dry_run()`. Lets a nested `transaction()` actually commit
    while everything outside the block keeps rolling back. Intended for
    audit/trace persistence (e.g. EngineRun rows the inspection UI relies
    on) where the dry-run consumer explicitly wants the trace preserved.

    Do NOT use this to escape dry-run for business writes — it defeats the
    dry-run contract.

    Example:
        with dry_run():
            with bypass_dry_run():
                # transactions in here commit normally
                assert not is_dry_run()
            # dry-run is restored on exit
            assert is_dry_run()
    """
    token = _DRY_RUN_CONTEXT.set(False)
    try:
        yield
    finally:
        _DRY_RUN_CONTEXT.reset(token)

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 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@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)

    try:
        # 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
        # FIXME This condition is not fully correct: in tests, it's possible to have has_db = True BUT be in a
        # scenario where the DB fixture is not enabled, and thus have current_app.db crash. See PR #84027 for details.
        # This is why this condition is within the `try: finally`: that way we can still correctly reset the
        # _DRY_RUN_CONTEXT even if the current_app.db call fails.
        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,
            )

        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 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@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)

    try:
        # 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
        # FIXME This condition is not fully correct: in tests, it's possible to have has_db = True BUT be in a
        # scenario where the DB fixture is not enabled, and thus have current_app.db crash. See PR #84027 for details.
        # This is why this condition is within the `try: finally`: that way we can still correctly reset the
        # _DRY_RUN_CONTEXT even if the current_app.db call fails.
        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,
            )

        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 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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()