how to get dict of lists from relationship in sqlalchemy?

  • Last Update :
  • Techknowledgy :

So, it turns out you can simply inherit MappedCollection and do whatever you like in setitem there.

from sqlalchemy.orm.collections import(MappedCollection,
   _SerializableAttrGetter,
   collection,
   _instrument_class)

#This will ensure that the MappedCollection has been properly
#initialized with custom __setitem__() and __delitem__() methods
#before used in a custom subclass
_instrument_class(MappedCollection)

class DictOfListsCollection(MappedCollection):

   @collection.internally_instrumented
def __setitem__(self, key, value, _sa_initiator = None):
   if not super(DictOfListsCollection, self).get(key):
   super(DictOfListsCollection, self).__setitem__(key, [], _sa_initiator)
super(DictOfListsCollection, self).__getitem__(key).append(value)

Suggestion : 2

sqlalchemy.orm.relationship() , .comparable_property() , sqlalchemy.ext.declarative , sqlalchemy.create_engine()

def object_as_dict(obj, relationships = False):
   ""
"
Converts an SQLAlchemy instance to a dictionary.

   : param relationships: If true, also include relationships in the output dict ""
"
properties = inspect(obj).mapper.all_orm_descriptors

if not relationships:
   properties = {
      key: value
      for key,
      value in properties.items()
      if not hasattr(value, "prop")
      or not isinstance(value.prop, RelationshipProperty)
   }

return {
   key: getattr(obj, key) for key,
   value in properties.items()
}
def contribute_to_class(self, mcs_args: McsArgs, relationships):
   if mcs_args.Meta.abstract:
   return

discovered_relationships = {}

def discover_relationships(d):
   for k, v in d.items():
   if isinstance(v, RelationshipProperty):
   discovered_relationships[v.argument] = k
if v.backref and mcs_args.Meta.lazy_mapped:
   raise NotImplementedError(
      f 'Discovered a lazy-mapped backref `{k}` on '
      f '`{mcs_args.qualname}`. Currently this '
      'is unsupported; please use `db.relationship` with '
      'the `back_populates` kwarg on both sides instead.')

for base in mcs_args.bases:
   discover_relationships(vars(base))
discover_relationships(mcs_args.clsdict)

relationships.update(discovered_relationships)
def get_field_type(model, fieldname):
   ""
"Helper which returns the SQLAlchemy type of the field.

""
"
field = getattr(model, fieldname)
if isinstance(field, ColumnElement):
   fieldtype = field.type
else:
   if isinstance(field, AssociationProxy):
   field = field.remote_attr
if hasattr(field, 'property'):
   prop = field.property
if isinstance(prop, RelProperty):
   return None
fieldtype = prop.columns[0].type
else:
   return None
return fieldtype
def is_like_list(instance, relation):
   ""
"Returns ``True`` if and only if the relation of `instance` whose name is
`relation`
is list - like.

A relation may be like a list
if,
for example, it is a non - lazy one - to - many
relation, or it is a dynamically loaded one - to - many.

""
"
if relation in instance._sa_class_manager:
   return instance._sa_class_manager[relation].property.uselist
elif hasattr(instance, relation):
   attr = getattr(instance._sa_instance_state.class_, relation)
if hasattr(attr, 'property'):
   return attr.property.uselist
related_value = getattr(type(instance), relation, None)
if isinstance(related_value, AssociationProxy):
   local_prop = related_value.local_attr.prop
if isinstance(local_prop, RelProperty):
   return local_prop.uselist
return False
def __init__(self, relationships: Mapping[str, RelationshipProperty]):
   ""
" Init relationships: param relationships: Model relationships ""
"
super(RelationshipsBag, self).__init__()
self._relations = relationships
self._rel_names = frozenset(self._relations.keys())
self._array_rel_names = frozenset(name
   for name, rel in self._relations.items() if _is_relationship_array(rel))
def __iter__(self) - > Iterable[Tuple[str, RelationshipProperty]]:
   ""
" Get relationships "
""
return iter(self._relations.items())

Suggestion : 3

If you're querying columns individually, the row is a KeyedTuple which has an _asdict method. The method name starts with a single underscore, to match the namedtuple API (it's not private!).,When using the ORM to retrieve objects, this is not available by default. The SQLAlchemy inspection system should be used.,Here, we created a function to do the conversion, but one option would be to add a method to the base class.,Get monthly updates about new articles, cheatsheets, and tricks.

First the setup for the example:

import datetime as dt
from sqlalchemy
import Column, Date, Integer, Text, create_engine, inspect
from sqlalchemy.orm
import sessionmaker
from sqlalchemy.ext.declarative
import declarative_base

Base = declarative_base()
Session = sessionmaker()

class User(Base):
   __tablename__ = 'users'

id = Column(Integer, primary_key = True)
name = Column(Text, nullable = False)
birthday = Column(Date)

engine = create_engine('sqlite://')
Base.metadata.create_all(bind = engine)
Session.configure(bind = engine)

session = Session()
session.add(User(name = 'Alice', birthday = dt.date(1990, 1, 1)))
session.commit()

If you're querying columns individually, the row is a KeyedTuple which has an _asdict method. The method name starts with a single underscore, to match the namedtuple API (it's not private!).

query = session.query(User.name, User.birthday)
for row in query:
   print(row._asdict())

When using the ORM to retrieve objects, this is not available by default. The SQLAlchemy inspection system should be used.

def object_as_dict(obj):
   return {
      c.key: getattr(obj, c.key)
      for c in inspect(obj).mapper.column_attrs
   }

query = session.query(User)
for user in query:
   print(object_as_dict(user))

Suggestion : 4

Dictalchemy adds utils.asdict() and utils.fromdict() methods to SQLAlchemy declarative models.,Currently this works with synonyms and simple relationships as one-to-many and many-to-many. Relationships can be followed in many levels.,Works almost identically as dictalchemy.utils.asdict(). However, it will not create missing instances or update collections.,Contains the methods DictableModel.__iter__(), DictableModel.asdict() and DictableModel.fromdict().

from dictalchemy
import DictableModel
from slqlachemy.ext.declarative
import declarative_base
Base = declarative_base(cls = DictableModel)
Base = declarative_base()
make_class_dictable(Base)
>>> session.query(User).asdict() {
   'id': 1,
   'username': 'Gerald'
}
>>> session.query(User).asdict(exclude_pk = True) {
   'username': 'Gerald'
}
>>> session.query(User).asdict(exclude = ['id']) {
   'username': 'Gerald'
}
>>> session.query(User).asdict(follow = {
   'groups': {}
}) {
   'username': 'Gerald',
   groups = [{
      'id': 1,
      'name': 'User'
   }]
}