如何使用 SqlAlchemy ORM 以一对多关系连接 2 个表并从其他表中获取最新记录

Question

假设我有 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 的问题。 我想要同样的结果as described here

谢谢。

Answer 1

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())

Answer 2

我使用这个问题来练习 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__)