Code Duplication    Length = 26-27 lines in 2 locations

pystratum_mysql/MySqlDataLayer.py 2 locations

@@ 374-400 (lines=27) @@
371
        return ret
372
373
    # ------------------------------------------------------------------------------------------------------------------
374
    def execute_sp_singleton1(self, sql: str, *params) -> Any:
375
        """
376
        Executes a stored routine with designation type "table", i.e a stored routine that is expected to select 1 row
377
        with 1 column.
378
379
        :param str sql: The SQL calling the the stored procedure.
380
        :param iterable params: The arguments for the stored procedure.
381
382
        :rtype: * The value of the selected column.
383
        """
384
        self._last_sql = sql
385
386
        cursor = MySQLCursorBuffered(self._connection)
387
        itr = cursor.execute(sql, params, multi=True)
388
        result = itr.__next__()
389
        rowcount = result.rowcount
390
        if rowcount == 1:
391
            ret = result.fetchone()[0]
392
        else:
393
            ret = None  # Keep our IDE happy.
394
        itr.__next__()
395
        cursor.close()
396
397
        if rowcount != 1:
398
            raise ResultException('1', rowcount, sql)
399
400
        return ret
401
402
    # ------------------------------------------------------------------------------------------------------------------
403
    def is_alive(self) -> bool:
@@ 346-371 (lines=26) @@
343
        return ret
344
345
    # ------------------------------------------------------------------------------------------------------------------
346
    def execute_sp_singleton0(self, sql: str, *params) -> Any:
347
        """
348
        Executes a stored procedure that selects 0 or 1 row with 1 column. Returns the value of selected column or None.
349
350
        :param str sql: The SQL calling the stored procedure.
351
        :param iterable params: The arguments for the stored procedure.
352
353
        :rtype: *
354
        """
355
        self._last_sql = sql
356
357
        cursor = MySQLCursorBuffered(self._connection)
358
        itr = cursor.execute(sql, params, multi=True)
359
        result = itr.__next__()
360
        rowcount = result.rowcount
361
        if rowcount == 1:
362
            ret = result.fetchone()[0]
363
        else:
364
            ret = None
365
        itr.__next__()
366
        cursor.close()
367
368
        if not (rowcount == 0 or rowcount == 1):
369
            raise ResultException('0 or 1', rowcount, sql)
370
371
        return ret
372
373
    # ------------------------------------------------------------------------------------------------------------------
374
    def execute_sp_singleton1(self, sql: str, *params) -> Any: