Hi All,
using 7.10 but falls over also in 8.0.
User has a One2One UserPerson.
UserPerson inherits from Person.
Person has a One2Many PersonAddress.
PersonAddress inherits from Address.
Address has a One2Many Phone and One2Many Email.
The following query falls over with an (OperationalError) no such column:
Address.Id.
session.query(User, UserPerson, PersonAddress).filter(User.Id == 1).\
outerjoin(User.Personal,UserPerson.Addresses,PersonAddress.Phones,PersonAddress.Emails).first()
What am I doing wrong ?
Thanks in advance for your help.
sqlalchemy.exc.OperationalError: (OperationalError) no such column:
Address.Id u'SELECT "User"."Id" AS "User_Id", anon_1."UserPerson_Id" AS
"anon_1_UserPerson_Id", anon_1."Person_Id" AS "anon_1_Person_Id",
anon_1."Person_PersonType" AS "anon_1_Person_PersonType",
anon_1."UserPerson_ItemUserPerson_Id" AS
"anon_1_UserPerson_ItemUserPerson_Id", anon_2."PersonAddress_Id" AS
"anon_2_PersonAddress_Id", anon_2."Address_Id" AS "anon_2_Address_Id",
anon_2."Address_AddressType" AS "anon_2_Address_AddressType",
anon_2."PersonAddress_ItemPerson_Id" AS
"anon_2_PersonAddress_ItemPerson_Id" \nFROM "User" LEFT OUTER JOIN (SELECT
"Person"."Id" AS "Person_Id", "Person"."PersonType" AS "Person_PersonType",
"UserPerson"."Id" AS "UserPerson_Id", "UserPerson"."ItemUserPerson_Id" AS
"UserPerson_ItemUserPerson_Id" \nFROM "Person" JOIN "UserPerson" ON
"UserPerson"."Id" = "Person"."Id") AS anon_1 ON
anon_1."UserPerson_ItemUserPerson_Id" = "User"."Id" LEFT OUTER JOIN (SELECT
"Address"."Id" AS "Address_Id", "Address"."AddressType" AS
"Address_AddressType", "PersonAddress"."Id" AS "PersonAddress_Id",
"PersonAddress"."ItemPerson_Id" AS "PersonAddress_ItemPerson_Id" \nFROM
"Address" JOIN "PersonAddress" ON "PersonAddress"."Id" = "Address"."Id") AS
anon_2 ON anon_1."Person_Id" = anon_2."PersonAddress_ItemPerson_Id" LEFT
OUTER JOIN "Phone" ON anon_2."Address_Id" = "Phone"."ItemPhone_Id" LEFT
OUTER JOIN "Email" ON "Address"."Id" = "Email"."ItemEmail_Id" \nWHERE
"User"."Id" = ?\n LIMIT ? OFFSET ?' (1, 1, 0)
--
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 http://groups.google.com/group/sqlalchemy?hl=en.
For more options, visit https://groups.google.com/groups/opt_out.
from sqlalchemy import __version__
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, ForeignKey, Integer, create_engine
from sqlalchemy.orm import relationship, Session
from decl_enum import DeclEnum
Base = declarative_base()
class Email(Base):
__tablename__ = 'Email'
Id = Column(Integer, primary_key=True)
# Many2One side of Address One2Many - backref Email.ItemEmail
ItemEmail_Id = Column(Integer,
ForeignKey('Address.Id', use_alter=True, name="FK_Email_ItemEmail_Id"))
class Phone(Base):
__tablename__ = 'Phone'
Id = Column(Integer, primary_key=True)
# Many2One side of Address One2Many - backref Phone.ItemPhone
ItemPhone_Id = Column(Integer,
ForeignKey('Address.Id', use_alter=True, name="FK_Phone_ItemPhone_Id"))
class AddressItemType(DeclEnum):
person = "PersonAddress", "Personal Address"
class Address(Base):
__tablename__ = 'Address'
Id = Column(Integer, primary_key=True)
AddressType = Column(AddressItemType.db_type(), nullable=False)
# One2Many side of Phone Many2One - backref Phone.ItemPhone
Phones = relationship("Phone", uselist=True, cascade="delete", backref="ItemPhone")
# One2Many side of Email Many2One - backref Email.ItemEmail
Emails = relationship("Email", uselist=True, cascade="delete", backref="ItemEmail")
__mapper_args__ = {'polymorphic_on': AddressType,
'with_polymorphic': '*'
}
class PersonAddress(Address):
__tablename__ = 'PersonAddress'
Id = Column(Integer, ForeignKey('Address.Id'), primary_key=True)
__mapper_args__ = {'polymorphic_identity': AddressItemType.person,
'inherit_condition': (Id == Address.Id)
}
# ManyToOne side of Person OneTomany - backref PersonAddress.ItemPerson
ItemPerson_Id = Column(Integer,
ForeignKey('Person.Id', use_alter=True, name="FK_PersonAddress_ItemPerson_Id"))
class PersonItemType(DeclEnum):
user = "UserPerson", "User Person"
class Person(Base):
__tablename__ = 'Person'
Id = Column(Integer, primary_key=True)
PersonType = Column(PersonItemType.db_type(), nullable=False)
# One2Many side of PersonAddress Many2One - backref PersonAddress.ItemPerson
Addresses = relationship("PersonAddress",
uselist=True, cascade="delete", backref="ItemPerson")
__mapper_args__ = {'polymorphic_on': PersonType,
'with_polymorphic': '*'
}
class UserPerson(Person):
__tablename__ = 'UserPerson'
Id = Column(Integer, ForeignKey('Person.Id'), primary_key=True)
__mapper_args__ = {'polymorphic_identity': PersonItemType.user,
'inherit_condition': (Id == Person.Id)
}
# One2One side of User One2One
ItemUserPerson_Id = Column(Integer,
ForeignKey("User.Id", use_alter=True, name="FK_UserPerson_ItemUserPerson_Id"))
ItemUserPerson = relationship("User",
primaryjoin="User.Id==UserPerson.ItemUserPerson_Id",
back_populates="Personal")
class User(Base):
__tablename__ = 'User'
Id = Column(Integer, primary_key=True)
# One2One side of UserPerson One2One
Personal = relationship("UserPerson",
primaryjoin="UserPerson.ItemUserPerson_Id==User.Id",
uselist=False, cascade="delete", back_populates="ItemUserPerson")
engine = create_engine('sqlite://', echo=False)
Base.metadata.create_all(engine)
print __version__
if __name__ == '__main__':
session = Session(engine)
# 1 - error - my preferred
session.query(User, UserPerson, PersonAddress).filter(User.Id == 1).\
outerjoin(User.Personal,UserPerson.Addresses,PersonAddress.Phones,PersonAddress.Emails).first()
# 2 - OK - but I need PersonAddress.Emails
#session.query(User, UserPerson, PersonAddress).filter(User.Id == 1).\
# outerjoin(User.Personal,UserPerson.Addresses,PersonAddress.Phones).first()
# 3 - error - just clutching at straws
#session.query(User, UserPerson, PersonAddress, Phone, Email).filter(User.Id == 1).\
# outerjoin(User.Personal,UserPerson.Addresses,PersonAddress.Phones,PersonAddress.Emails).first()
# 4 - OK - but why define Address
#session.query(User, UserPerson, PersonAddress, Address).filter(User.Id == 1).\
# outerjoin(User.Personal,UserPerson.Addresses,PersonAddress.Phones,PersonAddress.Emails).first()
# 5 - OK - but PersonAddress might have some fields not in Adddress
#session.query(User, UserPerson, Address).filter(User.Id == 1).\
# outerjoin(User.Personal,UserPerson.Addresses,PersonAddress.Phones,PersonAddress.Emails).first()
from sqlalchemy.types import SchemaType, TypeDecorator, Enum
from sqlalchemy import __version__
import re
if __version__ < '0.6.5':
raise NotImplementedError("Version 0.6.5 or higher of SQLAlchemy is required.")
class EnumSymbol(object):
"""Define a fixed symbol tied to a parent class."""
def __init__(self, cls_, name, value, description):
self.cls_ = cls_
self.name = name
self.value = value
self.description = description
def __reduce__(self):
"""Allow unpickling to return the symbol
linked to the DeclEnum class."""
return getattr, (self.cls_, self.name)
def __iter__(self):
return iter([self.value, self.description])
def __repr__(self):
return "<%s>" % self.name
class EnumMeta(type):
"""Generate new DeclEnum classes."""
def __init__(cls, classname, bases, dict_):
cls._reg = reg = cls._reg.copy()
for k, v in dict_.items():
if isinstance(v, tuple):
sym = reg[v[0]] = EnumSymbol(cls, k, *v)
setattr(cls, k, sym)
return type.__init__(cls, classname, bases, dict_)
def __iter__(cls):
return iter(cls._reg.values())
class DeclEnum(object):
"""Declarative enumeration."""
__metaclass__ = EnumMeta
_reg = {}
@classmethod
def from_string(cls, value):
try:
return cls._reg[value]
except KeyError:
raise ValueError(
"Invalid value for %r: %r" %
(cls.__name__, value)
)
@classmethod
def values(cls):
return cls._reg.keys()
@classmethod
def db_type(cls):
return DeclEnumType(cls)
class DeclEnumType(SchemaType, TypeDecorator):
def __init__(self, enum):
self.enum = enum
self.impl = Enum(
*enum.values(),
name="ck%s" % re.sub(
'([A-Z])',
lambda m:"_" + m.group(1).lower(),
enum.__name__)
)
def _set_table(self, table, column):
self.impl._set_table(table, column)
def copy(self):
return DeclEnumType(self.enum)
def process_bind_param(self, value, dialect):
if value is None:
return None
return value.value
def process_result_value(self, value, dialect):
if value is None:
return None
return self.enum.from_string(value.strip())
if __name__ == '__main__':
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import Session
Base = declarative_base()
class EmployeeType(DeclEnum):
part_time = "P", "Part Time"
full_time = "F", "Full Time"
contractor = "C", "Contractor"
class Employee(Base):
__tablename__ = 'employee'
id = Column(Integer, primary_key=True)
name = Column(String(60), nullable=False)
type = Column(EmployeeType.db_type())
def __repr__(self):
return "Employee(%r, %r)" % (self.name, self.type)
e = create_engine('sqlite://', echo=True)
Base.metadata.create_all(e)
sess = Session(e)
sess.add_all([
Employee(name='e1', type=EmployeeType.full_time),
Employee(name='e2', type=EmployeeType.full_time),
Employee(name='e3', type=EmployeeType.part_time),
Employee(name='e4', type=EmployeeType.contractor),
Employee(name='e5', type=EmployeeType.contractor),
])
sess.commit()
print sess.query(Employee).filter_by(type=EmployeeType.contractor).all()
