test_two_levels.py

# -*- coding: utf-8 -*-
import logging

from sqlalchemy import Column
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, relationship
from sqlalchemy.orm import sessionmaker
from zope.sqlalchemy import ZopeTransactionExtension

DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))
Model = declarative_base()

log = logging.getLogger(__name__)


class Department(Model):
    __tablename__ = 'department'
    id = Column(Integer, primary_key=True)
    employees = relationship('Employee', backref='department')


class Employee(Model):
    __tablename__ = 'employee'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    department_id = Column(Integer, ForeignKey('department.id'))
    email_addresses = relationship('EmailAddress', backref='employee')


class EmailAddress(Model):
    __tablename__ = 'email'
    id = Column(Integer, primary_key=True)
    value = Column(String)
    employee_id = Column(Integer, ForeignKey('employee.id'))

    def __init__(self, employee_id, email):
        self.employee_id = employee_id
        self.value = email

engine = create_engine('sqlite://')
DBSession.configure(bind=engine)
Model.metadata.create_all(engine)

department1 = Department()
department2 = Department()
DBSession.add(department1)
DBSession.add(department2)
employee = Employee()
employee.email_addresses.append(EmailAddress(employee.id, '[email protected]'))
DBSession.add(employee)
department1.employees.append(employee)
DBSession.flush()

print(DBSession.query(Department).filter(Department.employees.any(Employee.email_addresses.any(EmailAddress.value == '[email protected]'))).all())