Code Duplication    Length = 26-26 lines in 2 locations

pystratum_mysql/MySqlDataLayer.py 2 locations

@@ 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.