Skip to content

API Reference

capsqlalchemy_context 

capsqlalchemy_context(
    db_engine: AsyncEngine,
) -> Generator[SQLAlchemyCaptureContext]

The main fixture to get the SQLAlchemyCaptureContext.

This is the context for the full test, which captures all SQL expressions executed during the test.

To capture only the SQL expressions executed within a specific block, use the capsqlalchemy fixture.

Source code in pytest_capsqlalchemy/plugin.py 
 9
10
11
12
13
14
15
16
17
18
19
@pytest.fixture
def capsqlalchemy_context(db_engine: AsyncEngine) -> Generator[SQLAlchemyCaptureContext]:
    """The main fixture to get the [`SQLAlchemyCaptureContext`][pytest_capsqlalchemy.context.SQLAlchemyCaptureContext].

    This is the context for the full test, which captures all SQL expressions executed during the test.

    To capture only the SQL expressions executed within a specific block, use the
    [`capsqlalchemy`][pytest_capsqlalchemy.plugin.capsqlalchemy] fixture.
    """
    with SQLAlchemyCaptureContext(db_engine) as capsqlalchemy_ctx:
        yield capsqlalchemy_ctx

capsqlalchemy 

capsqlalchemy(
    capsqlalchemy_context: SQLAlchemyCaptureContext,
) -> SQLAlchemyCapturer

The main fixture to get the SQLAlchemyCapturer.

Example Usage:

async def test_some_sql_queries(db_session, capsqlalchemy):
    await db_session.execute(select(text("1")))
    capsqlalchemy.assert_query_count(1, include_tcl=False)

    async with capsqlalchemy:
        await db_session.execute(select(text("2")))
        await db_session.execute(select(text("3")))
        capsqlalchemy.assert_query_count(2, include_tcl=False)

    capsqlalchemy.assert_query_count(3, include_tcl=False)

Returns:

Type Description
SQLAlchemyCapturer

The capturer object with the full test context already set up.
Source code in pytest_capsqlalchemy/plugin.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@pytest.fixture()
def capsqlalchemy(capsqlalchemy_context: SQLAlchemyCaptureContext) -> SQLAlchemyCapturer:
    """The main fixture to get the [`SQLAlchemyCapturer`][pytest_capsqlalchemy.capturer.SQLAlchemyCapturer].

    Example Usage:

    ```python
    async def test_some_sql_queries(db_session, capsqlalchemy):
        await db_session.execute(select(text("1")))
        capsqlalchemy.assert_query_count(1, include_tcl=False)

        async with capsqlalchemy:
            await db_session.execute(select(text("2")))
            await db_session.execute(select(text("3")))
            capsqlalchemy.assert_query_count(2, include_tcl=False)

        capsqlalchemy.assert_query_count(3, include_tcl=False)
    ```

    Returns:
        The capturer object with the full test context already set up.
    """
    return SQLAlchemyCapturer(capsqlalchemy_context)

SQLAlchemyCapturer

The main fixture class for the capsqlalchemy plugin.

Used to perform asserts about the expressions SQLAlchemy has executed during the test.

Can be used either directly using the assert methods (to perform checks for all expressions executed in the test), as a context manager (to perform checks only for the expressions executed in a specific block), or a combination of both.

Intended to be used via the capsqlalchemy fixture

Source code in pytest_capsqlalchemy/capturer.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 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
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
class SQLAlchemyCapturer:
    """The main fixture class for the `capsqlalchemy` plugin.

    Used to perform asserts about the expressions SQLAlchemy has executed during the test.

    Can be used either directly using the assert methods (to perform checks for all expressions
    executed in the test), as a context manager (to perform checks only for the expressions
    executed in a specific block), or a combination of both.

    Intended to be used via the [`capsqlalchemy`][pytest_capsqlalchemy.plugin.capsqlalchemy] fixture
    """

    _full_test_context: SQLAlchemyCaptureContext
    _partial_context: Optional[SQLAlchemyCaptureContext]

    def __init__(self, full_test_context: SQLAlchemyCaptureContext):
        """Create a new SQLAlchemyCapturer instance."""
        self._full_test_context = full_test_context
        self._partial_context = None

    @property
    def engine(self) -> AsyncEngine:
        """The SQLAlchemy engine instance being captured."""
        return self._full_test_context._engine

    @property
    def captured_expressions(self) -> list[SQLExpression]:
        """Returns all SQL expressions captured in the current context.

        When used outside a context manager block, returns all expressions captured
        during the entire test. When used inside a context manager block, returns
        only the expressions captured within that specific block.

        This property is useful for performing specific assertions on the captured expressions which
        cannot be easily achieved with the provided assert methods.
        """
        if self._partial_context is not None:
            return self._partial_context.captured_expressions

        return self._full_test_context.captured_expressions

    def __enter__(self) -> Self:
        self._partial_context = SQLAlchemyCaptureContext(self.engine)
        self._partial_context = self._partial_context.__enter__()
        return self

    def __exit__(
        self,
        exc_type: Optional[type[BaseException]] = None,
        exc_val: Optional[BaseException] = None,
        exc_tb: Optional[TracebackType] = None,
    ) -> Optional[bool]:
        if self._partial_context is None:  # pragma: no cover
            raise RuntimeError(f"{self.__class__.__name__}: attempting to call __exit__ before __enter__")

        result = self._partial_context.__exit__(exc_type, exc_val, exc_tb)

        self._partial_context = None

        return result

    def assert_query_types(
        self,
        *expected_query_types: Union[SQLExpressionType, str],
        include_tcl: bool = True,
    ) -> None:
        """Asserts that the captured SQL expressions match the expected query types in order.

        This is useful for ensuring that your code is generating correct query types but
        their exact structure is not important (e.g. complex SELECT statements).

        Args:
            *expected_query_types: Variable number of expected query types
            include_tcl: Whether to include transaction control language statements (BEGIN,
                COMMIT, ROLLBACK) in the comparison

        Raises:
            AssertionError: If the actual query types don't match the expected ones.
        """
        actual_query_types_values = []

        for query in self.captured_expressions:
            if not include_tcl and query.type.is_tcl:
                continue

            actual_query_types_values.append(query.type._value_)

        # Converting to strings as the error message diff will be shorter and more readable
        expected_query_types_values = [
            query_type._value_ if isinstance(query_type, SQLExpressionType) else query_type
            for query_type in expected_query_types
        ]

        assert expected_query_types_values == actual_query_types_values

    def assert_query_count(self, expected_query_count: int, *, include_tcl: bool = True) -> None:
        """Asserts that the number of captured SQL expressions matches the expected count.

        This is useful for ensuring that your code is not generating more statements than expected
        (e.g. due to N+1 queries), however the exact queries are not important.

        Args:
            expected_query_count: The expected number of SQL expressions.
            include_tcl: Whether to include transaction control language statements (BEGIN,
                COMMIT, ROLLBACK) in the count.

        Raises:
            AssertionError: If the actual query count doesn't match the expected count.
        """
        actual_query_count = sum(1 for query in self.captured_expressions if include_tcl or not query.type.is_tcl)

        assert expected_query_count == actual_query_count, (
            f"Query count mismatch: expected {expected_query_count}, got {actual_query_count}"
        )

    def assert_max_query_count(self, expected_max_query_count: int, *, include_tcl: bool = True) -> None:
        """Asserts that the number of captured SQL expressions doesn't exceed the expected count.

        This is useful for ensuring that your code is not generating more statements than expected
        (e.g. due to N+1 queries), however the exact number of queries is not important -- for example
        SQLAlchemy's caching mechanism may generate fewer queries than expected.

        Args:
            expected_max_query_count: The expected maximum number of SQL expressions.
            include_tcl: Whether to include transaction control language statements (BEGIN,
                COMMIT, ROLLBACK) in the count.

        Raises:
            AssertionError: If the actual query count exceeds the expected maximum count.
        """
        actual_query_count = sum(1 for query in self.captured_expressions if include_tcl or not query.type.is_tcl)

        assert expected_max_query_count < actual_query_count, (
            f"Query count mismatch: expected maximum {expected_max_query_count}, got {actual_query_count}"
        )

    def assert_captured_queries(
        self,
        *expected_queries: str,
        include_tcl: bool = True,
        bind_params: bool = False,
    ) -> None:
        """Asserts that the captured SQL queries match the expected SQL strings in order.

        This is useful for ensuring that your code is generating the exact SQL statements you expect.

        Args:
            *expected_queries: Variable number of expected SQL query strings.
            include_tcl: Whether to include transaction control language statements (BEGIN,
                COMMIT, ROLLBACK) in the comparison.
            bind_params: Whether to include bound parameters in the SQL strings. When `False`,
                parameters are represented as placeholders instead.

        Raises:
            AssertionError: If the actual SQL queries don't match the expected ones.
        """
        actual_queries = []

        for query in self.captured_expressions:
            if not include_tcl and query.type.is_tcl:
                continue

            actual_queries.append(query.get_sql(bind_params=bind_params))

        assert list(expected_queries) == actual_queries

