Execute the given SQL statement, with optional parameters.
(self, sql, params=())
| 177 | # Core utility functions |
| 178 | |
| 179 | def execute(self, sql, params=()): |
| 180 | """Execute the given SQL statement, with optional parameters.""" |
| 181 | # Don't perform the transactional DDL check if SQL is being collected |
| 182 | # as it's not going to be executed anyway. |
| 183 | if ( |
| 184 | not self.collect_sql |
| 185 | and self.connection.in_atomic_block |
| 186 | and not self.connection.features.can_rollback_ddl |
| 187 | ): |
| 188 | raise TransactionManagementError( |
| 189 | "Executing DDL statements while in a transaction on databases " |
| 190 | "that can't perform a rollback is prohibited." |
| 191 | ) |
| 192 | # Account for non-string statement objects. |
| 193 | sql = str(sql) |
| 194 | # Log the command we're running, then run it |
| 195 | logger.debug( |
| 196 | "%s; (params %r)", sql, params, extra={"params": params, "sql": sql} |
| 197 | ) |
| 198 | if self.collect_sql: |
| 199 | ending = "" if sql.rstrip().endswith(";") else ";" |
| 200 | if params is not None: |
| 201 | self.collected_sql.append( |
| 202 | (sql % tuple(map(self.quote_value, params))) + ending |
| 203 | ) |
| 204 | else: |
| 205 | self.collected_sql.append(sql + ending) |
| 206 | else: |
| 207 | with self.connection.cursor() as cursor: |
| 208 | cursor.execute(sql, params) |
| 209 | |
| 210 | def quote_name(self, name): |
| 211 | return self.connection.ops.quote_name(name) |
no test coverage detected