Как объединить 2 таблицы и получить последние записи из другой таблицы в соотношении один ко многим, используя SqlAlchemy ORM - PullRequest
Новая
       12

Как объединить 2 таблицы и получить последние записи из другой таблицы в соотношении один ко многим, используя SqlAlchemy ORM

1 голос
/ 15 марта 2019

допустим, у меня есть 2 таблицы пользователей и устройств. Они имеют отношение один ко многим. В Sql я могу решить упомянутую проблему следующим запросом. 

SELECT
    users.*, devices.*
FROM
    users

    LEFT JOIN ( SELECT d1.*
        FROM devices as d1
        LEFT JOIN devices AS d2
             ON d1.user_id = d2.user_id AND d1.date < d2.date
        WHERE d2.user_id IS NULL ) as device_temp

    ON (users.id = device_temp.user_id)

Вот мой код Python 

#user_model.py

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy import create_engine

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    first_name = Column(String(500), nullable=False)
    last_name = Column(String(250), nullable=False)

device_model.py 

#device_model.py

from sqlalchemy import Column, ForeignKey, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, relation
from sqlalchemy import create_engine


from user_model import User 
Base = declarative_base()

class DeviceModel(Base):
    __tablename__ = 'device'

    id = Column(Integer, primary_key=True)
    created_at = Column(DateTime(), nullable=False)
    device_id = Column(String(250), nullable=False)
    user_uid = Column(String, ForeignKey((User.id)))
    owner = relation(User, backref='user_device')

run.py 

#run.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


from user_model import User, Base
from sleep_session import SleepSession, Base
from device_model import DeviceModel, Base

engine = create_engine(connection_string)
Base.metadata.bind = engine

DBSession = sessionmaker(bind=engine)

session = DBSession()
query = session.query(User,DeviceModel).join(DeviceModel)
results = query.all()

for row in results:
        print(row.User.first_name +" "+ row.DeviceModel.device_id + " "+ str(row.DeviceModel.created_at))

Я знаю, что этот тип вопросов задают несколько раз, но я не смог найти один с SqlAlchemy ORM. Я хочу тот же результат , как описано здесь

Спасибо.

Ответы [ 2 ]

0 голосов
/ 15 марта 2019 
from db_config import connection_string 

from sqlalchemy import create_engine ,  and_ , inspect
from sqlalchemy.orm import sessionmaker, aliased

from user_model import User, Base
from device_model import DeviceModel, Base

engine = create_engine(connection_string)
Base.metadata.bind = engine
DBSession = sessionmaker(bind=engine)    

session = DBSession()
DeviceModel_aliased = aliased(DeviceModel)

#make sub-query
query_for_latest_device = session.query(DeviceModel).\
        outerjoin(DeviceModel_aliased,
                and_(DeviceModel_aliased.user_uid == DeviceModel.user_uid,
                        DeviceModel_aliased.created_at > DeviceModel.created_at)).\
        filter(DeviceModel_aliased.id == None).\
        subquery()

use_subquery_and_join = session.query(User.first_name,latest_device).\
        join(query_for_latest_device,
            query_for_latest_device.c.user_uid == User.user_id).\
        all()

for row in join_user_and_device:
    print(row._asdict())
0 голосов
/ 15 марта 2019

Я использовал этот вопрос для практической sqlalchemy, поскольку я новичок в этом. Ближайший ответ, который я могу получить, следующий:

Если вы хотите, чтобы 1 файл полностью работоспособного кода был внесен в правку - я удалю шаблонный код 

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy import Column, ForeignKey, Integer, String, DateTime
from sqlalchemy.orm import relationship, relation
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

connection_string = 'postgres://postgres:password@localhost/test'

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    first_name = Column(String(500), nullable=False)
    last_name = Column(String(250), nullable=False)

class DeviceModel(Base):
    __tablename__ = 'device'

    id = Column(Integer, primary_key=True)
    created_at = Column(DateTime(), nullable=False)
    device_id = Column(String(250), nullable=False)
    user_uid = Column(Integer, ForeignKey((User.id))) # error Key columns "user_uid" and "id" are of incompatible types: character varying and integer.
    owner = relation(User, backref='user_device')


engine = create_engine(connection_string)
Base.metadata.bind = engine

#User.__table__.create(engine)
#DeviceModel.__table__.create(engine)

DBSession = sessionmaker(bind=engine)
session = DBSession()

Мой ответ: 

from sqlalchemy import and_, or_
from sqlalchemy.orm import aliased

DeviceModel2 = aliased(DeviceModel)

subquery = (session
    .query(DeviceModel.created_at)
    .outerjoin(DeviceModel2, 
      and_(DeviceModel.user_uid == DeviceModel2.user_uid,
        DeviceModel.created_at < DeviceModel2.created_at))
    .filter(DeviceModel2.user_uid == None)
    .subquery('subq'))

query = (session
    .query(User, DeviceModel)
    .outerjoin(DeviceModel)
    .filter(or_(
        DeviceModel.created_at == subquery.c.created_at,
        DeviceModel.id == None)))

print(query)
results = query.all()


for row in results:
    if row[1]:
        print({**row.User.__dict__, **row.DeviceModel.__dict__})
    else:
        print(row.User.__dict__)
...