python - tutorial - sqlalchemy delete object

jsonify un resultado de SQLAlchemy en Frasco

Question

Intento conjurar un resultado de SQLAlchemy en Flask / Python.

La lista de correo de Flask sugirió el siguiente método http://librelist.com/browser//flask/2011/2/16/jsonify-sqlalchemy-pagination-collection-result/#04a0754b63387f87e59dda564bde426e :

return jsonify(json_list = qryresult)

Sin embargo, me aparece el siguiente error:

TypeError: <flaskext.sqlalchemy.BaseQuery object at 0x102c2df90> 
is not JSON serializable

¿Qué estoy pasando por alto aquí?

He encontrado esta pregunta: ¿Cómo serializar el resultado de SqlAlchemy a JSON? que parece muy similar, sin embargo, no sabía si Flask tenía algo de magia para hacerlo más fácil, como sugería la publicación de la lista de correo.

Editar: para aclarar, así es como se ve mi modelo

class Rating(db.Model):

    __tablename__ = ''rating''

    id = db.Column(db.Integer, primary_key=True)
    fullurl = db.Column(db.String())
    url = db.Column(db.String())
    comments = db.Column(db.Text)
    overall = db.Column(db.Integer)
    shipping = db.Column(db.Integer)
    cost = db.Column(db.Integer)
    honesty = db.Column(db.Integer)
    communication = db.Column(db.Integer)
    name = db.Column(db.String())
    ipaddr = db.Column(db.String())
    date = db.Column(db.String())

    def __init__(self, fullurl, url, comments, overall, shipping, cost, honesty, communication, name, ipaddr, date):
        self.fullurl = fullurl
        self.url = url
        self.comments = comments
        self.overall = overall
        self.shipping = shipping
        self.cost = cost
        self.honesty = honesty
        self.communication = communication
        self.name = name
        self.ipaddr = ipaddr
        self.date = date

Answer 1

Aquí está mi enfoque:

MODELO:

class AutoSerialize(object):
    ''Mixin for retrieving public fields of model in json-compatible format''
    __public__ = None

    def get_public(self, exclude=(), extra=()):
        "Returns model''s PUBLIC data for jsonify"
        data = {}
        keys = self._sa_instance_state.attrs.items()
        public = self.__public__ + extra if self.__public__ else extra
        for k, field in  keys:
            if public and k not in public: continue
            if k in exclude: continue
            value = self._serialize(field.value)
            if value:
                data[k] = value
        return data

    @classmethod
    def _serialize(cls, value, follow_fk=False):
        if type(value) in (datetime, date):
            ret = value.isoformat()
        elif hasattr(value, ''__iter__''):
            ret = []
            for v in value:
                ret.append(cls._serialize(v))
        elif AutoSerialize in value.__class__.__bases__:
            ret = value.get_public()
        else:
            ret = value

        return ret

class User(db.Model, AutoSerialize):
    __tablename__ = ''users''
    __public__ = (''id'', ''name'', ''email'')
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.Unicode(50))
    email = db.Column(db.String(120), unique=True)
    passhash = db.Column(db.String(100))
    ...

VER:

from flask import jsonfy

@mod.route(''/<int:id>/'', methods=[''GET''])
def get_user_by_id(id):
    u = User.query.get(id)
    return jsonify(u.get_public())

No estoy seguro de esto:

self._sa_instance_state.attrs.items()

pero funciona. No tuve tiempo suficiente para hacerlo más elegante, tal vez alguien sugiera una mejor manera de obtener campos de SA

Answer 2

Aquí está mi respuesta si estás usando la base declarativa (con la ayuda de algunas de las respuestas ya publicadas):

# in your models definition where you define and extend declarative_base()
from sqlalchemy.ext.declarative import declarative_base
...
Base = declarative_base()
Base.query = db_session.query_property()
...

# define a new class (call "Model" or whatever) with an as_dict() method defined
class Model():
    def as_dict(self):
        return { c.name: getattr(self, c.name) for c in self.__table__.columns }

# and extend both the Base and Model class in your model definition, e.g.
class Rating(Base, Model):
    ____tablename__ = ''rating''
    id = db.Column(db.Integer, primary_key=True)
    fullurl = db.Column(db.String())
    url = db.Column(db.String())
    comments = db.Column(db.Text)
    ...

# then after you query and have a resultset (rs) of ratings
rs = Rating.query.all()

# you can jsonify it with
s = json.dumps([r.as_dict() for r in rs], default=alchemyencoder)
print (s)

# or if you have a single row
r = Rating.query.first()

# you can jsonify it with
s = json.dumps(r.as_dict(), default=alchemyencoder)

