On Tue, Dec 25, 2018 at 10:18 PM Pavel Pristupa <[email protected]> wrote:
>
> Hello everybody!
>
> What I want sometimes is to query some columns but to map the resulting rows
> into custom data classes rather than tuples, like with values(*columns), or
> SQLA entities.
> I found that Bundles
> (https://docs.sqlalchemy.org/en/latest/orm/loading_columns.html#bundles)
> could potentially help, but what I see there is that I can process rows with
> create_row_processor and give the immediate result of one-to-one converting
> every single row.
> Let's say I have User and Address with 1:N relation:
>
>
> class User(Base):
> __tablename__ = 'users'
>
> id = Column(Integer, primary_key=True)
> name = Column(String)
>
> addresses = relationship('Address')
>
>
> class Address(Base):
> __tablename__ = 'addresses'
>
> id = Column(Integer, primary_key=True)
> user_id = Column(ForeignKey(User.id))
> city = Column(String)
>
>
>
> I'd like to query all users into a custom structure like this:
>
>
> [{
> 'id': 1,
> 'name': 'User 1',
> 'addresses': [{
> 'id': 1,
> 'city': 'City 1',
> }, {
> 'id': 2,
> 'city': 'City 2',
> }],
> }, {
> 'id': 2,
> 'name': 'User 2',
> 'addresses': [{
> 'id': 3,
> 'city': 'City 3',
> }],
> }]
>
>
>
> It's similar to what ORM does for me when using joinedload for relationships,
> but how to nest related items without ORM identity-mapped classes?
> Your help would be much appreciated!
by far the simplest way would be to just query User/Address normally
then have them serialize into that structure with some method like
User.to_json(). Is the reason you don't want to use ORM entities due
to performance?
If you'd like the Bundle.create_row_processor to do it, that's
possible also but you need to re-implement a miniature identity map as
well as receive the nested rows and also deduplicate on the result
side. Basically re-implementing what the ORM already does for you:
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import declared_attr
Base = declarative_base()
class A(Base):
__tablename__ = 'a'
id = Column(Integer, primary_key=True)
data = Column(String)
bs = relationship("B")
class B(Base):
__tablename__ = 'b'
id = Column(Integer, primary_key=True)
a_id = Column(ForeignKey("a.id"))
data = Column(String)
e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
s = Session(e)
s.add_all([
A(data='a1', bs=[B(data='b1'), B(data='b2')]),
A(data='a2', bs=[B(data='b3'), B(data='b4')])
])
s.commit()
class ABundle(Bundle):
def create_row_processor(self, query, procs, labels):
"""Override create_row_processor to return values as dictionaries"""
a_map = {}
def proc(row):
pk = row['a_id']
if pk in a_map:
rec = a_map[pk]
else:
rec = a_map[pk] = {
"pk": row['a_id'], "data": row['a_data'], "bs": []}
if row['b_id']:
rec['bs'].append({'pk': row['b_id'], 'data': row['b_data']})
return rec
return proc
def _unique_dict_rows(iterator):
# uniquify dupe rows
map_by_pk = {}
for rec, in iterator:
pk = rec['pk']
if pk not in map_by_pk:
yield rec
map_by_pk[rec['pk']] = rec
q = s.query(
ABundle('mybundle', A.id, A.data, B.id, B.data)).select_from(A).join(B)
print(list(_unique_dict_rows(q)))
>
> --
> SQLAlchemy -
> The Python SQL Toolkit and Object Relational Mapper
>
> http://www.sqlalchemy.org/
>
> To post example code, please provide an MCVE: Minimal, Complete, and
> Verifiable Example. See http://stackoverflow.com/help/mcve for a full
> description.
> ---
> You received this message because you are subscribed to the Google Groups
> "sqlalchemy" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to [email protected].
> To post to this group, send email to [email protected].
> Visit this group at https://groups.google.com/group/sqlalchemy.
> For more options, visit https://groups.google.com/d/optout.
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
