在python中将SQL表作为JSON返回

Question

我正在使用 web.py 中的一个小网络应用程序，并且正在设置一个 url 以返回一个 JSON 对象。使用 python 将 SQL 表转换为 JSON 的最佳方法是什么？

Answer 1

这是a pythonic way to do that 的一个非常好的例子：

import json
import psycopg2

def db(database_name='pepe'):
    return psycopg2.connect(database=database_name)

def query_db(query, args=(), one=False):
    cur = db().cursor()
    cur.execute(query, args)
    r = [dict((cur.description[i][0], value) \
               for i, value in enumerate(row)) for row in cur.fetchall()]
    cur.connection.close()
    return (r[0] if r else None) if one else r

my_query = query_db("select * from majorroadstiger limit %s", (3,))

json_output = json.dumps(my_query)

你得到一个 JSON 对象数组：

>>> json_output
'[{"divroad": "N", "featcat": null, "countyfp": "001",...

或使用以下内容：

>>> j2 = query_db("select * from majorroadstiger where fullname= %s limit %s",\
 ("Mission Blvd", 1), one=True)

你得到一个 JSON 对象：

>>> j2 = json.dumps(j2)
>>> j2
'{"divroad": "N", "featcat": null, "countyfp": "001",...

Answer 2

import sqlite3
import json

DB = "./the_database.db"

def get_all_users( json_str = False ):
    conn = sqlite3.connect( DB )
    conn.row_factory = sqlite3.Row # This enables column access by name: row['column_name'] 
    db = conn.cursor()

    rows = db.execute('''
    SELECT * from Users
    ''').fetchall()

    conn.commit()
    conn.close()

    if json_str:
        return json.dumps( [dict(ix) for ix in rows] ) #CREATE JSON

    return rows

调用没有json的方法...

print get_all_users()

打印：

[(1, u'orvar', u'password123'), (2, u'kalle', u'password123')]

用json调用方法...

print get_all_users( json_str = True )

打印：

[{"password": "password123", "id": 1, "name": "orvar"}, {"password": "password123", "id": 2, "name": "kalle"}]

Answer 3

就我个人而言，我更喜欢SQLObject。我修改了一些我必须得到的快速而肮脏的测试代码：

import simplejson

from sqlobject import *

# Replace this with the URI for your actual database
connection = connectionForURI('sqlite:/:memory:')
sqlhub.processConnection = connection

# This defines the columns for your database table. See SQLObject docs for how it
# does its conversions for class attributes <-> database columns (underscores to camel
# case, generally)

class Song(SQLObject):

    name = StringCol()
    artist = StringCol()
    album = StringCol()

# Create fake data for demo - this is not needed for the real thing
def MakeFakeDB():
    Song.createTable()
    s1 = Song(name="B Song",
              artist="Artist1",
              album="Album1")
    s2 = Song(name="A Song",
              artist="Artist2",
              album="Album2")

def Main():
    # This is an iterable, not a list
    all_songs = Song.select().orderBy(Song.q.name)

    songs_as_dict = []

    for song in all_songs:
        song_as_dict = {
            'name' : song.name,
            'artist' : song.artist,
            'album' : song.album}
        songs_as_dict.append(song_as_dict)

    print simplejson.dumps(songs_as_dict)


if __name__ == "__main__":
    MakeFakeDB()
    Main()

Answer 4

有关在传输数据之前如何处理数据的更多信息将大有帮助。 json 模块提供了 dump(s) 和 load(s) 方法，如果您使用的是 2.6 或更高版本，它们会有所帮助：http://docs.python.org/library/json.html。

-- 已编辑--

在不知道您正在使用哪些库的情况下，我无法确定您是否会找到这样的方法。通常，我会像这样处理查询结果（以 kinterbasdb 为例，因为这是我们目前正在使用的）：

