sqlalchemy desc error

test.py

# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
#
# Use this file to build your own SSCCE
# SSCCE = Short, Self Contained, Correct (Compatible) Example
# see http://sscce.org/
#
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -


# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Standard imports

import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import deferred, class_mapper
from sqlalchemy import Integer, String, Text, Binary, Column, ForeignKey, DateTime
from sqlalchemy import inspect
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# You probably don't need to overwrite this
Base = declarative_base()


# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Define some models that inherit from Base

class Foo(Base):
    __tablename__ = 'foo'
    id = Column(Integer, primary_key=True)
    timestamp_created = Column(DateTime, nullable=True)

class Bar(Base):
    __tablename__ = 'bar'
    id_field = Column(Integer, primary_key=True)
    timestamp_created = Column(DateTime, nullable=True)



# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# we'll use sqlite+memory to handle this
# if your issue is backend specific, then the engine will need to be different


engine = create_engine('postgresql://sqlalchemy_test:sqla@localhost/sqlalchemy_test')
Base.metadata.create_all(engine)


# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# you shouldn't change these 2 line

sessionFactory = sessionmaker(bind=engine)
s = sessionFactory()

# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Start your example here:

# fun utility
import sqlparse
from sqlalchemy.dialects import postgresql as dialect_postgresql
def print_query(q):
    """prints a sqlalchemy query"""
    print "-" * 30
    if hasattr(q, 'statement'):
        print "[q.statement.compile %s]" % type(q)
        statement = str(q.statement.compile(dialect=dialect_postgresql.dialect(), compile_kwargs={"literal_binds": True}))
    elif hasattr(q, 'compile'):
        print "[q.compile %s]" % type(q)
        statement = str(q.compile(dialect=dialect_postgresql.dialect(), compile_kwargs={"literal_binds": True}))
    else:
        print "[q %s]" % type(q)
        statement = str(q)
    print sqlparse.format(statement, reindent=True, keyword_case='upper')
    print "-" * 30


def query_form_a():
    """
        label the stream alias column to sort
        1.0.0b2 - fail
        0.9.9 - pass
    """
    q1 = s.query(
            Foo.id,
            Foo.timestamp_created.label('event_timestamp')
        )\
        .subquery('qFoo')
    q2 = s.query(
            Bar.id_field.label('id'),
            Bar.timestamp_created.label('event_timestamp')
        )\
        .subquery('qBar')
    q_stream = sqlalchemy.union(q1.select(), q2.select())
    q_stream_alias = q_stream.alias('q_stream')
    qa = s.query(
            q_stream_alias.c.id.label('id_label'),
            sqlalchemy.func.max(q_stream_alias.c.event_timestamp).label('event_timestamp'),
        )\
        .join(
            Bar,
            q_stream_alias.c.id == Bar.id_field
        )\
        .group_by(
            q_stream_alias.c.id
        )\
        .order_by(
            sqlalchemy.desc('event_timestamp') 
        )
    print_query(qa)
    qa.count()


def query_form_b():
    """
    don't label the stream alias column
    status
        1.0.0b2 - fail
        0.9.9 - fail
    """
    q1 = s.query(
            Foo.id,
            Foo.timestamp_created.label('event_timestamp')
        )\
        .subquery('qFoo')
    q2 = s.query(
            Bar.id_field.label('id'),
            Bar.timestamp_created.label('event_timestamp')
        )\
        .subquery('qBar')
    q_stream = sqlalchemy.union(q1.select(), q2.select())
    q_stream_alias = q_stream.alias('q_stream')
    qa = s.query(
            q_stream_alias.c.id.label('id_label'),
            sqlalchemy.func.max(q_stream_alias.c.event_timestamp),
        )\
        .join(
            Bar,
            q_stream_alias.c.id == Bar.id_field
        )\
        .group_by(
            q_stream_alias.c.id
        )\
        .order_by(
            sqlalchemy.desc('event_timestamp') 
        )
    print_query(qa)
    qa.count()


def query_form_c():
    """
    rename the stream alias column with a label
    status
        1.0.0b2 - pass
        0.9.9 - pass
    """
    q1 = s.query(
            Foo.id,
            Foo.timestamp_created.label('event_timestamp')
        )\
        .subquery('qFoo')
    q2 = s.query(
            Bar.id_field.label('id'),
            Bar.timestamp_created.label('event_timestamp')
        )\
        .subquery('qBar')
    q_stream = sqlalchemy.union(q1.select(), q2.select())
    q_stream_alias = q_stream.alias('q_stream')
    qa = s.query(
            q_stream_alias.c.id.label('id_label'),
            sqlalchemy.func.max(q_stream_alias.c.event_timestamp).label('timestamp_renamed'),
        )\
        .join(
            Bar,
            q_stream_alias.c.id == Bar.id_field
        )\
        .group_by(
            q_stream_alias.c.id
        )\
        .order_by(
            sqlalchemy.desc('timestamp_renamed') 
        )
    print_query(qa)
    qa.count()


try:
    s.rollback()
    query_form_a()    
    print "pass - query_form_a"
except Exception as e:
    print e
    print "FAIL - query_form_a"

try:
    s.rollback()
    query_form_b()    
    print "pass - query_form_b"
except Exception as e:
    print e
    print "FAIL - query_form_b"

try:
    s.rollback()
    query_form_c()
    print "pass - query_form_c"
except Exception as e:
    print e
    print "FAIL - query_form_c"