Code Duplication    Length = 26-26 lines in 2 locations

pystratum_mysql/MySqlDataLayer.py 2 locations

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