Code Duplication    Length = 26-27 lines in 2 locations

pystratum_mysql/MySqlDataLayer.py 2 locations

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