(
self,
context: DefaultExecutionContext,
cursor_strategy: ResultFetchStrategy,
cursor_description: Optional[_DBAPICursorDescription],
)
| 1645 | connection: Connection |
| 1646 | |
| 1647 | def __init__( |
| 1648 | self, |
| 1649 | context: DefaultExecutionContext, |
| 1650 | cursor_strategy: ResultFetchStrategy, |
| 1651 | cursor_description: Optional[_DBAPICursorDescription], |
| 1652 | ): |
| 1653 | self.context = context |
| 1654 | self.dialect = context.dialect |
| 1655 | self.cursor = context.cursor |
| 1656 | self.cursor_strategy = cursor_strategy |
| 1657 | self.connection = context.root_connection |
| 1658 | self._echo = echo = ( |
| 1659 | self.connection._echo and context.engine._should_log_debug() |
| 1660 | ) |
| 1661 | |
| 1662 | if cursor_description is not None: |
| 1663 | self._init_metadata(context, cursor_description) |
| 1664 | |
| 1665 | if echo: |
| 1666 | log = self.context.connection._log_debug |
| 1667 | |
| 1668 | def _log_row(row: Any) -> Any: |
| 1669 | log("Row %r", sql_util._repr_row(row)) |
| 1670 | return row |
| 1671 | |
| 1672 | self._row_logging_fn = _log_row |
| 1673 | |
| 1674 | # call Result._row_getter to set up the row factory |
| 1675 | self._row_getter |
| 1676 | |
| 1677 | else: |
| 1678 | assert context._num_sentinel_cols == 0 |
| 1679 | self._metadata = self._no_result_metadata |
| 1680 | |
| 1681 | def _init_metadata( |
| 1682 | self, |
nothing calls this directly
no test coverage detected