PgDB: fix doSelect() method to retreive list of dicts instead of list of lists.

This commit is contained in:
Benjamin Renard 2023-12-15 12:12:48 +01:00
parent dcaec24ea4
commit 371d194728
Signed by: bn8
GPG key ID: 3E2E1CE1907115BC
2 changed files with 11 additions and 7 deletions

View file

@ -5,6 +5,7 @@ import logging
import sys import sys
import psycopg2 import psycopg2
from psycopg2.extras import RealDictCursor
from mylib.db import DB, DBFailToConnect from mylib.db import DB, DBFailToConnect
@ -114,20 +115,16 @@ class PgDB(DB):
:return: List of selected rows as dict on success, False otherwise :return: List of selected rows as dict on success, False otherwise
:rtype: list, bool :rtype: list, bool
""" """
cursor = self._conn.cursor() cursor = self._conn.cursor(cursor_factory=RealDictCursor)
try: try:
self._log_query(sql, params) self._log_query(sql, params)
cursor.execute(sql, params) cursor.execute(sql, params)
results = cursor.fetchall() results = cursor.fetchall()
return results return list(map(dict, results))
except psycopg2.Error: except psycopg2.Error:
self._log_query_exception(sql, params) self._log_query_exception(sql, params)
return False return False
@staticmethod
def _map_row_fields_by_index(fields, row):
return {field: row[idx] for idx, field in enumerate(fields)}
# #
# Depreated helpers # Depreated helpers
# #

View file

@ -3,6 +3,7 @@
import psycopg2 import psycopg2
import pytest import pytest
from psycopg2.extras import RealDictCursor
from mylib.pgsql import PgDB from mylib.pgsql import PgDB
@ -57,6 +58,7 @@ class FakePsycopg2:
expected_sql = None expected_sql = None
expected_params = None expected_params = None
expected_cursor_factory = None
expected_return = True expected_return = True
expected_just_try = False expected_just_try = False
expected_exception = False expected_exception = False
@ -81,7 +83,8 @@ class FakePsycopg2:
raise psycopg2.Error(f"set_client_encoding({arg[0]}): Expected exception") raise psycopg2.Error(f"set_client_encoding({arg[0]}): Expected exception")
return self.expected_return return self.expected_return
def cursor(self): def cursor(self, cursor_factory=None):
assert cursor_factory is self.expected_cursor_factory
return FakePsycopg2Cursor( return FakePsycopg2Cursor(
self.expected_sql, self.expected_sql,
self.expected_params, self.expected_params,
@ -455,6 +458,7 @@ def test_doSQL_on_exception(fake_connected_pgdb):
def test_doSelect(fake_connected_pgdb): def test_doSelect(fake_connected_pgdb):
fake_connected_pgdb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s" fake_connected_pgdb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s"
fake_connected_pgdb._conn.expected_params = {"test1": 1} fake_connected_pgdb._conn.expected_params = {"test1": 1}
fake_connected_pgdb._conn.expected_cursor_factory = RealDictCursor
fake_connected_pgdb._conn.expected_return = [{"test1": 1}] fake_connected_pgdb._conn.expected_return = [{"test1": 1}]
assert ( assert (
fake_connected_pgdb.doSelect( fake_connected_pgdb.doSelect(
@ -466,6 +470,7 @@ def test_doSelect(fake_connected_pgdb):
def test_doSelect_without_params(fake_connected_pgdb): def test_doSelect_without_params(fake_connected_pgdb):
fake_connected_pgdb._conn.expected_sql = "SELECT * FROM table" fake_connected_pgdb._conn.expected_sql = "SELECT * FROM table"
fake_connected_pgdb._conn.expected_cursor_factory = RealDictCursor
fake_connected_pgdb._conn.expected_return = [{"test1": 1}] fake_connected_pgdb._conn.expected_return = [{"test1": 1}]
assert ( assert (
fake_connected_pgdb.doSelect(fake_connected_pgdb._conn.expected_sql) fake_connected_pgdb.doSelect(fake_connected_pgdb._conn.expected_sql)
@ -474,6 +479,7 @@ def test_doSelect_without_params(fake_connected_pgdb):
def test_doSelect_on_exception(fake_connected_pgdb): def test_doSelect_on_exception(fake_connected_pgdb):
fake_connected_pgdb._conn.expected_cursor_factory = RealDictCursor
fake_connected_pgdb._conn.expected_exception = True fake_connected_pgdb._conn.expected_exception = True
assert fake_connected_pgdb.doSelect("SELECT * FROM table") is False assert fake_connected_pgdb.doSelect("SELECT * FROM table") is False
@ -481,6 +487,7 @@ def test_doSelect_on_exception(fake_connected_pgdb):
def test_doSelect_just_try(fake_connected_just_try_pgdb): def test_doSelect_just_try(fake_connected_just_try_pgdb):
fake_connected_just_try_pgdb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s" fake_connected_just_try_pgdb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s"
fake_connected_just_try_pgdb._conn.expected_params = {"test1": 1} fake_connected_just_try_pgdb._conn.expected_params = {"test1": 1}
fake_connected_just_try_pgdb._conn.expected_cursor_factory = RealDictCursor
fake_connected_just_try_pgdb._conn.expected_return = [{"test1": 1}] fake_connected_just_try_pgdb._conn.expected_return = [{"test1": 1}]
assert ( assert (
fake_connected_just_try_pgdb.doSelect( fake_connected_just_try_pgdb.doSelect(