Initialize execution context for a string SQL statement.
(
cls,
dialect: Dialect,
connection: Connection,
dbapi_connection: PoolProxiedConnection,
execution_options: _ExecuteOptions,
statement: str,
parameters: _DBAPIMultiExecuteParams,
)
| 1675 | |
| 1676 | @classmethod |
| 1677 | def _init_statement( |
| 1678 | cls, |
| 1679 | dialect: Dialect, |
| 1680 | connection: Connection, |
| 1681 | dbapi_connection: PoolProxiedConnection, |
| 1682 | execution_options: _ExecuteOptions, |
| 1683 | statement: str, |
| 1684 | parameters: _DBAPIMultiExecuteParams, |
| 1685 | ) -> ExecutionContext: |
| 1686 | """Initialize execution context for a string SQL statement.""" |
| 1687 | |
| 1688 | self = cls.__new__(cls) |
| 1689 | self.root_connection = connection |
| 1690 | self._dbapi_connection = dbapi_connection |
| 1691 | self.dialect = connection.dialect |
| 1692 | self.is_text = True |
| 1693 | |
| 1694 | self.execution_options = execution_options |
| 1695 | |
| 1696 | if not parameters: |
| 1697 | if self.dialect.positional: |
| 1698 | self.parameters = [dialect.execute_sequence_format()] |
| 1699 | else: |
| 1700 | self.parameters = [self._empty_dict_params] |
| 1701 | elif isinstance(parameters[0], dialect.execute_sequence_format): |
| 1702 | self.parameters = parameters |
| 1703 | elif isinstance(parameters[0], dict): |
| 1704 | self.parameters = parameters |
| 1705 | else: |
| 1706 | self.parameters = [ |
| 1707 | dialect.execute_sequence_format(p) for p in parameters |
| 1708 | ] |
| 1709 | |
| 1710 | if len(parameters) > 1: |
| 1711 | self.execute_style = ExecuteStyle.EXECUTEMANY |
| 1712 | |
| 1713 | self.statement = self.unicode_statement = statement |
| 1714 | |
| 1715 | self.cursor = self.create_cursor() |
| 1716 | return self |
| 1717 | |
| 1718 | @classmethod |
| 1719 | def _init_default( |
nothing calls this directly
no test coverage detected