qry = "Select Id, Name, Artist, Album From MP3s Order By Name, Artist"
# Assumes conn is a database connection.
cursor = conn.cursor()
cursor.execute(qry)
rows = [x for x in cursor]
cols = [x[0] for x in cursor.description]
songs = []
for row in rows:
  song = {}
  for prop, val in zip(cols, row):
    song[prop] = val
  songs.append(song)
# Create a string representation of your array of songs.
songsJSON = json.dumps(songs)

毫无疑问，有更好的专家可以通过列表解析来消除对写出循环的需要，但这很有效，并且应该是您可以适应任何用于检索记录的库的东西。

Answer 5

我拼凑了一个简短的脚本，它将所有表中的所有数据转储为列名的字典：值。与其他解决方案不同，它不需要任何有关表或列的信息，它只是找到所有内容并将其转储。希望有人觉得它有用！

from contextlib import closing
from datetime import datetime
import json
import MySQLdb
DB_NAME = 'x'
DB_USER = 'y'
DB_PASS = 'z'

def get_tables(cursor):
    cursor.execute('SHOW tables')
    return [r[0] for r in cursor.fetchall()] 

def get_rows_as_dicts(cursor, table):
    cursor.execute('select * from {}'.format(table))
    columns = [d[0] for d in cursor.description]
    return [dict(zip(columns, row)) for row in cursor.fetchall()]
 
def dump_date(thing):
    if isinstance(thing, datetime):
        return thing.isoformat()
    return str(thing)


with closing(MySQLdb.connect(user=DB_USER, passwd=DB_PASS, db=DB_NAME)) as conn, closing(conn.cursor()) as cursor:
    dump = {
        table: get_rows_as_dicts(cursor, table)
        for table in get_tables(cursor)
    }
    print(json.dumps(dump, default=dump_date, indent=2))

Answer 6

最简单的方法，

使用json.dumps，但如果它的日期时间需要将日期时间解析为json序列化器。

这是我的，

import MySQLdb, re, json
from datetime import date, datetime

def json_serial(obj):
    """JSON serializer for objects not serializable by default json code"""

    if isinstance(obj, (datetime, date)):
        return obj.isoformat()
    raise TypeError ("Type %s not serializable" % type(obj))

conn = MySQLdb.connect(instance)
curr = conn.cursor()
curr.execute("SELECT * FROM `assets`")
data = curr.fetchall()
print json.dumps(data, default=json_serial)

它将返回 json 转储

另一种没有 json 转储的简单方法， 这里获取标头并使用 zip 映射每个最终都将其设置为 json 但这不是将日期时间更改为 json 序列化程序...

data_json = []
header = [i[0] for i in curr.description]
data = curr.fetchall()
for i in data:
    data_json.append(dict(zip(header, i)))
print data_json

Answer 7

似乎没有人提供使用 postgres JSON 功能直接从 Postgresql 服务器获取 JSON 的选项 https://www.postgresql.org/docs/9.4/static/functions-json.html

在 python 端没有解析、循环或任何内存消耗，如果您要处理 100,000 行或数百万行，您可能真的需要考虑这些。

from django.db import connection

sql = 'SELECT to_json(result) FROM (SELECT * FROM TABLE table) result)'
with connection.cursor() as cursor:
  cursor.execute(sql)
  output = cursor.fetchall()

像这样的表格：

id, value
----------
1     3
2     7

将返回一个 Python JSON 对象

[{"id": 1, "value": 3},{"id":2, "value": 7}]

然后使用json.dumps 转储为 JSON 字符串

Answer 8

我会用 psycopg2 版本补充The Demz 答案：

import psycopg2 
import psycopg2.extras
import json
connection = psycopg2.connect(dbname=_cdatabase, host=_chost, port=_cport , user=_cuser, password=_cpassword)
cursor = connection.cursor(cursor_factory=psycopg2.extras.DictCursor) # This line allows dictionary access.
#select some records into "rows"
jsonout= json.dumps([dict(ix) for ix in rows])

Answer 9

