reconstruct the statement and params in terms of a target dialect, which for CompiledSQL is just DefaultDialect.
(self, execute_observed)
| 107 | return url.URL.create(self.dialect).get_dialect()() |
| 108 | |
| 109 | def _received_statement(self, execute_observed): |
| 110 | """reconstruct the statement and params in terms |
| 111 | of a target dialect, which for CompiledSQL is just DefaultDialect.""" |
| 112 | |
| 113 | context = execute_observed.context |
| 114 | compare_dialect = self._compile_dialect(execute_observed) |
| 115 | |
| 116 | # received_statement runs a full compile(). we should not need to |
| 117 | # consider extracted_parameters; if we do this indicates some state |
| 118 | # is being sent from a previous cached query, which some misbehaviors |
| 119 | # in the ORM can cause, see #6881 |
| 120 | cache_key = None # execute_observed.context.compiled.cache_key |
| 121 | extracted_parameters = ( |
| 122 | None # execute_observed.context.extracted_parameters |
| 123 | ) |
| 124 | |
| 125 | if "schema_translate_map" in context.execution_options: |
| 126 | map_ = context.execution_options["schema_translate_map"] |
| 127 | else: |
| 128 | map_ = None |
| 129 | |
| 130 | if isinstance(execute_observed.clauseelement, BaseDDLElement): |
| 131 | compiled = execute_observed.clauseelement.compile( |
| 132 | dialect=compare_dialect, |
| 133 | schema_translate_map=map_, |
| 134 | ) |
| 135 | else: |
| 136 | compiled = execute_observed.clauseelement.compile( |
| 137 | cache_key=cache_key, |
| 138 | dialect=compare_dialect, |
| 139 | column_keys=context.compiled.column_keys, |
| 140 | for_executemany=context.compiled.for_executemany, |
| 141 | schema_translate_map=map_, |
| 142 | ) |
| 143 | _received_statement = re.sub(r"[\n\t]", "", str(compiled)) |
| 144 | parameters = execute_observed.parameters |
| 145 | |
| 146 | if not parameters: |
| 147 | _received_parameters = [ |
| 148 | compiled.construct_params( |
| 149 | extracted_parameters=extracted_parameters |
| 150 | ) |
| 151 | ] |
| 152 | else: |
| 153 | _received_parameters = [ |
| 154 | compiled.construct_params( |
| 155 | m, extracted_parameters=extracted_parameters |
| 156 | ) |
| 157 | for m in parameters |
| 158 | ] |
| 159 | |
| 160 | return _received_statement, _received_parameters |
| 161 | |
| 162 | def process_statement(self, execute_observed): |
| 163 | context = execute_observed.context |
no test coverage detected