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)
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())
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))
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'
}]
}