engine property 

engine: AsyncEngine

The SQLAlchemy engine instance being captured.

captured_expressions property 

captured_expressions: list[SQLExpression]

Returns all SQL expressions captured in the current context.

When used outside a context manager block, returns all expressions captured during the entire test. When used inside a context manager block, returns only the expressions captured within that specific block.

This property is useful for performing specific assertions on the captured expressions which cannot be easily achieved with the provided assert methods.

__init__ 

__init__(full_test_context: SQLAlchemyCaptureContext)

Create a new SQLAlchemyCapturer instance.

Source code in pytest_capsqlalchemy/capturer.py 
31
32
33
34
def __init__(self, full_test_context: SQLAlchemyCaptureContext):
    """Create a new SQLAlchemyCapturer instance."""
    self._full_test_context = full_test_context
    self._partial_context = None

assert_query_types 

assert_query_types(
    *expected_query_types: Union[SQLExpressionType, str],
    include_tcl: bool = True,
) -> None

Asserts that the captured SQL expressions match the expected query types in order.

This is useful for ensuring that your code is generating correct query types but their exact structure is not important (e.g. complex SELECT statements).

Parameters:

Name Type Description Default
*expected_query_types Union[SQLExpressionType, str]

Variable number of expected query types

 ()
include_tcl bool

Whether to include transaction control language statements (BEGIN, COMMIT, ROLLBACK) in the comparison

 True

Raises:

Type Description
AssertionError

If the actual query types don't match the expected ones.
Source code in pytest_capsqlalchemy/capturer.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def assert_query_types(
    self,
    *expected_query_types: Union[SQLExpressionType, str],
    include_tcl: bool = True,
) -> None:
    """Asserts that the captured SQL expressions match the expected query types in order.

    This is useful for ensuring that your code is generating correct query types but
    their exact structure is not important (e.g. complex SELECT statements).

    Args:
        *expected_query_types: Variable number of expected query types
        include_tcl: Whether to include transaction control language statements (BEGIN,
            COMMIT, ROLLBACK) in the comparison

    Raises:
        AssertionError: If the actual query types don't match the expected ones.
    """
    actual_query_types_values = []

    for query in self.captured_expressions:
        if not include_tcl and query.type.is_tcl:
            continue

        actual_query_types_values.append(query.type._value_)

    # Converting to strings as the error message diff will be shorter and more readable
    expected_query_types_values = [
        query_type._value_ if isinstance(query_type, SQLExpressionType) else query_type
        for query_type in expected_query_types
    ]

    assert expected_query_types_values == actual_query_types_values

assert_query_count 

assert_query_count(
    expected_query_count: int, *, include_tcl: bool = True
) -> None

Asserts that the number of captured SQL expressions matches the expected count.

This is useful for ensuring that your code is not generating more statements than expected (e.g. due to N+1 queries), however the exact queries are not important.

Parameters:

Name Type Description Default
expected_query_count int

The expected number of SQL expressions.

 required
include_tcl bool

Whether to include transaction control language statements (BEGIN, COMMIT, ROLLBACK) in the count.

 True

Raises:

Type Description
AssertionError

If the actual query count doesn't match the expected count.
Source code in pytest_capsqlalchemy/capturer.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def assert_query_count(self, expected_query_count: int, *, include_tcl: bool = True) -> None:
    """Asserts that the number of captured SQL expressions matches the expected count.

    This is useful for ensuring that your code is not generating more statements than expected
    (e.g. due to N+1 queries), however the exact queries are not important.

    Args:
        expected_query_count: The expected number of SQL expressions.
        include_tcl: Whether to include transaction control language statements (BEGIN,
            COMMIT, ROLLBACK) in the count.

    Raises:
        AssertionError: If the actual query count doesn't match the expected count.
    """
    actual_query_count = sum(1 for query in self.captured_expressions if include_tcl or not query.type.is_tcl)

    assert expected_query_count == actual_query_count, (
        f"Query count mismatch: expected {expected_query_count}, got {actual_query_count}"
    )

