python - 如何使用 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')

运行.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

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