Code Duplication    Length = 26-27 lines in 2 locations

pystratum_mysql/MySqlDataLayer.py 2 locations

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