# you will need this alchemyencoder where your are calling json.dumps to handle datetime and decimal format
# credit to Joonas @ http://codeandlife.com/2014/12/07/sqlalchemy-results-to-json-the-easy-way/
def alchemyencoder(obj):
    """JSON encoder function for SQLAlchemy special classes."""
    if isinstance(obj, datetime.date):
        return obj.isoformat()
    elif isinstance(obj, decimal.Decimal):
        return float(obj)

Answer 3

Aquí hay una manera de agregar un método as_dict () en cada clase, así como cualquier otro método que desee tener en cada clase. No estoy seguro de si esta es la manera deseada o no, pero funciona ...

class Base(object):
    def as_dict(self):
        return dict((c.name,
                     getattr(self, c.name))
                     for c in self.__table__.columns)


Base = declarative_base(cls=Base)

Answer 4

Estaba buscando algo así como el enfoque de rieles utilizado en ActiveRecord to_json e implementado algo similar con este Mixin después de estar insatisfecho con otras sugerencias. Maneja modelos anidados e incluye o excluye atributos del nivel superior o modelos anidados.

class Serializer(object):

    def serialize(self, include={}, exclude=[], only=[]):
        serialized = {}
        for key in inspect(self).attrs.keys():
            to_be_serialized = True
            value = getattr(self, key)
            if key in exclude or (only and key not in only):
                to_be_serialized = False
            elif isinstance(value, BaseQuery):
                to_be_serialized = False
                if key in include:
                    to_be_serialized = True
                    nested_params = include.get(key, {})
                    value = [i.serialize(**nested_params) for i in value]

            if to_be_serialized:
                serialized[key] = value

        return serialized

Luego, para obtener el BaseQuery serializable, extendí BaseQuery

class SerializableBaseQuery(BaseQuery):

    def serialize(self, include={}, exclude=[], only=[]):
        return [m.serialize(include, exclude, only) for m in self]

Para los siguientes modelos

class ContactInfo(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey(''user.id''))
    full_name = db.Column(db.String())
    source = db.Column(db.String())
    source_id = db.Column(db.String())

    email_addresses = db.relationship(''EmailAddress'', backref=''contact_info'', lazy=''dynamic'')
    phone_numbers = db.relationship(''PhoneNumber'', backref=''contact_info'', lazy=''dynamic'')


