"""The collection of modern alternatives to deprecated & removed functionality.

Collects specimens of old ORM code and explicitly covers the recommended
modern (i.e. not deprecated) alternative to them.  The tests snippets here can
be migrated directly to the wiki, docs, etc.

"""
import testenv; testenv.configure_for_tests()
from testlib import testing
from testlib.sa import Table, Column, Integer, String, ForeignKey, func
from testlib.sa.orm import mapper, relation, create_session, sessionmaker
from orm import _base


class QueryAlternativesTest(_base.MappedTest):
    '''Collects modern idioms for Queries

    The docstring for each test case serves as miniature documentation about
    the deprecated use case, and the test body illustrates (and covers) the
    intended replacement code to accomplish the same task.

    Documenting the "old way" including the argument signature helps these
    cases remain useful to readers even after the deprecated method has been
    removed from the modern codebase.

    Format:

    def test_deprecated_thing(self):
        """Query.methodname(old, arg, **signature)

        output = session.query(User).deprecatedmethod(inputs)

        """
        # 0.4+
        output = session.query(User).newway(inputs)
        assert output is correct

        # 0.5+
        output = session.query(User).evennewerway(inputs)
        assert output is correct

    '''

    run_inserts = 'once'
    run_deletes = None

    def define_tables(self, metadata):
        Table('users_table', metadata,
              Column('id', Integer, primary_key=True),
              Column('name', String(64)))

        Table('addresses_table', metadata,
              Column('id', Integer, primary_key=True),
              Column('user_id', Integer, ForeignKey('users_table.id')),
              Column('email_address', String(128)),
              Column('purpose', String(16)),
              Column('bounces', Integer, default=0))

    def setup_classes(self):
        class User(_base.BasicEntity):
            pass

        class Address(_base.BasicEntity):
            pass

    @testing.resolve_artifact_names
    def setup_mappers(self):
        mapper(User, users_table, properties=dict(
            addresses=relation(Address, backref='user'),
            ))
        mapper(Address, addresses_table)

    def fixtures(self):
        return dict(
            users_table=(
            ('id', 'name'),
            (1, 'jack'),
            (2, 'ed'),
            (3, 'fred'),
            (4, 'chuck')),

            addresses_table=(
            ('id', 'user_id', 'email_address', 'purpose', 'bounces'),
            (1, 1, 'jack@jack.home', 'Personal', 0),
            (2, 1, 'jack@jack.bizz', 'Work', 1),
            (3, 2, 'ed@foo.bar', 'Personal', 0),
            (4, 3, 'fred@the.fred', 'Personal', 10)))


    ######################################################################

    @testing.resolve_artifact_names
    def test_override_get(self):
        """MapperExtension.get()
        
        x = session.query.get(5)
        
        """
        from sqlalchemy.orm.query import Query
        cache = {}
        class MyQuery(Query):
            def get(self, ident, **kwargs):
                if ident in cache:
                    return cache[ident]
                else:
                    x = super(MyQuery, self).get(ident)
                    cache[ident] = x
                    return x
                    
        session = sessionmaker(query_cls=MyQuery)()
        
        ad1 = session.query(Address).get(1)
        assert ad1 in cache.values()
    
    @testing.resolve_artifact_names
    def test_load(self):
        """x = session.query(Address).load(1)
            
            x = session.load(Address, 1)
        
        """

        session = create_session()
        ad1 = session.query(Address).populate_existing().get(1)
        assert bool(ad1)
        
        
    @testing.resolve_artifact_names
    def test_apply_max(self):
        """Query.apply_max(col)

        max = session.query(Address).apply_max(Address.bounces)

        """
        session = create_session()

        # 0.5.0
        maxes = list(session.query(Address).values(func.max(Address.bounces)))
        max = maxes[0][0]
        assert max == 10

        max = session.query(func.max(Address.bounces)).one()[0]
        assert max == 10

    @testing.resolve_artifact_names
    def test_apply_min(self):
        """Query.apply_min(col)

        min = session.query(Address).apply_min(Address.bounces)

        """
        session = create_session()

        # 0.5.0
        mins = list(session.query(Address).values(func.min(Address.bounces)))
        min = mins[0][0]
        assert min == 0

        min = session.query(func.min(Address.bounces)).one()[0]
        assert min == 0

    @testing.resolve_artifact_names
    def test_apply_avg(self):
        """Query.apply_avg(col)

        avg = session.query(Address).apply_avg(Address.bounces)

        """
        session = create_session()

        avgs = list(session.query(Address).values(func.avg(Address.bounces)))
        avg = avgs[0][0]
        assert avg > 0 and avg < 10

        avg = session.query(func.avg(Address.bounces)).one()[0]
        assert avg > 0 and avg < 10

    @testing.resolve_artifact_names
    def test_apply_sum(self):
        """Query.apply_sum(col)

        avg = session.query(Address).apply_avg(Address.bounces)

        """
        session = create_session()

        avgs = list(session.query(Address).values(func.sum(Address.bounces)))
        avg = avgs[0][0]
        assert avg == 11

        avg = session.query(func.sum(Address.bounces)).one()[0]
        assert avg == 11

    @testing.resolve_artifact_names
    def test_count_by(self):
        """Query.count_by(*args, **params)

        num = session.query(Address).count_by(purpose='Personal')

        # old-style implicit *_by join
        num = session.query(User).count_by(purpose='Personal')

        """
        session = create_session()

        num = session.query(Address).filter_by(purpose='Personal').count()
        assert num == 3, num

        num = (session.query(User).join('addresses').
               filter(Address.purpose=='Personal')).count()
        assert num == 3, num

    @testing.resolve_artifact_names
    def test_count_whereclause(self):
        """Query.count(whereclause=None, params=None, **kwargs)

        num = session.query(Address).count(address_table.c.bounces > 1)

        """
        session = create_session()

        num = session.query(Address).filter(Address.bounces > 1).count()
        assert num == 1, num

    @testing.resolve_artifact_names
    def test_execute(self):
        """Query.execute(clauseelement, params=None, *args, **kwargs)

        users = session.query(User).execute(users_table.select())

        """
        session = create_session()

        users = session.query(User).from_statement(users_table.select()).all()
        assert len(users) == 4

    @testing.resolve_artifact_names
    def test_get_by(self):
        """Query.get_by(*args, **params)

        user = session.query(User).get_by(name='ed')

        # 0.3-style implicit *_by join
        user = session.query(User).get_by(email_addresss='fred@the.fred')

        """
        session = create_session()

        user = session.query(User).filter_by(name='ed').first()
        assert user.name == 'ed'

        user = (session.query(User).join('addresses').
                filter(Address.email_address=='fred@the.fred')).first()
        assert user.name == 'fred'

        user = session.query(User).filter(
            User.addresses.any(Address.email_address=='fred@the.fred')).first()
        assert user.name == 'fred'

    @testing.resolve_artifact_names
    def test_instances_entities(self):
        """Query.instances(cursor, *mappers_or_columns, **kwargs)

        sel = users_table.join(addresses_table).select(use_labels=True)
        res = session.query(User).instances(sel.execute(), Address)

        """
        session = create_session()

        sel = users_table.join(addresses_table).select(use_labels=True)
        res = list(session.query(User, Address).instances(sel.execute()))

        assert len(res) == 4
        cola, colb = res[0]
        assert isinstance(cola, User) and isinstance(colb, Address)

    @testing.resolve_artifact_names
    def test_join_by(self):
        """Query.join_by(*args, **params)

        TODO
        """
        session = create_session()


    @testing.resolve_artifact_names
    def test_join_to(self):
        """Query.join_to(key)

        TODO
        """
        session = create_session()


    @testing.resolve_artifact_names
    def test_join_via(self):
        """Query.join_via(keys)

        TODO
        """
        session = create_session()


    @testing.resolve_artifact_names
    def test_list(self):
        """Query.list()

        users = session.query(User).list()

        """
        session = create_session()

        users = session.query(User).all()
        assert len(users) == 4

    @testing.resolve_artifact_names
    def test_scalar(self):
        """Query.scalar()

        user = session.query(User).filter(User.id==1).scalar()

        """
        session = create_session()

        user = session.query(User).filter(User.id==1).first()
        assert user.id==1

    @testing.resolve_artifact_names
    def test_select(self):
        """Query.select(arg=None, **kwargs)

        users = session.query(User).select(users_table.c.name != None)

        """
        session = create_session()

        users = session.query(User).filter(User.name != None).all()
        assert len(users) == 4

    @testing.resolve_artifact_names
    def test_select_by(self):
        """Query.select_by(*args, **params)

        users = session.query(User).select_by(name='fred')

        # 0.3 magic join on *_by methods
        users = session.query(User).select_by(email_address='fred@the.fred')

        """
        session = create_session()

        users = session.query(User).filter_by(name='fred').all()
        assert len(users) == 1

        users = session.query(User).filter(User.name=='fred').all()
        assert len(users) == 1

        users = (session.query(User).join('addresses').
                 filter_by(email_address='fred@the.fred')).all()
        assert len(users) == 1

        users = session.query(User).filter(User.addresses.any(
            Address.email_address == 'fred@the.fred')).all()
        assert len(users) == 1

    @testing.resolve_artifact_names
    def test_selectfirst(self):
        """Query.selectfirst(arg=None, **kwargs)

        bounced = session.query(Address).selectfirst(
          addresses_table.c.bounces > 0)

        """
        session = create_session()

        bounced = session.query(Address).filter(Address.bounces > 0).first()
        assert bounced.bounces > 0

    @testing.resolve_artifact_names
    def test_selectfirst_by(self):
        """Query.selectfirst_by(*args, **params)

        onebounce = session.query(Address).selectfirst_by(bounces=1)

        # 0.3 magic join on *_by methods
        onebounce_user = session.query(User).selectfirst_by(bounces=1)

        """
        session = create_session()

        onebounce = session.query(Address).filter_by(bounces=1).first()
        assert onebounce.bounces == 1

        onebounce_user = (session.query(User).join('addresses').
                          filter_by(bounces=1)).first()
        assert onebounce_user.name == 'jack'

        onebounce_user = (session.query(User).join('addresses').
                          filter(Address.bounces == 1)).first()
        assert onebounce_user.name == 'jack'

        onebounce_user = session.query(User).filter(User.addresses.any(
            Address.bounces == 1)).first()
        assert onebounce_user.name == 'jack'

    @testing.resolve_artifact_names
    def test_selectone(self):
        """Query.selectone(arg=None, **kwargs)

        ed = session.query(User).selectone(users_table.c.name == 'ed')

        """
        session = create_session()

        ed = session.query(User).filter(User.name == 'jack').one()

    @testing.resolve_artifact_names
    def test_selectone_by(self):
        """Query.selectone_by

        ed = session.query(User).selectone_by(name='ed')

        # 0.3 magic join on *_by methods
        ed = session.query(User).selectone_by(email_address='ed@foo.bar')

        """
        session = create_session()

        ed = session.query(User).filter_by(name='jack').one()

        ed = session.query(User).filter(User.name == 'jack').one()

        ed = session.query(User).join('addresses').filter(
            Address.email_address == 'ed@foo.bar').one()

        ed = session.query(User).filter(User.addresses.any(
            Address.email_address == 'ed@foo.bar')).one()

    @testing.resolve_artifact_names
    def test_select_statement(self):
        """Query.select_statement(statement, **params)

        users = session.query(User).select_statement(users_table.select())

        """
        session = create_session()

        users = session.query(User).from_statement(users_table.select()).all()
        assert len(users) == 4

    @testing.resolve_artifact_names
    def test_select_text(self):
        """Query.select_text(text, **params)

        users = session.query(User).select_text('SELECT * FROM users_table')

        """
        session = create_session()

        users = (session.query(User).
                 from_statement('SELECT * FROM users_table')).all()
        assert len(users) == 4

    @testing.resolve_artifact_names
    def test_select_whereclause(self):
        """Query.select_whereclause(whereclause=None, params=None, **kwargs)


        users = session,query(User).select_whereclause(users.c.name=='ed')
        users = session.query(User).select_whereclause("name='ed'")

        """
        session = create_session()

        users = session.query(User).filter(User.name=='ed').all()
        assert len(users) == 1 and users[0].name == 'ed'

        users = session.query(User).filter("name='ed'").all()
        assert len(users) == 1 and users[0].name == 'ed'


if __name__ == '__main__':
    testenv.main()