Code Duplication    Length = 30-30 lines in 2 locations

pystratum_pgsql/PgSqlDataLayer.py 2 locations

@@ 193-222 (lines=30) @@
190
        return ret
191
192
    # ------------------------------------------------------------------------------------------------------------------
193
    def execute_sp_row1(self, sql: str, *params) -> Dict:
194
        """
195
        Executes a stored procedure that selects 1 row. Returns the selected row.
196
197
        :param str sql: The SQL call the the stored procedure.
198
        :param iterable params: The arguments for the stored procedure.
199
200
        :rtype: dict[str,object]
201
        """
202
        cursor = self._connection.cursor()
203
        if params:
204
            cursor.execute(sql, params)
205
        else:
206
            cursor.execute(sql)
207
        portal = self._connection.cursor(cursor.fetchone()[0])
208
        rows = portal.fetchall()
209
        self._get_column_names(portal)
210
        row_count = len(rows)
211
        if row_count == 1:
212
            column_names = self._get_column_names(portal)
213
            ret = dict(zip(column_names, rows[0]))
214
        else:
215
            ret = None  # Keep our IDE happy.
216
        portal.close()
217
        cursor.close()
218
219
        if row_count != 1:
220
            raise ResultException('1', row_count, sql)
221
222
        return ret
223
224
    # ------------------------------------------------------------------------------------------------------------------
225
    def execute_sp_rows(self, sql: str, *params) -> List:
@@ 161-190 (lines=30) @@
158
        cursor.close()
159
160
    # ------------------------------------------------------------------------------------------------------------------
161
    def execute_sp_row0(self, sql: str, *params) -> Union[None, Dict]:
162
        """
163
        Executes a stored procedure that selects 0 or 1 row. Returns the selected row or None.
164
165
        :param str sql: The SQL statement.
166
        :param iterable params: The arguments for the statement.
167
168
        :rtype: None|dict[str,object]
169
        """
170
        cursor = self._connection.cursor()
171
172
        if params:
173
            cursor.execute(sql, params)
174
        else:
175
            cursor.execute(sql)
176
        portal = self._connection.cursor(cursor.fetchone()[0])
177
        rows = portal.fetchall()
178
        row_count = len(rows)
179
        if row_count == 1:
180
            column_names = self._get_column_names(portal)
181
            ret = dict(zip(column_names, rows[0]))
182
        else:
183
            ret = None
184
        portal.close()
185
        cursor.close()
186
187
        if not (row_count == 0 or row_count == 1):
188
            raise ResultException('0 or 1', row_count, sql)
189
190
        return ret
191
192
    # ------------------------------------------------------------------------------------------------------------------
193
    def execute_sp_row1(self, sql: str, *params) -> Dict: