little bit of a bug regarding the bound parameter being created in the primary
join condition, here's a workaround:
class Line(Base):
__tablename__ = 'line'
id = Column(Integer, primary_key=True)
start_point_id = Column(Integer, nullable=True)
start_point = relationship(
'Point',
primaryjoin='and_('
'(Line.start_point_id == Point.id),'
'(Point.flag == literal_column("1")))',
foreign_keys=[start_point_id])
end_point_id = Column(Integer, nullable=True)
end_point = relationship(
'Point',
primaryjoin='and_('
'(Line.end_point_id == Point.id),'
'(Point.flag == literal_column("1")))',
foreign_keys=[end_point_id])
patch which fixes is at http://www.sqlalchemy.org/trac/ticket/2425, we'll have
it in 0.7.6.
On Mar 8, 2012, at 12:25 PM, Christopher Abiad wrote:
> from sqlalchemy.engine import engine_from_config
> from sqlalchemy.exc import CompileError
> from sqlalchemy.ext.declarative import declarative_base
> from sqlalchemy.orm import relationship
> from sqlalchemy.orm.session import sessionmaker
> from sqlalchemy.schema import Column
> from sqlalchemy.sql.expression import and_, or_
> from sqlalchemy.types import Boolean, Integer
>
> settings = {'sqlalchemy.url': 'sqlite://'}
> engine = engine_from_config(settings, 'sqlalchemy.')
> DBSession = sessionmaker()
> Base = declarative_base()
>
> Base.metadata.bind = engine
> DBSession = sessionmaker(bind=engine)
>
>
> class Coordinate(Base):
> __tablename__ = 'coordinate'
> id = Column(Integer, primary_key=True)
> x = Column(Integer)
>
>
> class Point(Base):
> __tablename__ = 'point'
> id = Column(Integer, primary_key=True)
> flag = Column(Boolean, default=True)
>
> coordinate_id = Column(Integer, nullable=True)
> coordinate = relationship(
> 'Coordinate',
> primaryjoin='Point.coordinate_id == Coordinate.id',
> foreign_keys=[coordinate_id])
>
>
> class Line(Base):
> __tablename__ = 'line'
> id = Column(Integer, primary_key=True)
>
> start_point_id = Column(Integer, nullable=True)
> start_point = relationship(
> 'Point',
> primaryjoin='and_('
> '(Line.start_point_id == Point.id),'
> '(Point.flag == True))',
> foreign_keys=[start_point_id])
>
> end_point_id = Column(Integer, nullable=True)
> end_point = relationship(
> 'Point',
> primaryjoin='and_('
> '(Line.end_point_id == Point.id),'
> '(Point.flag == True))',
> foreign_keys=[end_point_id])
>
>
> Base.metadata.create_all(engine)
>
> session = DBSession()
>
> starts_one = Line.start_point.has(Point.coordinate.has(Coordinate.x ==
> 1))
> ends_one = Line.end_point.has(Point.coordinate.has(Coordinate.x == 1))
> starts_other = Line.start_point.has(Point.coordinate.has(Coordinate.x !
> = 1))
> ends_other = Line.end_point.has(Point.coordinate.has(Coordinate.x !=
> 1))
>
> one_to_other = and_(starts_one, ends_other)
> other_to_one = and_(starts_other, ends_one)
>
> try:
> or_results = session.query(Line).filter(or_(
> one_to_other, other_to_one
> )).all()
> except CompileError as e:
> print "query with or_() causes CompileError. Error was:"
> print str(e)
>
> q1 = session.query(Line).filter(one_to_other)
> q2 = session.query(Line).filter(other_to_one)
> q3 = q1.union(q2)
>
> try:
> union_results = q3.all()
> except CompileError as e:
> print "query with union() causes CompileError. Error was:"
> print str(e)
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalchemy@googlegroups.com.
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com.
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
