| 214 | self.lastrowid: Optional[int] = None |
| 215 | |
| 216 | def execute(self, sql: str, params: Optional[Sequence[Any]] = None): |
| 217 | self.lastrowid = None |
| 218 | |
| 219 | if self._backend == "postgres": |
| 220 | query = _adapt_sql_for_postgres(sql) |
| 221 | should_capture_id = _should_append_returning_id(query) |
| 222 | if should_capture_id: |
| 223 | query = f"{query.strip().rstrip(';')} RETURNING id" |
| 224 | self._cursor.execute(query, tuple(params or ())) |
| 225 | if should_capture_id: |
| 226 | row = self._cursor.fetchone() |
| 227 | if row is not None: |
| 228 | self.lastrowid = int(row["id"] if isinstance(row, dict) else row[0]) |
| 229 | return self |
| 230 | |
| 231 | if params is None: |
| 232 | self._cursor.execute(sql) |
| 233 | else: |
| 234 | self._cursor.execute(sql, tuple(params)) |
| 235 | self.lastrowid = getattr(self._cursor, "lastrowid", None) |
| 236 | return self |
| 237 | |
| 238 | def executemany(self, sql: str, seq_of_params: Iterable[Sequence[Any]]): |
| 239 | self.lastrowid = None |