Sortie des résultats du curseur pyodbc sous forme de dictionnaire python

Question

Comment puis-je sérialiser la sortie du curseur de pyodbc (à partir de .fetchone , .fetchmany o .fetchall ) comme un dictionnaire Python ?

J'utilise bottlepy et j'ai besoin de retourner un dict pour qu'il puisse le retourner en JSON.

Answer 1

Ce dont j'avais besoin, ce qui est légèrement différent de ce que l'OP demandait :
Si vous voulez généraliser complètement une routine qui exécute des requêtes SQL Select, mais que vous devez référencer les résultats par un numéro d'index, et non par un nom, vous pouvez le faire, avec une liste de listes au lieu d'un dictionnaire.

Chaque ligne de données retournée est représentée dans la liste retournée comme une liste de valeurs de champs (colonnes).
Les noms des colonnes peuvent être fournis comme première entrée de la liste retournée, de sorte que l'analyse syntaxique de la liste retournée dans la routine d'appel peut être vraiment facile et flexible.
De cette façon, la routine qui effectue l'appel à la base de données n'a pas besoin de savoir quoi que ce soit sur les données qu'elle manipule. Voici une telle routine :

    def read_DB_Records(self, tablename, fieldlist, wherefield, wherevalue) -> list:

        DBfile = 'C:/DATA/MyDatabase.accdb'
        # this connection string is for Access 2007, 2010 or later .accdb files
        conn = pyodbc.connect(r'Driver={Microsoft Access Driver (*.mdb, *.accdb)};DBQ='+DBfile)
        cursor = conn.cursor()

        # Build the SQL Query string using the passed-in field list:
        SQL = "SELECT "
        for i in range(0, len(fieldlist)):
            SQL = SQL + "[" + fieldlist[i] + "]"
            if i < (len(fieldlist)-1):
                SQL = SQL + ", "
        SQL = SQL + " FROM " + tablename

        # Support an optional WHERE clause:
        if wherefield != "" and wherevalue != "" :
            SQL = SQL + " WHERE [" + wherefield + "] = " + "'" + wherevalue + "';"

        results = []    # Create the results list object

        cursor.execute(SQL) # Execute the Query

        # (Optional) Get a list of the column names returned from the query:
        columns = [column[0] for column in cursor.description]
        results.append(columns) # append the column names to the return list

        # Now add each row as a list of column data to the results list
        for row in cursor.fetchall():   # iterate over the cursor
            results.append(list(row))   # add the row as a list to the list of lists

        cursor.close()  # close the cursor
        conn.close()    # close the DB connection

        return results  # return the list of lists

Answer 2

J'aime les réponses de @bryan et @foo-stack. Si vous travaillez avec postgresql et que vous utilisez psycopg2 vous pourriez utiliser goodies de psycopg2 de réaliser la même chose en spécifiant que le curseur est un DictCursor lorsque vous créez votre curseur à partir de la connexion, comme ceci :

cur = conn.cursor( cursor_factory=psycopg2.extras.DictCursor )

Vous pouvez donc maintenant exécuter votre requête sql et vous obtiendrez un dictionnaire pour récupérer vos résultats, sans avoir besoin de les mapper à la main.

cur.execute( sql_query )
results = cur.fetchall()

for row in results:
    print row['row_no']

Veuillez noter que vous devrez import psycopg2.extras pour que ça marche.

Answer 3

Les étapes sont les suivantes :

1. Import Libs :

   from pandas import DataFrame
   import pyodbc
   import sqlalchemy

2. Obtenez vos résultats à partir de la base de données locale :

   db_file = r'xxx.accdb' odbc_conn_str = 'DRIVER={Microsoft Access Driver (.mdb, .accdb)};DBQ=%s' % (db_file)

   conn = pyodbc.connect(odbc_conn_str) cur = conn.cursor() qry = cur.execute("SELECT * FROM tbl") columns = [column[0] for column in cur.description]

   results = [] for row in cur.fetchall():

   results.append(dict(zip(columns, row))) df = DataFrame(results) df

Answer 4

En supposant que vous connaissez les noms de vos colonnes ! Voici également trois solutions différentes,
vous voulez probablement regarder le dernier !

colnames = ['city', 'area', 'street']
data = {}

counter = 0
for row in x.fetchall():
    if not counter in data:
        data[counter] = {}

    colcounter = 0
    for colname in colnames:
        data[counter][colname] = row[colcounter]
        colcounter += 1

    counter += 1

Il s'agit d'une version indexée. Ce n'est pas la plus belle solution, mais elle fonctionne. Une autre solution serait d'indexer le nom de la colonne comme clé de dictionnaire avec une liste dans chaque clé contenant les données dans l'ordre du numéro de ligne. en faisant :

colnames = ['city', 'area', 'street']
data = {}

for row in x.fetchall():
    colindex = 0
    for col in colnames:
        if not col in data:
            data[col] = []
        data[col].append(row[colindex])
        colindex += 1

En écrivant ceci, je comprends que faire for col in colnames pourrait être remplacé par for colindex in range(0, len()) mais vous voyez l'idée. Le dernier exemple serait utile lorsqu'il ne s'agit pas de récupérer toutes les données, mais une ligne à la fois, par exemple :

Utilisation d'un dict pour chaque ligne de données

def fetchone_dict(stuff):
    colnames = ['city', 'area', 'street']
    data = {}

    for colindex in range(0, colnames):
        data[colnames[colindex]] = stuff[colindex]
    return data

row = x.fetchone()
print fetchone_dict(row)['city']

---

Récupération des noms de table (je pense grâce à Foo Stack) :
une solution plus directe de beargle ci-dessous !

cursor.execute("SELECT sys.objects.name, sys.columns.name FROM sys.objects INNER JOIN sys.columns ON sys.objects.object_id = sys.columns. object_id WHERE sys.objects.type = 'U';")
schema = {}
for it in cursor.fetchall():
    if it[0] in schema:
       schema[it[0]].append(it[1])
    else:
        schema[it[0]] = [it[1]]

Answer 5

Je sais que c'est vieux et je ne fais que répéter ce que d'autres ont déjà dit. Mais j'ai trouvé cette méthode géniale, car elle est aussi sans danger pour les injections.

def to_dict(row):
    return dict(zip([t[0] for t in row.cursor_description], row))

def query(cursor, query, params=[], cursor_func=to_dict):
    cursor.execute(query, params) 
    results = [cursor_func(row) for row in cursor.fetchall()]
    return results

quotes = query(cursor, "select * from currency where abbreviation like ?", ["USD"])