class EmailAddress(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    email_address = db.Column(db.String())
    type = db.Column(db.String())
    contact_info_id = db.Column(db.Integer, db.ForeignKey(''contact_info.id''))


class PhoneNumber(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    phone_number = db.Column(db.String())
    type = db.Column(db.String())
    contact_info_id = db.Column(db.Integer, db.ForeignKey(''contact_info.id''))

    phone_numbers = db.relationship(''Invite'', backref=''phone_number'', lazy=''dynamic'')

Podrías hacer algo como

@app.route("/contact/search", methods=[''GET''])
def contact_search():
    contact_name = request.args.get("name")
    matching_contacts = ContactInfo.query.filter(ContactInfo.full_name.like("%{}%".format(contact_name)))

    serialized_contact_info = matching_contacts.serialize(
        include={
            "phone_numbers" : {
                "exclude" : ["contact_info", "contact_info_id"]
            },
            "email_addresses" : {
                "exclude" : ["contact_info", "contact_info_id"]
            }
        }
    )

    return jsonify(serialized_contact_info)

Answer 5

Estaba trabajando con un querylic de SQL query default de listas de objetos RowProxy llamados jobDict. Me llevó un tiempo averiguar qué tipo de objetos eran.

Esta era una forma realmente simple y rápida de resolver algunos jsonEncoding limpios solo por el método de conversión de la fila a una lista y definiendo inicialmente el dict con un valor de lista.

    jobDict = defaultdict(list)
    def set_default(obj):
        # trickyness needed here via import to know type
        if isinstance(obj, RowProxy):
            return list(obj)
        raise TypeError


    jsonEncoded = json.dumps(jobDict, default=set_default)

Answer 6

Esto es lo que suele ser suficiente para mí:

Creo una mezcla de serialización que uso con mis modelos. La función de serialización básicamente recupera los atributos que expone el inspector SQLAlchemy y los coloca en un dict.

from sqlalchemy.inspection import inspect

class Serializer(object):

    def serialize(self):
        return {c: getattr(self, c) for c in inspect(self).attrs.keys()}

    @staticmethod
    def serialize_list(l):
        return [m.serialize() for m in l]

Todo lo que se necesita ahora es extender el modelo de SQLAlchemy con la clase mixin de Serializer .

Si hay campos que no desea exponer, o que necesitan un formato especial, simplemente anule la función serialize() en la subclase del modelo.

class User(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String)
    password = db.Column(db.String)

    # ...

    def serialize(self):
        d = Serializer.serialize(self)
        del d[''password'']
        return d

En sus controladores, todo lo que tiene que hacer es llamar a la función serialize() (o serialize_list(l) si la consulta da como resultado una lista) en los resultados:

def get_user(id):
    user = User.query.get(id)
    return json.dumps(user.serialize())

def get_users():
    users = User.query.all()
    return json.dumps(User.serialize_list(users))

Answer 7

Ha habido muchas veces y hay muchas respuestas válidas, pero el siguiente bloque de código parece funcionar:

my_object = SqlAlchemyModel()
my_serializable_obj = my_object.__dict__
del my_serializable_obj["_sa_instance_state"]
print(jsonify(my_serializable_object))

Soy consciente de que esta no es una solución perfecta, ni tan elegante como las demás, sin embargo, para aquellos que quieren una solución rápida, pueden intentar esto.

Answer 8

He estado observando este problema la mayor parte del día, y esto es lo que se me ocurrió (crédito a https://.com/a/5249214/196358 por señalarme en esta dirección).

(Nota: estoy usando el frasco-sqlalchemy, entonces el formato de declaración de mi modelo es un poco diferente del sqlalchemy directo).

En mi archivo models.py :

import json

class Serializer(object):
  __public__ = None
  "Must be implemented by implementors"

  def to_serializable_dict(self):
    dict = {}
    for public_key in self.__public__:
      value = getattr(self, public_key)
      if value:
        dict[public_key] = value
    return dict

class SWEncoder(json.JSONEncoder):
  def default(self, obj):
    if isinstance(obj, Serializer):
      return obj.to_serializable_dict()
    if isinstance(obj, (datetime)):
      return obj.isoformat()
    return json.JSONEncoder.default(self, obj)


def SWJsonify(*args, **kwargs):
  return current_app.response_class(json.dumps(dict(*args, **kwargs), cls=SWEncoder, indent=None if request.is_xhr else 2), mimetype=''application/json'')
  # stolen from https://github.com/mitsuhiko/flask/blob/master/flask/helpers.py

y todos mis objetos modelo se ven así:

class User(db.Model, Serializer):
  __public__ = [''id'',''username'']
  ... field definitions ...

En mi opinión, llamo a SWJsonify donde quiera que haya llamado Jsonify , así:

@app.route(''/posts'')
def posts():
  posts = Post.query.limit(PER_PAGE).all()
  return SWJsonify({''posts'':posts })

Parece funcionar bastante bien. Incluso en las relaciones. No he llegado muy lejos, así que YMMV, pero hasta ahora se siente bastante "correcto" para mí.

Sugerencias bienvenidas.

Answer 9

Ok, he estado trabajando en esto por unas horas, y he desarrollado lo que creo que es la solución más pitonica hasta el momento. Los siguientes fragmentos de código son python3 pero no deberían ser demasiado dolorosos para respaldar si lo necesita.

Lo primero que vamos a hacer es comenzar con un mixin que haga que tus modelos db actúen como dict s:

from sqlalchemy.inspection import inspect

class ModelMixin:
    """Provide dict-like interface to db.Model subclasses."""

    def __getitem__(self, key):
        """Expose object attributes like dict values."""
        return getattr(self, key)

    def keys(self):
        """Identify what db columns we have."""
        return inspect(self).attrs.keys()

Ahora vamos a definir nuestro modelo, heredando el mixin:

class MyModel(db.Model, ModelMixin):
    id = db.Column(db.Integer, primary_key=True)
    foo = db.Column(...)
    bar = db.Column(...)
    # etc ...

Eso es todo lo que se necesita para poder pasar una instancia de MyModel() a dict() y obtener una instancia real de dict real, lo que nos lleva bastante lejos para hacer que jsonify() entienda. A continuación, tenemos que extender JSONEncoder para obtener el resto del camino:

from flask.json import JSONEncoder
from contextlib import suppress

class MyJSONEncoder(JSONEncoder):
    def default(self, obj):
        # Optional: convert datetime objects to ISO format
        with suppress(AttributeError):
            return obj.isoformat()
        return dict(obj)

app.json_encoder = MyJSONEncoder

Puntos de bonificación: si su modelo contiene campos computados (es decir, desea que su salida JSON contenga campos que en realidad no están almacenados en la base de datos), eso también es fácil. Simplemente defina sus campos calculados como @property s, y amplíe el método keys() manera:

class MyModel(db.Model, ModelMixin):
    id = db.Column(db.Integer, primary_key=True)
    foo = db.Column(...)
    bar = db.Column(...)

    @property
    def computed_field(self):
        return ''this value did not come from the db''

    def keys(self):
        return super().keys() + [''computed_field'']

Ahora es trivial jsonificar:

@app.route(''/whatever'', methods=[''GET''])
def whatever():
    return jsonify(dict(results=MyModel.query.all()))

Answer 10

Para una consulta plana (sin uniones) puede hacer esto

@app.route(''/results/'')
def results():
    data = Table.query.all()
    result = [d.__dict__ for d in data]
    return jsonify(result=result)

y si solo quiere devolver ciertas columnas de la base de datos, puede hacer esto

@app.route(''/results/'')
def results():
    cols = [''id'', ''url'', ''shipping'']
    data = Table.query.all()
    result = [{col: getattr(d, col) for col in cols} for d in data]
    return jsonify(result=result)

Answer 11

Parece que en realidad no has ejecutado tu consulta. Pruebe lo siguiente:

return jsonify(json_list = qryresult.all())

[Editar] : El problema con jsonify es que, por lo general, los objetos no pueden ser jsonificados automáticamente. Incluso la fecha y hora de Python falla;)

Lo que suelo hacer es agregar una propiedad adicional (como serialize ) a las clases que necesitan ser serializadas:

def dump_datetime(value):
    """Deserialize datetime object into string form for JSON processing."""
    if value is None:
        return None
    return [value.strftime("%Y-%m-%d"), value.strftime("%H:%M:%S")]

class Foo(db.Model):
    # ... SQLAlchemy defs here..
    def __init__(self, ...):
       # self.foo = ...
       pass

    @property
    def serialize(self):
       """Return object data in easily serializeable format"""
       return {
           ''id''         : self.id,
           ''modified_at'': dump_datetime(self.modified_at),
           # This is an example how to deal with Many2Many relations
           ''many2many''  : self.serialize_many2many
       }
    @property
    def serialize_many2many(self):
       """
       Return object''s relations in easily serializeable format.
       NB! Calls many2many''s serialize property.
       """
       return [ item.serialize for item in self.many2many]

Y ahora para las vistas que puedo hacer:

return jsonify(json_list=[i.serialize for i in qryresult.all()])

Espero que esto ayude ;)

Answer 12

Si está utilizando un flask-restful puede usar marshal :

from flask.ext.restful import Resource, fields, marshal

topic_fields = {
    ''title'':   fields.String,
    ''content'': fields.String,
    ''uri'':     fields.Url(''topic''),
    ''creator'': fields.String,
    ''created'': fields.DateTime(dt_format=''rfc822'')
}

class TopicListApi(Resource):
    def get(self):
        return {''topics'': [marshal(topic, topic_fields) for topic in DbTopic.query.all()]}

Necesitas enumerar explícitamente lo que estás devolviendo y de qué tipo es, que de todos modos prefiero para una API. La serialización se soluciona fácilmente (sin necesidad de jsonify ), las fechas tampoco son un problema. Tenga en cuenta que el contenido para el campo uri se genera automáticamente en función del punto final del topic y la identificación.

Answer 13

Solo quiero agregar mi método para hacer esto.

simplemente defina un codificador de custome json para serilizar sus modelos db.

class ParentEncoder(json.JSONEncoder):
    def default(self, obj):
        # convert object to a dict
        d = {}
        if isinstance(obj, Parent):
            return {"id": obj.id, "name": obj.name, ''children'': list(obj.child)}
        if isinstance(obj, Child):
            return {"id": obj.id, "name": obj.name}

        d.update(obj.__dict__)
        return d

entonces en tu función de vista

parents = Parent.query.all()
dat = json.dumps({"data": parents}, cls=ParentEncoder)
resp = Response(response=dat, status=200, mimetype="application/json")
return (resp)

funciona bien aunque los padres tienen relaciones

Answer 14

Tenía la misma necesidad, serializarme en json. Eche un vistazo a esta pregunta . Muestra cómo descubrir columnas programáticamente. Entonces, de eso creé el código a continuación. Funciona para mí, y lo usaré en mi aplicación web. Feliz codificación!

def to_json(inst, cls):
    """
    Jsonify the sql alchemy query result.
    """
    convert = dict()
    # add your coversions for things like datetime''s 
    # and what-not that aren''t serializable.
    d = dict()
    for c in cls.__table__.columns:
        v = getattr(inst, c.name)
        if c.type in convert.keys() and v is not None:
            try:
                d[c.name] = convert[c.type](v)
            except:
                d[c.name] = "Error:  Failed to covert using ", str(convert[c.type])
        elif v is None:
            d[c.name] = str()
        else:
            d[c.name] = v
    return json.dumps(d)

class Person(base):
    __tablename__ = ''person''
    id = Column(Integer, Sequence(''person_id_seq''), primary_key=True)
    first_name = Column(Text)
    last_name = Column(Text)
    email = Column(Text)

    @property
    def json(self):
        return to_json(self, self.__class__)