@@ 282-307 (lines=26) @@ | ||
279 | return ret |
|
280 | ||
281 | # ------------------------------------------------------------------------------------------------------------------ |
|
282 | def execute_sp_row1(self, sql: str, *params) -> Dict[str, Any]: |
|
283 | """ |
|
284 | Executes a stored procedure that selects 1 row. Returns the selected row. |
|
285 | ||
286 | :param sql: The SQL code to execute the stored procedure. |
|
287 | :param params: The arguments for the stored procedure. |
|
288 | """ |
|
289 | self._last_sql = sql |
|
290 | ||
291 | cursor = MySQLCursorBufferedDict(self._connection) |
|
292 | itr = cursor.execute(sql, params, multi=True) |
|
293 | result = itr.__next__() |
|
294 | rowcount = result.rowcount |
|
295 | if rowcount == 1: |
|
296 | ret = result.fetchone() |
|
297 | else: |
|
298 | ret = None # Keep our IDE happy. |
|
299 | itr.__next__() |
|
300 | cursor.close() |
|
301 | ||
302 | if rowcount != 1: |
|
303 | raise ResultException('1', rowcount, sql) |
|
304 | ||
305 | return ret |
|
306 | ||
307 | # ------------------------------------------------------------------------------------------------------------------ |
|
308 | def execute_sp_rows(self, sql: str, *params) -> List[Dict[str, Any]]: |
|
309 | """ |
|
310 | Executes a stored procedure that selects 0 or more rows. Returns the selected rows (an empty list if no rows |
|
@@ 256-281 (lines=26) @@ | ||
253 | return rowcount |
|
254 | ||
255 | # ------------------------------------------------------------------------------------------------------------------ |
|
256 | def execute_sp_row0(self, sql: str, *params) -> Optional[Dict[str, Any]]: |
|
257 | """ |
|
258 | Executes a stored procedure that selects 0 or 1 row. Returns the selected row or None. |
|
259 | ||
260 | :param sql: The SQL code to execute the stored procedure. |
|
261 | :param params: The arguments for the stored procedure. |
|
262 | """ |
|
263 | self._last_sql = sql |
|
264 | ||
265 | cursor = MySQLCursorBufferedDict(self._connection) |
|
266 | itr = cursor.execute(sql, params, multi=True) |
|
267 | result = itr.__next__() |
|
268 | rowcount = result.rowcount |
|
269 | if rowcount == 1: |
|
270 | ret = result.fetchone() |
|
271 | else: |
|
272 | ret = None |
|
273 | itr.__next__() |
|
274 | cursor.close() |
|
275 | ||
276 | if not (rowcount == 0 or rowcount == 1): |
|
277 | raise ResultException('0 or 1', rowcount, sql) |
|
278 | ||
279 | return ret |
|
280 | ||
281 | # ------------------------------------------------------------------------------------------------------------------ |
|
282 | def execute_sp_row1(self, sql: str, *params) -> Dict[str, Any]: |
|
283 | """ |
|
284 | Executes a stored procedure that selects 1 row. Returns the selected row. |