python - sqlalchemy column default
¿Cómo serializar el resultado de SqlAlchemy a JSON? (16)
Django tiene una buena serialización automática de modelos ORM devueltos de formato DB a JSON.
¿Cómo serializar el resultado de la consulta SQLAlchemy al formato JSON?
Intenté jsonpickle.encode pero codifica el objeto de consulta. Intenté json.dumps(items) pero vuelve
TypeError: <Product(''3'', ''some name'', ''some desc'')> is not JSON serializable
¿Es realmente tan difícil serializar objetos SQLAlchemy ORM a JSON / XML? ¿No hay ningún serializador predeterminado para eso? Es una tarea muy común serializar resultados de consultas ORM hoy en día.
Lo que necesito es solo devolver la representación de datos JSON o XML del resultado de la consulta SQLAlchemy.
El resultado de la consulta de objetos SQLAlchemy en formato JSON / XML es necesario para ser utilizado en javascript datagird (JQGrid http://www.trirand.com/blog/ )
Una implementación plana
Podrías usar algo como esto:
from sqlalchemy.ext.declarative import DeclarativeMeta
class AlchemyEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj.__class__, DeclarativeMeta):
# an SQLAlchemy class
fields = {}
for field in [x for x in dir(obj) if not x.startswith(''_'') and x != ''metadata'']:
data = obj.__getattribute__(field)
try:
json.dumps(data) # this will fail on non-encodable values, like other classes
fields[field] = data
except TypeError:
fields[field] = None
# a json-encodable dict
return fields
return json.JSONEncoder.default(self, obj)
y luego convierte a JSON usando:
c = YourAlchemyClass()
print json.dumps(c, cls=AlchemyEncoder)
Ignorará los campos que no son codificables (configúrelos en ''Ninguno'').
No expande automáticamente las relaciones (ya que esto podría conducir a auto-referencias y bucle para siempre).
Una implementación recursiva, no circular
Sin embargo, si prefieres un ciclo para siempre, puedes usar:
from sqlalchemy.ext.declarative import DeclarativeMeta
def new_alchemy_encoder():
_visited_objs = []
class AlchemyEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj.__class__, DeclarativeMeta):
# don''t re-visit self
if obj in _visited_objs:
return None
_visited_objs.append(obj)
# an SQLAlchemy class
fields = {}
for field in [x for x in dir(obj) if not x.startswith(''_'') and x != ''metadata'']:
fields[field] = obj.__getattribute__(field)
# a json-encodable dict
return fields
return json.JSONEncoder.default(self, obj)
return AlchemyEncoder
Y luego codifica objetos usando:
print json.dumps(e, cls=new_alchemy_encoder(), check_circular=False)
Esto codificaría a todos los niños, a todos sus hijos y a todos sus hijos ... Potencialmente, codifica toda su base de datos, básicamente. Cuando llegue a algo codificado antes, lo codificará como ''Ninguno''.
Una implementación recursiva, posiblemente circular, selectiva
Otra alternativa, probablemente mejor, es poder especificar los campos que desea expandir:
def new_alchemy_encoder(revisit_self = False, fields_to_expand = []):
_visited_objs = []
class AlchemyEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj.__class__, DeclarativeMeta):
# don''t re-visit self
if revisit_self:
if obj in _visited_objs:
return None
_visited_objs.append(obj)
# go through each field in this SQLalchemy class
fields = {}
for field in [x for x in dir(obj) if not x.startswith(''_'') and x != ''metadata'']:
val = obj.__getattribute__(field)
# is this field another SQLalchemy object, or a list of SQLalchemy objects?
if isinstance(val.__class__, DeclarativeMeta) or (isinstance(val, list) and len(val) > 0 and isinstance(val[0].__class__, DeclarativeMeta)):
# unless we''re expanding this field, stop here
if field not in fields_to_expand:
# not expanding this field: set it to None and continue
fields[field] = None
continue
fields[field] = val
# a json-encodable dict
return fields
return json.JSONEncoder.default(self, obj)
return AlchemyEncoder
Ahora puede llamarlo con:
print json.dumps(e, cls=new_alchemy_encoder(False, [''parents'']), check_circular=False)
Para expandir solo los campos de SQLAlchemy llamados ''padres'', por ejemplo.
Aquí hay una solución que le permite seleccionar las relaciones que desea incluir en su resultado tan profundo como le gustaría. NOTA: Esta es una reescritura completa tomando un dict / str como un arg en lugar de una lista. arregla algunas cosas ...
def deep_dict(self, relations={}):
"""Output a dict of an SA object recursing as deep as you want.
Takes one argument, relations which is a dictionary of relations we''d
like to pull out. The relations dict items can be a single relation
name or deeper relation names connected by sub dicts
Example:
Say we have a Person object with a family relationship
person.deep_dict(relations={''family'':None})
Say the family object has homes as a relation then we can do
person.deep_dict(relations={''family'':{''homes'':None}})
OR
person.deep_dict(relations={''family'':''homes''})
Say homes has a relation like rooms you can do
person.deep_dict(relations={''family'':{''homes'':''rooms''}})
and so on...
"""
mydict = dict((c, str(a)) for c, a in
self.__dict__.items() if c != ''_sa_instance_state'')
if not relations:
# just return ourselves
return mydict
# otherwise we need to go deeper
if not isinstance(relations, dict) and not isinstance(relations, str):
raise Exception("relations should be a dict, it is of type {}".format(type(relations)))
# got here so check and handle if we were passed a dict
if isinstance(relations, dict):
# we were passed deeper info
for left, right in relations.items():
myrel = getattr(self, left)
if isinstance(myrel, list):
mydict[left] = [rel.deep_dict(relations=right) for rel in myrel]
else:
mydict[left] = myrel.deep_dict(relations=right)
# if we get here check and handle if we were passed a string
elif isinstance(relations, str):
# passed a single item
myrel = getattr(self, relations)
left = relations
if isinstance(myrel, list):
mydict[left] = [rel.deep_dict(relations=None)
for rel in myrel]
else:
mydict[left] = myrel.deep_dict(relations=None)
return mydict
entonces para un ejemplo que usa persona / familia / casas / habitaciones ... convirtiéndolo en json todo lo que necesitas es
json.dumps(person.deep_dict(relations={''family'':{''homes'':''rooms''}}))
Debajo de Flask, esto funciona y maneja campos de datos y datos, transformando un campo de tipo
''time'': datetime.datetime(2018, 3, 22, 15, 40) en
"time": "2018-03-22 15:40:00" :
obj = {c.name: str(getattr(self, c.name)) for c in self.__table__.columns}
# This to get the JSON body
return json.dumps(obj)
# Or this to get a response object
return jsonify(obj)
My take utiliza (too many?) Diccionarios:
def serialize(_query):
#d = dictionary written to per row
#D = dictionary d is written to each time, then reset
#Master = dictionary of dictionaries; the id Key (int, unique from database)
from D is used as the Key for the dictionary D entry in Master
Master = {}
D = {}
x = 0
for u in _query:
d = u.__dict__
D = {}
for n in d.keys():
if n != ''_sa_instance_state'':
D[n] = d[n]
x = d[''id'']
Master[x] = D
return Master
Ejecutar con matraz (incluido jsonify) y flask_sqlalchemy para imprimir salidas como JSON.
Llame a la función con jsonify (serialize ()).
Funciona con todas las consultas de SQLAlchemy que he intentado hasta ahora (ejecutando SQLite3)
No es tan directo. Escribí un código para hacer esto. Todavía estoy trabajando en eso, y usa el framework MochiKit. Básicamente traduce objetos compuestos entre Python y Javascript usando un proxy y convertidores JSON registrados.
El lado del navegador para los objetos de la base de datos es db.js Necesita el origen del proxy de Python básico en proxy.js .
En el lado de Python está el módulo de proxy base. Luego, finalmente, el codificador de objetos SqlAlchemy en webserver.py . También depende de los extractores de metadatos que se encuentran en el archivo models.py .
Podrías simplemente generar tu objeto como un dict:
class User:
def as_dict(self):
return {c.name: getattr(self, c.name) for c in self.__table__.columns}
Y luego usa User.as_dict () para serializar su objeto.
Como se explica en Convert sqlalchemy row object to python dict
Por razones de seguridad, nunca debe devolver todos los campos del modelo. Prefiero elegirlos selectivamente.
La codificación json de Flask ahora admite UUID, datetime y relaciones (y query y query_class de query query_class para la clase db.Model de db.Model ). Actualicé el codificador de la siguiente manera:
app / json_encoder.py
from sqlalchemy.ext.declarative import DeclarativeMeta
from flask import json
class AlchemyEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o.__class__, DeclarativeMeta):
data = {}
fields = o.__json__() if hasattr(o, ''__json__'') else dir(o)
for field in [f for f in fields if not f.startswith(''_'') and f not in [''metadata'', ''query'', ''query_class'']]:
value = o.__getattribute__(field)
try:
json.dumps(value)
data[field] = value
except TypeError:
data[field] = None
return data
return json.JSONEncoder.default(self, o)
app/__init__.py
# json encoding
from app.json_encoder import AlchemyEncoder
app.json_encoder = AlchemyEncoder
Con esto puedo opcionalmente agregar una propiedad __json__ que devuelve la lista de campos que deseo codificar:
app/models.py
class Queue(db.Model):
id = db.Column(db.Integer, primary_key=True)
song_id = db.Column(db.Integer, db.ForeignKey(''song.id''), unique=True, nullable=False)
song = db.relationship(''Song'', lazy=''joined'')
type = db.Column(db.String(20), server_default=u''audio/mpeg'')
src = db.Column(db.String(255), nullable=False)
created_at = db.Column(db.DateTime, server_default=db.func.now())
updated_at = db.Column(db.DateTime, server_default=db.func.now(), onupdate=db.func.now())
def __init__(self, song):
self.song = song
self.src = song.full_path
def __json__(self):
return [''song'', ''src'', ''type'', ''created_at'']
Agrego @jsonapi a mi vista, devuelvo la lista de resultados y luego mi salida es la siguiente:
[
{
"created_at": "Thu, 23 Jul 2015 11:36:53 GMT",
"song":
{
"full_path": "/static/music/Audioslave/Audioslave [2002]/1 Cochise.mp3",
"id": 2,
"path_name": "Audioslave/Audioslave [2002]/1 Cochise.mp3"
},
"src": "/static/music/Audioslave/Audioslave [2002]/1 Cochise.mp3",
"type": "audio/mpeg"
}
]
Puede convertir un RowProxy a un dict como este:
d = dict(row.items())
A continuación, serialícelo en JSON (tendrá que especificar un codificador para cosas como valores de datetime ) No es tan difícil si solo desea un registro (y no una jerarquía completa de registros relacionados).
json.dumps([(dict(row.items())) for row in rs])
Puede usar la introspección de SqlAlchemy como esta:
mysql = SQLAlchemy()
from sqlalchemy import inspect
class Contacts(mysql.Model):
__tablename__ = ''CONTACTS''
id = mysql.Column(mysql.Integer, primary_key=True)
first_name = mysql.Column(mysql.String(128), nullable=False)
last_name = mysql.Column(mysql.String(128), nullable=False)
phone = mysql.Column(mysql.String(128), nullable=False)
email = mysql.Column(mysql.String(128), nullable=False)
street = mysql.Column(mysql.String(128), nullable=False)
zip_code = mysql.Column(mysql.String(128), nullable=False)
city = mysql.Column(mysql.String(128), nullable=False)
def toDict(self):
return { c.key: getattr(self, c.key) for c in inspect(self).mapper.column_attrs }
@app.route(''/contacts'',methods=[''GET''])
def getContacts():
contacts = Contacts.query.all()
contactsArr = []
for contact in contacts:
contactsArr.append(contact.toDict())
return jsonify(contactsArr)
@app.route(''/contacts/<int:id>'',methods=[''GET''])
def getContact(id):
contact = Contacts.query.get(id)
return jsonify(contact.toDict())
Inspírate de una respuesta aquí: convierte el objeto de fila sqlalchemy en pitón dict
Recomiendo usar un marshmallow de biblioteca reciente con superficie. Le permite crear serializadores para representar las instancias de su modelo con soporte para relaciones y objetos anidados.
Eche un vistazo a su ejemplo SQLAlchemy .
Sé que esta es una publicación bastante más antigua. Tomé la solución dada por @SashaB y la modifiqué según mi necesidad.
Le agregué las siguientes cosas:
- Lista de ignorar campos: una lista de campos a ignorar mientras se serializa
- Lista de reemplazo de campo: un diccionario que contiene los nombres de campo para ser reemplazado por valores mientras se serializa.
- Se eliminaron los métodos y BaseQuery se serializó
Mi código es el siguiente:
def alchemy_json_encoder(revisit_self = False, fields_to_expand = [], fields_to_ignore = [], fields_to_replace = {}):
"""
Serialize SQLAlchemy result into JSon
:param revisit_self: True / False
:param fields_to_expand: Fields which are to be expanded for including their children and all
:param fields_to_ignore: Fields to be ignored while encoding
:param fields_to_replace: Field keys to be replaced by values assigned in dictionary
:return: Json serialized SQLAlchemy object
"""
_visited_objs = []
class AlchemyEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj.__class__, DeclarativeMeta):
# don''t re-visit self
if revisit_self:
if obj in _visited_objs:
return None
_visited_objs.append(obj)
# go through each field in this SQLalchemy class
fields = {}
for field in [x for x in dir(obj) if not x.startswith(''_'') and x != ''metadata'' and x not in fields_to_ignore]:
val = obj.__getattribute__(field)
# is this field method defination, or an SQLalchemy object
if not hasattr(val, "__call__") and not isinstance(val, BaseQuery):
field_name = fields_to_replace[field] if field in fields_to_replace else field
# is this field another SQLalchemy object, or a list of SQLalchemy objects?
if isinstance(val.__class__, DeclarativeMeta) or /
(isinstance(val, list) and len(val) > 0 and isinstance(val[0].__class__, DeclarativeMeta)):
# unless we''re expanding this field, stop here
if field not in fields_to_expand:
# not expanding this field: set it to None and continue
fields[field_name] = None
continue
fields[field_name] = val
# a json-encodable dict
return fields
return json.JSONEncoder.default(self, obj)
return AlchemyEncoder
Espero que ayude a alguien!
Serialización personalizada y deserialización.
"from_json" (método de clase) construye un objeto Model basado en datos json.
"deserialize" solo se puede invocar en instancia y fusionar todos los datos de json en instancia de Model.
"serializar" - serialización recursiva
La propiedad __write_only__ es necesaria para definir propiedades solo de escritura ("password_hash" por ejemplo).
class Serializable(object):
__exclude__ = (''id'',)
__include__ = ()
__write_only__ = ()
@classmethod
def from_json(cls, json, selfObj=None):
if selfObj is None:
self = cls()
else:
self = selfObj
exclude = (cls.__exclude__ or ()) + Serializable.__exclude__
include = cls.__include__ or ()
if json:
for prop, value in json.iteritems():
# ignore all non user data, e.g. only
if (not (prop in exclude) | (prop in include)) and isinstance(
getattr(cls, prop, None), QueryableAttribute):
setattr(self, prop, value)
return self
def deserialize(self, json):
if not json:
return None
return self.__class__.from_json(json, selfObj=self)
@classmethod
def serialize_list(cls, object_list=[]):
output = []
for li in object_list:
if isinstance(li, Serializable):
output.append(li.serialize())
else:
output.append(li)
return output
def serialize(self, **kwargs):
# init write only props
if len(getattr(self.__class__, ''__write_only__'', ())) == 0:
self.__class__.__write_only__ = ()
dictionary = {}
expand = kwargs.get(''expand'', ()) or ()
prop = ''props''
if expand:
# expand all the fields
for key in expand:
getattr(self, key)
iterable = self.__dict__.items()
is_custom_property_set = False
# include only properties passed as parameter
if (prop in kwargs) and (kwargs.get(prop, None) is not None):
is_custom_property_set = True
iterable = kwargs.get(prop, None)
# loop trough all accessible properties
for key in iterable:
accessor = key
if isinstance(key, tuple):
accessor = key[0]
if not (accessor in self.__class__.__write_only__) and not accessor.startswith(''_''):
# force select from db to be able get relationships
if is_custom_property_set:
getattr(self, accessor, None)
if isinstance(self.__dict__.get(accessor), list):
dictionary[accessor] = self.__class__.serialize_list(object_list=self.__dict__.get(accessor))
# check if those properties are read only
elif isinstance(self.__dict__.get(accessor), Serializable):
dictionary[accessor] = self.__dict__.get(accessor).serialize()
else:
dictionary[accessor] = self.__dict__.get(accessor)
return dictionary
Una explicación más detallada. En su modelo, agregue:
def as_dict(self):
return {c.name: str(getattr(self, c.name)) for c in self.__table__.columns}
El str() es para python 3 así que si usa python 2 use unicode() . Debería ayudar a deserializar las fechas. Puede eliminarlo si no se trata de eso.
Ahora puede consultar la base de datos de esta manera
some_result = User.query.filter_by(id=current_user.id).first().as_dict()
First() es necesario para evitar errores extraños. as_dict() ahora deserializará el resultado. Después de la deserialización, está listo para ser cambiado a json
jsonify(some_result)
Use el serializador incorporado en SQLAlchemy:
from sqlalchemy.ext.serializer import loads, dumps
obj = MyAlchemyObject()
# serialize object
serialized_obj = dumps(obj)
# deserialize object
obj = loads(serialized_obj)
Si está transfiriendo el objeto entre sesiones, recuerde separar el objeto de la sesión actual usando session.expunge(obj) . Para adjuntarlo nuevamente, solo haga session.add(obj) .
Flask-JsonTools paquete Flask-JsonTools tiene una implementación de clase base JsonSerializableBase para sus modelos.
Uso:
from sqlalchemy.ext.declarative import declarative_base
from flask.ext.jsontools import JsonSerializableBase
Base = declarative_base(cls=(JsonSerializableBase,))
class User(Base):
#...
Ahora el modelo de User es mágicamente serializable.
Si su marco no es Flask, puede simplemente agarrar el código
def alc2json(row):
return dict([(col, str(getattr(row,col))) for col in row.__table__.columns.keys()])
Pensé que jugaría un pequeño código de golf con este.
FYI: Estoy usando automap_base ya que tenemos un esquema diseñado por separado de acuerdo a los requerimientos del negocio. Empecé a usar SQLAlchemy hoy, pero la documentación indica que la base_automática es una extensión de declarative_base que parece ser el paradigma típico en el SQLAlchemy ORM, así que creo que esto debería funcionar.
No es lujoso con las siguientes claves foráneas según la solución de Tjorriemorrie , pero simplemente combina columnas con valores y maneja tipos de Python mediante str () - los valores de la columna. Nuestros valores consisten en los resultados de tipo datetime.time y decimal.Decimal clase de Python para que haga el trabajo.
Espero que esto ayude a cualquier transeúnte!