Disons que j'ai 2 tables d'utilisateurs et d'appareils. Ils ont une relation un à plusieurs. En SQL, je peux résoudre le problème mentionné en suivant la requête.

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)

Voici mon code 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)) 

Je sais que ce type de question est posé plusieurs fois, mais je n'en ai pas trouvé avec SqlAlchemy ORM. Je veux le même résultat comme décrit ici

Merci.

1
danish 15 mars 2019 à 11:00

2 réponses

Meilleure réponse
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
danish 25 mars 2019 à 08:24

J'ai utilisé cette question pour pratiquer le sqlalchemy car je suis nouveau dans ce domaine. La réponse la plus proche que je peux obtenir est la suivante:

Si vous voulez voir 1 fichier complet de code exploitable aller dans les éditions - je supprimerai le code passe-partout

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

Ma réponse:

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__)
0
Ilja Everilä 15 mars 2019 à 11:39