Hello everyone,
I am modelling a relationship between two groups which is a many-to-many
relationship, with an additional constraint.
I have an Employer and a Worker class. Employers can have a relationship to
a worker in a number of ways, which
I'm representing by an enum: Contractor, Employee, Part-Time etc.
The join table has three columns, employer_id, worker_id and worker_type,
which is the enum value.
The below code works for creating the relationship, but I don't quite
understand how to query it.
I create an employer, and then a worker.
Then I use
employer.contractor.append(worker)
The join table contains the correct ids and the 'CONTRACTOR' enum value.
But when I iterate through employer.part_time, I get back the same worker
who is a 'CONTRACTOR'. I'm expecting that only
employer.contractor should contain values, and that employer.employee and
employer.part_time should be empty.
Is there something corresponding to *creator* that can be used when
retrieving the values?
Or, what is the best way for me to get back the correct values? Do I need
to issue a query, and with what parameters?
from sqlalchemy import Column, ForeignKey, Enum, Text, Integer,
create_engine
from sqlalchemy.orm import relationship, backref, sessionmaker,
scoped_session
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.declarative import declarative_base
from enum import Enum as eEnum
Base = declarative_base()
class WorkerType(eEnum):
CONTRACTOR = 0
EMPLOYEE = 1
PART_TIME = 2
class Employer(Base):
__tablename__ = "employer"
id = Column(Integer, primary_key=True)
name = Column(Text)
contractor = association_proxy(
"employer_workers", "worker",
creator=lambda el:EmployerWorkerAssociation(
worker=el,
worker_type=WorkerType.CONTRACTOR
)
)
employee = association_proxy(
"employer_workers", "worker",
creator=lambda el:EmployerWorkerAssociation(
worker=el,
worker_type=WorkerType.EMPLOYEE
)
)
part_time = association_proxy(
"employer_workers", "worker",
creator=lambda el:EmployerWorkerAssociation(
worker=el,
worker_type=WorkerType.PART_TIME
)
)
class EmployerWorkerAssociation(Base):
__tablename__ = "employer_to_worker"
employer_id = Column(ForeignKey("employer.id"), primary_key=True)
worker_id = Column(ForeignKey("worker.id"), primary_key=True)
worker_type = Column(Enum(WorkerType), primary_key=True)
employer = relationship(
"Employer",
backref=backref("employer_workers", cascade="all, delete-orphan")
)
worker = relationship(
"Worker",
backref=backref("worker_employers")
)
class Worker(Base):
__tablename__ = "worker"
id = Column(Integer, primary_key=True)
name = Column(Text)
engine = create_engine("sqlite://")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
e1 = Employer(name="The Company")
session.add(e1)
session.commit()
w1 = Worker(name="The Programmer")
session.add(w1)
session.commit()
e1.contractor.append(w1)
session.add(e1)
44,0-1
81%
session.commit()
# I'm expecting contractor=1 employee=0 part_time=0
print(f"Contractors: {len(e1.contractor)}")
print(f"Employees : {len(e1.employee)}")
print(f"Part Timers: {len(e1.part_time)}")
# There shouldn't be anyone 'PART_TIME'
for worker in e1.part_time:
print(f"{worker.id} {worker.name}")
Thank you for your time.
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