assert_max_query_count 

assert_max_query_count(
    expected_max_query_count: int,
    *,
    include_tcl: bool = True,
) -> None

Asserts that the number of captured SQL expressions doesn't exceed the expected count.

This is useful for ensuring that your code is not generating more statements than expected (e.g. due to N+1 queries), however the exact number of queries is not important -- for example SQLAlchemy's caching mechanism may generate fewer queries than expected.

Parameters:

Name Type Description Default
expected_max_query_count int

The expected maximum number of SQL expressions.

 required
include_tcl bool

Whether to include transaction control language statements (BEGIN, COMMIT, ROLLBACK) in the count.

 True

Raises:

Type Description
AssertionError

If the actual query count exceeds the expected maximum count.
Source code in pytest_capsqlalchemy/capturer.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def assert_max_query_count(self, expected_max_query_count: int, *, include_tcl: bool = True) -> None:
    """Asserts that the number of captured SQL expressions doesn't exceed the expected count.

    This is useful for ensuring that your code is not generating more statements than expected
    (e.g. due to N+1 queries), however the exact number of queries is not important -- for example
    SQLAlchemy's caching mechanism may generate fewer queries than expected.

    Args:
        expected_max_query_count: The expected maximum number of SQL expressions.
        include_tcl: Whether to include transaction control language statements (BEGIN,
            COMMIT, ROLLBACK) in the count.

    Raises:
        AssertionError: If the actual query count exceeds the expected maximum count.
    """
    actual_query_count = sum(1 for query in self.captured_expressions if include_tcl or not query.type.is_tcl)

    assert expected_max_query_count < actual_query_count, (
        f"Query count mismatch: expected maximum {expected_max_query_count}, got {actual_query_count}"
    )

assert_captured_queries 

assert_captured_queries(
    *expected_queries: str,
    include_tcl: bool = True,
    bind_params: bool = False,
) -> None

Asserts that the captured SQL queries match the expected SQL strings in order.

This is useful for ensuring that your code is generating the exact SQL statements you expect.

Parameters:

Name Type Description Default
*expected_queries str

Variable number of expected SQL query strings.

 ()
include_tcl bool

Whether to include transaction control language statements (BEGIN, COMMIT, ROLLBACK) in the comparison.

 True
bind_params bool

Whether to include bound parameters in the SQL strings. When False, parameters are represented as placeholders instead.

 False

Raises:

Type Description
AssertionError

If the actual SQL queries don't match the expected ones.
Source code in pytest_capsqlalchemy/capturer.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def assert_captured_queries(
    self,
    *expected_queries: str,
    include_tcl: bool = True,
    bind_params: bool = False,
) -> None:
    """Asserts that the captured SQL queries match the expected SQL strings in order.

    This is useful for ensuring that your code is generating the exact SQL statements you expect.

    Args:
        *expected_queries: Variable number of expected SQL query strings.
        include_tcl: Whether to include transaction control language statements (BEGIN,
            COMMIT, ROLLBACK) in the comparison.
        bind_params: Whether to include bound parameters in the SQL strings. When `False`,
            parameters are represented as placeholders instead.

    Raises:
        AssertionError: If the actual SQL queries don't match the expected ones.
    """
    actual_queries = []

    for query in self.captured_expressions:
        if not include_tcl and query.type.is_tcl:
            continue

        actual_queries.append(query.get_sql(bind_params=bind_params))

    assert list(expected_queries) == actual_queries

SQLAlchemyCaptureContext

Captures expressions executed on a SQLAlchemy engine within a specific context.

These expressions include:

* SELECT
* INSERT
* UPDATE
* DELETE
* BEGIN
* COMMIT
* ROLLBACK

Every expression is captured as a SQLExpression object, allowing it to be parsed correctly and compared against.

See SQLAlchemyCapturer for the available assertions on the captured expressions.

