I'm trying to turn this SQL expression into a subquery using SQL Alchemy
I'm trying to read the documentation for subqueries but I'm lost. Can anyone write this in SqlAlchemy so I can understand how to do it?
SELECT from_
FROM uploaded_emails1
WHERE from_ != '[email protected]'
AND from_ NOT IN (SELECT email FROM candidate_emails2)
How about using a left join instead of subquery :
SELECT from_
FROM uploaded_emails1 ue
LEFT JOIN candidate_emails2 ce ON ue.from_ = ce.email
WHERE ce.email IS NULL AND from_! = "[email protected]"
This could be expressed in SQLAlchemy as:
emails = dbsession.query(
UploadedEmail.from_
).outerjoin(
CandidateEmail, CandidateEmail.email == UploadedEmail.from_
).filter(
CandidateEmail.email == None,
UploadedEmail.from_ != '[email protected]',
)
Firstly, I think you might be trying to dive in a little bit at the deep end here (that SQL query is probably not really the right way to achieve what you want) so you might want to think about getting some SQL concepts a bit clearer in your head before diving into subqueries in sqlalchemy. Related to that, having tables named uploaded_emails1 and candidate_emails2 suggests that your underlying DB schema design might need a bit of tweaking.
That being said, here's a minimal example that should do what you want (get all email addresses that are in the uploaded emails that aren't also in the candidate emails table)
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
import sqlalchemy as sq
from sqlalchemy.orm import sessionmaker
engine = create_engine('sqlite://', echo=True)
Base = declarative_base()
class UploadedUser(Base):
__tablename__ = 'uploaded_user'
uploaded_user_id = sq.Column(sq.Integer, primary_key=True)
email_address = sq.Column(sq.Text(100))
class CandidateUser(Base):
__tablename__ = 'candidate_user'
candidate_user_id = sq.Column(sq.Integer, primary_key=True)
email_address = sq.Column(sq.Text(100))
def __repr__(self):
return '<Candidate User with email: {}>'.format(self.email_address)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
uploaded_user = UploadedUser(email_address='[email protected]')
session.add(uploaded_user)
for idx in range(1, 3):
candidate_user = CandidateUser(email_address='email{}@foo.com'.format(idx))
session.add(candidate_user)
session.commit()
query_result = session.query(CandidateUser.email_address,
UploadedUser.email_address).outerjoin(UploadedUser,
CandidateUser.email_address == UploadedUser.email_address).filter(
UploadedUser.email_address.isnot(None)).with_entities(UploadedUser.email_address).all()
print(query_result)
The echo=True tells sqlalchemy to print the actual queries it's executing so the query you're talking about as executed is:
SELECT uploaded_user.email_address AS uploaded_user_email_address
FROM candidate_user LEFT OUTER JOIN uploaded_user ON
candidate_user.email_address = uploaded_user.email_address
WHERE uploaded_user.email_address IS NOT NULL
which is quite close to the query you'd want to write if you were writing raw SQL.
for email, in db_session.query(Uploaded_Emails1.from_).filter(
Uploaded_Emails1.from_ != '[email protected]').filter(
~Uploaded_Emails1.from_.in_(db_session.query(Candidate_Emails2.email))):
print(email)
this worked