如果您使用的是 MSSQL Server 2008 及更高版本，则可以使用 FOR JSON AUTO 子句 E.G 执行 SELECT 查询以返回 json

SELECT name, surname FROM users FOR JSON AUTO

将返回 Json 为

[{"name": "Jane","surname": "Doe" }, {"name": "Foo","surname": "Samantha" }, ..., {"name": "John", "surname": "boo" }]

Answer 10

10 年后 :) 。没有列表理解

从如下所示的选择查询中返回单行值。

"select name,userid, address from table1 where userid = 1"

json 输出

{ name : "name1", userid : 1, address : "adress1, street1" }

代码

cur.execute(f"select name,userid, address from table1 where userid = 1 ")
row = cur.fetchone()
desc = list(zip(*cur.description))[0]  #To get column names
rowdict = dict(zip(desc,row))
jsondict = jsonify(rowdict)  #Flask jsonify

cur.description 是一个元组，如下所示。 unzip 和 zip 将列名与值结合起来

(('name', None, None, None, None, None, None), ('userid', None, None, None, None, None, None), ('address', None, None, None, None, None, None))

Answer 11

from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import String

Base = declarative_base()
metadata = Base.metadata


class UserTable(Base):
    __tablename__ = 'UserTable'

    Id = Column("ID", Integer, primary_key=True)
    Name = Column("Name", String(100))

        
class UserTableDTO:
    def __init__(self, ob):
        self.Id = ob.Id
        self.Name = ob.Name
        
rows = dbsession.query(Table).all()

json_string = [json.loads(json.dumps(UserTableDTO(ob).__dict__, default=lambda x: str(x)))for ob in rows]
print(json_string)

Answer 12

将 SQL 表返回为 formatted JSON 并修复错误的一个简单示例，就像他有 @Whitecat 一样

我收到错误 datetime.datetime(1941, 10, 31, 0, 0) is not JSON serializable

在该示例中，您应该使用JSONEncoder。

import json
import pymssql

# subclass JSONEncoder
class DateTimeEncoder(JSONEncoder):
        #Override the default method
        def default(self, obj):
            if isinstance(obj, (datetime.date, datetime.datetime)):
                return obj.isoformat()

def mssql_connection():
    try:
        return pymssql.connect(server="IP.COM", user="USERNAME", password="PASSWORD", database="DATABASE")
    except Exception:
        print("\nERROR: Unable to connect to the server.")
        exit(-1)

def query_db(query):
    cur = mssql_connection().cursor()
    cur.execute(query)
    r = [dict((cur.description[i][0], value) for i, value in enumerate(row)) for row in cur.fetchall()]
    cur.connection.close()
    return r

def write_json(query_path):
    # read sql from file
    with open("../sql/my_sql.txt", 'r') as f:
        sql = f.read().replace('\n', ' ')
    # creating and writing to a json file and Encode DateTime Object into JSON using custom JSONEncoder
    with open("../output/my_json.json", 'w', encoding='utf-8') as f:
        json.dump(query_db(sql), f, ensure_ascii=False, indent=4, cls=DateTimeEncoder) 

if __name__ == "__main__":
    write_json()

# You get formatted my_json.json, for example:
[
   {
      "divroad":"N",
      "featcat":null,
      "countyfp":"001",
      "date":"2020-08-28"
   }
]

Answer 13

对于 sqlite，可以设置一个可调用的 connection.row_factory 并将查询结果的格式更改为 python 字典对象。请参阅documentation。这是一个例子：

import sqlite3, json

def dict_factory(cursor, row):
    d = {}
    for idx, col in enumerate(cursor.description):
        # col[0] is the column name
        d[col[0]] = row[idx]
    return d

def get_data_to_json():
    conn = sqlite3.connect("database.db")
    conn.row_factory = dict_factory
    c = conn.cursor()
    c.execute("SELECT * FROM table")
    rst = c.fetchall() # rst is a list of dict
    return jsonify(rst)