Test overriding reflected column types in with sqlalchemy.

test-reflect.py

#!/usr/bin/env python
""" Demonstrates an issue with overriding reflected column types.

Author:  Ralph Bean <ralph.bean@gmail.com>

Take a look at :meth:`reflect_and_verify_legacy_db_mangle`.

Running this script produces the following output::

     % ./test-reflect.py

    Using dbfile: mysql://test:test@localhost/testdb

    The value in the db was 1314725346 of type <type 'long'>
    Traceback (most recent call last):
      File "./test-reflect.py", line 130, in <module>
        reflect_and_verify_legacy_db_mangle(engine)  # fails
      File "./test-reflect.py", line 115, in reflect_and_verify_legacy_db_mangle
        assert(type(from_db) == type(once_upon_a_datetime))     # This fails
    AssertionError
"""

import tempfile
import time
import datetime

from sqlalchemy import create_engine, MetaData
from sqlalchemy import Table, Column, Integer, DateTime
from sqlalchemy import event
from sqlalchemy import types
from sqlalchemy.orm import scoped_session, sessionmaker, mapper
from sqlalchemy.ext.declarative import declarative_base

maker = sessionmaker()
DeclarativeBase = declarative_base()


# Constants we use for testing
once_upon_a_time = int(time.time())
once_upon_a_datetime = datetime.datetime.fromtimestamp(once_upon_a_time)


class Thing(DeclarativeBase):
    __tablename__ = 'thing_table'
    id = Column(Integer, primary_key=True)
    create_time = Column(Integer)


def setup_pretend_legacy_db(engine):
    """ Setup a pretend legacy db that we'll reflect later.  This works. """

    DBSession = scoped_session(maker)
    DBSession.configure(bind=engine)
    metadata = DeclarativeBase.metadata
    metadata.drop_all(engine)  # Lordy!
    metadata.create_all(engine)

    t = Thing(create_time=once_upon_a_time)
    DBSession.add(t)
    DBSession.commit()


def verify_legacy_db_integer(engine):
    """ Verify that the legacy db works as we expect.  This works. """

    DBSession = scoped_session(maker)
    DBSession.configure(bind=engine)
    assert(DBSession.query(Thing).one().create_time == once_upon_a_time)


def reflect_and_verify_legacy_db_normal(engine):
    """ Verify that normal old reflection works as we expected.  It does. """

    DBSession = scoped_session(maker)
    DBSession.configure(bind=engine)
    metadata = MetaData(engine.url)
    table = Table('thing_table', metadata, autoload=True)

    class Thing(object):
        pass

    mapper(Thing, table)
    assert(DBSession.query(Thing).one().create_time == once_upon_a_time)


def reflect_and_verify_legacy_db_mangle(engine):
    """ The bulk of the problem.

    I have a database with tons of tables (many are dynamically created by
    another app).  Each table has a few '*_time' fields that are stored as
    Integers.  I can succesfully reflect the db (and the dynamic tables) but it
    would be sweet if any '*_time' column played out in python as a datetime
    object, not an int.

    Here I use the events system to modify the column_info.  It works, but
    fails on the query.

    Question is:
        1) Am I doing this wrong?  Is SQLAlchemy not supposed to be able to
        support this kind of thing?
        2) Is this a bug in SQLAlchemy somewhere.  If so, I'd love to turn this
        into a patch for the test-suite.
    """

    def listen_for_reflect(table, column_info):
        if 'time' in column_info['name']:
            column_info['type'] = types.DateTime()

    event.listen(Table, 'column_reflect', listen_for_reflect)

    DBSession = scoped_session(maker)
    DBSession.configure(bind=engine)
    metadata = MetaData(engine.url)
    table = Table('thing_table', metadata, autoload=True)
    t = table.columns.create_time.type

    # Here's the crazy thing.  It works as far as the table is concerned.
    assert(type(table.columns.create_time.type) == types.DateTime)

    class Thing(object):
        pass

    mapper(Thing, table)

    from_db = DBSession.query(Thing).one().create_time
    print "The value in the db was", from_db, "of type", type(from_db)

    # But the mapped class returns not what you'd expect.
    assert(type(from_db) == type(once_upon_a_datetime))     # This fails
    assert(from_db == once_upon_a_datetime)                 # This fails too


if __name__ == '__main__':
    url = "mysql://test:test@localhost/testdb"
    engine = create_engine(url)

    print
    print "Using dbfile:", engine.url
    print

    setup_pretend_legacy_db(engine)              # succeeds
    verify_legacy_db_integer(engine)             # succeeds
    reflect_and_verify_legacy_db_normal(engine)  # succeeds
    reflect_and_verify_legacy_db_mangle(engine)  # fails