continue the _execute_context() method for a single DBAPI cursor.execute() or cursor.executemany() call.
(
self,
dialect: Dialect,
context: ExecutionContext,
statement: Union[str, Compiled],
parameters: Optional[_AnyMultiExecuteParams],
)
| 1824 | ) |
| 1825 | |
| 1826 | def _exec_single_context( |
| 1827 | self, |
| 1828 | dialect: Dialect, |
| 1829 | context: ExecutionContext, |
| 1830 | statement: Union[str, Compiled], |
| 1831 | parameters: Optional[_AnyMultiExecuteParams], |
| 1832 | ) -> CursorResult[Unpack[TupleAny]]: |
| 1833 | """continue the _execute_context() method for a single DBAPI |
| 1834 | cursor.execute() or cursor.executemany() call. |
| 1835 | |
| 1836 | """ |
| 1837 | if dialect.bind_typing is BindTyping.SETINPUTSIZES: |
| 1838 | generic_setinputsizes = context._prepare_set_input_sizes() |
| 1839 | |
| 1840 | if generic_setinputsizes: |
| 1841 | try: |
| 1842 | dialect.do_set_input_sizes( |
| 1843 | context.cursor, generic_setinputsizes, context |
| 1844 | ) |
| 1845 | except BaseException as e: |
| 1846 | self._handle_dbapi_exception( |
| 1847 | e, str(statement), parameters, None, context |
| 1848 | ) |
| 1849 | |
| 1850 | cursor, str_statement, parameters = ( |
| 1851 | context.cursor, |
| 1852 | context.statement, |
| 1853 | context.parameters, |
| 1854 | ) |
| 1855 | |
| 1856 | effective_parameters: Optional[_AnyExecuteParams] |
| 1857 | |
| 1858 | if not context.executemany: |
| 1859 | effective_parameters = parameters[0] |
| 1860 | else: |
| 1861 | effective_parameters = parameters |
| 1862 | |
| 1863 | evt_handled: bool = False |
| 1864 | try: |
| 1865 | if self._has_events or self.engine._has_events: |
| 1866 | for fn in self.dispatch.before_cursor_execute: |
| 1867 | str_statement, effective_parameters = fn( |
| 1868 | self, |
| 1869 | cursor, |
| 1870 | str_statement, |
| 1871 | effective_parameters, |
| 1872 | context, |
| 1873 | context.executemany, |
| 1874 | ) |
| 1875 | |
| 1876 | if self._echo: |
| 1877 | self._log_info(str_statement) |
| 1878 | |
| 1879 | stats = context._get_cache_stats() |
| 1880 | |
| 1881 | if not self.engine.hide_parameters: |
| 1882 | self._log_info( |
| 1883 | "[%s] %r", |
no test coverage detected