Source code in pytest_capsqlalchemy/context.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 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
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class SQLAlchemyCaptureContext:
    """Captures expressions executed on a SQLAlchemy engine within a specific context.

    These expressions include:

        * SELECT
        * INSERT
        * UPDATE
        * DELETE
        * BEGIN
        * COMMIT
        * ROLLBACK

    Every expression is captured as a SQLExpression object, allowing it to be parsed correctly
    and compared against.

    See [`SQLAlchemyCapturer`][pytest_capsqlalchemy.capturer.SQLAlchemyCapturer] for the available
    assertions on the captured expressions.
    """

    _engine: AsyncEngine
    _captured_expressions: list[SQLExpression]

    def __init__(self, engine: AsyncEngine):
        """Create a new SQLAlchemyCaptureContext instance."""
        self._engine = engine
        self._captured_expressions = []
        self._sqlaclhemy_events_stack = contextlib.ExitStack()

    @property
    def captured_expressions(self) -> list[SQLExpression]:
        """Returns all SQL expressions captured in the current context."""
        return self._captured_expressions

    def clear(self) -> None:
        """Clear all SQL expressions captured so far in the current context."""
        self._captured_expressions = []

    def _on_begin(self, conn: Connection) -> None:
        self._captured_expressions.append(SQLExpression(executable=text("BEGIN")))

    def _on_commit(self, conn: Connection) -> None:
        self._captured_expressions.append(SQLExpression(executable=text("COMMIT")))

    def _on_rollback(self, conn: Connection) -> None:
        self._captured_expressions.append(SQLExpression(executable=text("ROLLBACK")))

    def _on_after_execute(
        self,
        conn: Connection,
        clauseelement: Executable,
        multiparams: list[dict[str, Any]],
        params: dict[str, Any],
        execution_options: Mapping[str, Any],
        result: CursorResult,
    ) -> None:
        self._captured_expressions.append(
            SQLExpression(executable=clauseelement, params=params, multiparams=multiparams)
        )

    def __enter__(self) -> Self:
        events_stack = self._sqlaclhemy_events_stack.__enter__()

        for event_name, listener in (
            ("begin", self._on_begin),
            ("commit", self._on_commit),
            ("rollback", self._on_rollback),
            ("after_execute", self._on_after_execute),
        ):
            events_stack.enter_context(
                temp_sqlalchemy_event(
                    self._engine.sync_engine,
                    event_name,
                    listener,
                )
            )

        return self

    def __exit__(
        self,
        exc_type: Optional[type[BaseException]],
        exc_value: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> Optional[bool]:
        return self._sqlaclhemy_events_stack.__exit__(exc_type, exc_value, traceback)

captured_expressions property 

captured_expressions: list[SQLExpression]

Returns all SQL expressions captured in the current context.

__init__ 

__init__(engine: AsyncEngine)

Create a new SQLAlchemyCaptureContext instance.

Source code in pytest_capsqlalchemy/context.py 
42
43
44
45
46
def __init__(self, engine: AsyncEngine):
    """Create a new SQLAlchemyCaptureContext instance."""
    self._engine = engine
    self._captured_expressions = []
    self._sqlaclhemy_events_stack = contextlib.ExitStack()

clear 

clear() -> None

Clear all SQL expressions captured so far in the current context.

Source code in pytest_capsqlalchemy/context.py 
53
54
55
def clear(self) -> None:
    """Clear all SQL expressions captured so far in the current context."""
    self._captured_expressions = []

SQLExpressionType

Bases: str, Enum

An enumeration of the different types of SQL expressions that can be captured.

Source code in pytest_capsqlalchemy/expression.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class SQLExpressionType(str, enum.Enum):
    """An enumeration of the different types of SQL expressions that can be captured."""

    SELECT = "SELECT"
    INSERT = "INSERT"
    UPDATE = "UPDATE"
    DELETE = "DELETE"
    BEGIN = "BEGIN"
    COMMIT = "COMMIT"
    ROLLBACK = "ROLLBACK"
    UNKNOWN = "UNKNOWN"

    @property
    def is_tcl(self) -> bool:
        """Check if the SQL expression type is a transaction control language statement."""
        return self in {SQLExpressionType.BEGIN, SQLExpressionType.COMMIT, SQLExpressionType.ROLLBACK}

is_tcl property 

is_tcl: bool

Check if the SQL expression type is a transaction control language statement.

SQLExpression dataclass

A representation of a single SQL expression captured by SQLAlchemy.

Stores the SQLAlchemy Executable object and any parameters used in the query, so that it can be compared against expected queries in tests. This is useful for performing specific assertions on the captured expressions which cannot be easily achieved with the provided assert methods.

Source code in pytest_capsqlalchemy/expression.py 
26
27
28
29
30
31
32
33
34
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
87
88
89
90
91
92
93
94
@dataclass
class SQLExpression:
    """A representation of a single SQL expression captured by SQLAlchemy.

    Stores the SQLAlchemy `Executable` object and any parameters used in the query, so that it can be
    compared against expected queries in tests. This is useful for performing specific assertions
    on the captured expressions which cannot be easily achieved with the provided assert methods.
    """

    executable: Executable
    params: dict[str, Any] = field(default_factory=dict)
    multiparams: list[dict[str, Any]] = field(default_factory=list)

    def get_sql(self, *, bind_params: bool = False) -> str:
        """Get the SQL string generated by SQLAlchemy of the captured expression.

        Args:
            bind_params: If True, the SQL string will include the bound parameters in the query. Otherwise the
                SQL string will contain placeholders for the bound parameters.

        Returns:
            The SQL string of the captured expression
        """
        assert isinstance(self.executable, ClauseElement)

        if self.executable.is_insert:
            assert isinstance(self.executable, Insert)

            if self.multiparams:
                expr = self.executable.values(self.multiparams)
            elif self.params:
                expr = self.executable.values(self.params)
            else:
                expr = self.executable
        else:
            expr = self.executable

        compile_kwargs = {}
        if bind_params:
            compile_kwargs["literal_binds"] = True

        return str(expr.compile(compile_kwargs=compile_kwargs))

    @property
    def type(self) -> SQLExpressionType:
        """Get the type of the captured SQL expression."""
        if self.executable.is_insert:
            return SQLExpressionType.INSERT

        if self.executable.is_select:
            return SQLExpressionType.SELECT

        if self.executable.is_update:
            return SQLExpressionType.UPDATE

        if self.executable.is_delete:
            return SQLExpressionType.DELETE

        if isinstance(self.executable, TextClause):
            if self.executable.compare(text("BEGIN")):
                return SQLExpressionType.BEGIN

            if self.executable.compare(text("COMMIT")):
                return SQLExpressionType.COMMIT

            if self.executable.compare(text("ROLLBACK")):
                return SQLExpressionType.ROLLBACK

        return SQLExpressionType.UNKNOWN

type property 

type: SQLExpressionType

Get the type of the captured SQL expression.

get_sql 

get_sql(*, bind_params: bool = False) -> str

Get the SQL string generated by SQLAlchemy of the captured expression.

Parameters:

Name Type Description Default
bind_params bool

If True, the SQL string will include the bound parameters in the query. Otherwise the SQL string will contain placeholders for the bound parameters.

 False

Returns:

Type Description
str

The SQL string of the captured expression
Source code in pytest_capsqlalchemy/expression.py 
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
def get_sql(self, *, bind_params: bool = False) -> str:
    """Get the SQL string generated by SQLAlchemy of the captured expression.

    Args:
        bind_params: If True, the SQL string will include the bound parameters in the query. Otherwise the
            SQL string will contain placeholders for the bound parameters.

    Returns:
        The SQL string of the captured expression
    """
    assert isinstance(self.executable, ClauseElement)

    if self.executable.is_insert:
        assert isinstance(self.executable, Insert)

        if self.multiparams:
            expr = self.executable.values(self.multiparams)
        elif self.params:
            expr = self.executable.values(self.params)
        else:
            expr = self.executable
    else:
        expr = self.executable

    compile_kwargs = {}
    if bind_params:
        compile_kwargs["literal_binds"] = True

    return str(expr.compile(compile_kwargs=compile_kwargs))

temp_sqlalchemy_event 

temp_sqlalchemy_event(
    target: Any,
    identifier: str,
    fn: Callable[..., Any],
    *args: Any,
    **kwargs: Any,
) -> Generator[None, None, None]

Temporarily add a SQLAlchemy event listener to the target object.

The event listener is automatically removed when the context manager exits.

Source code in pytest_capsqlalchemy/utils.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@contextlib.contextmanager
def temp_sqlalchemy_event(
    target: Any,
    identifier: str,
    fn: Callable[..., Any],
    *args: Any,
    **kwargs: Any,
) -> Generator[None, None, None]:
    """Temporarily add a SQLAlchemy event listener to the target object.

    The event listener is automatically removed when the context manager exits.
    """
    event.listen(target, identifier, fn, *args, **kwargs)

    try:
        yield
    finally:
        event.remove(target, identifier, fn)