Code Duplication    Length = 26-26 lines in 2 locations

pystratum_mysql/MySqlDataLayer.py 2 locations

@@ 297-322 (lines=26) @@
294
        return ret
295
296
    # ------------------------------------------------------------------------------------------------------------------
297
    def execute_sp_row1(self, sql: str, *params) -> Dict[str, Any]:
298
        """
299
        Executes a stored procedure that selects 1 row. Returns the selected row.
300
301
        :param str sql: The SQL calling the the stored procedure.
302
        :param iterable params: The arguments for the stored procedure.
303
304
        :rtype: dict[str,*]
305
        """
306
        self._last_sql = sql
307
308
        cursor = MySQLCursorBufferedDict(self._connection)
309
        itr = cursor.execute(sql, params, multi=True)
310
        result = itr.__next__()
311
        rowcount = result.rowcount
312
        if rowcount == 1:
313
            ret = result.fetchone()
314
        else:
315
            ret = None  # Keep our IDE happy.
316
        itr.__next__()
317
        cursor.close()
318
319
        if rowcount != 1:
320
            raise ResultException('1', rowcount, sql)
321
322
        return ret
323
324
    # ------------------------------------------------------------------------------------------------------------------
325
    def execute_sp_rows(self, sql: str, *params) -> List[Dict[str, Any]]:
@@ 269-294 (lines=26) @@
266
        return rowcount
267
268
    # ------------------------------------------------------------------------------------------------------------------
269
    def execute_sp_row0(self, sql: str, *params) -> Optional[Dict[str, Any]]:
270
        """
271
        Executes a stored procedure that selects 0 or 1 row. Returns the selected row or None.
272
273
        :param str sql: The SQL call the the stored procedure.
274
        :param iterable params: The arguments for the stored procedure.
275
276
        :rtype: None|dict[str,*]
277
        """
278
        self._last_sql = sql
279
280
        cursor = MySQLCursorBufferedDict(self._connection)
281
        itr = cursor.execute(sql, params, multi=True)
282
        result = itr.__next__()
283
        rowcount = result.rowcount
284
        if rowcount == 1:
285
            ret = result.fetchone()
286
        else:
287
            ret = None
288
        itr.__next__()
289
        cursor.close()
290
291
        if not (rowcount == 0 or rowcount == 1):
292
            raise ResultException('0 or 1', rowcount, sql)
293
294
        return ret
295
296
    # ------------------------------------------------------------------------------------------------------------------
297
    def execute_sp_row1(self, sql: str, *params) -> Dict[str, Any]: