Hey All, I'm running into an interesting problem. There's a quick obvious solution, but I think it's a placebo, so I want to run this by everyone else first.
In this case (attached .sql creates the necessary schema) I have three interconnected tables that I'm viewing using reflection. This is my normal setup which functions without any problems. However, I recently created a mapper sourcing from a subselect, and discovered an interesting problem crop up. At one point, the foreign key is being retrieved from the database as unicode, but is not being converted to a Python string. This causes an 'if type(...) == str' check to fail, at which point it's expected to be an object and things fail. If I set convert_unicode=False, I don't have this problem (but, of course, other things elsewhere fail...) As an interesting complication, I was unable to reproduce this effect in the attached test case (pg_mapper_select_working.py) which, as far as I can determine, almost exactly reproduces the schema generated by pg_audit_pub.sql. Attached is my system information, but in short I'm running 1368 with the DISCRETE ON() patch and modula support for PostgreSQL 8.x. Reproduce this by loading the pg_audit_pub.sql schema into your 'test' database and executing the pg_audit_pub.py with appropriate credentials. Hopefully this helps! Cheers, -G
pg_audit_pub.sql
Description: Binary data
pg_audit_pub.log
Description: Binary data
import sys
from sqlalchemy import *
# For verbose logging, change the echo and LOG booleans to True
engine = create_engine('postgres', {'database':'test', 'host':'localhost',
'user':'xxx', 'password':'xxx'}, echo=True, echo_uow=True,
convert_unicode=True)
objectstore.LOG = True
class AuditItem(object):
pass
audit_trail = Table('audit_trail', engine, autoload = True, schema =
'equipment')
items_table = Table('items', engine, autoload = True, schema = 'equipment')
audit_item = select([audit_trail.c.ts, audit_trail.c.event_type,
audit_trail.c.user_id,
audit_trail.c.new_state_id, audit_trail.c.owner_id,
items_table.c.item_id,
items_table.c.state_id],
from_obj=[outerjoin(items_table, audit_trail)],
order_by=[items_table.c.item_id,
desc(audit_trail.c.ts)]).alias('audit_items')
AuditItem = mapper(AuditItem, audit_item)
a_items = AuditItem.select()
for i in a_items:
print i.item_id, i.state_id, i.ts, i.event_type, i.user_id,
i.new_state_id, i.owner_id
import sys
from sqlalchemy import *
def CreateTables():
engine = create_engine('postgres', {'database':'test',
'host':'localhost', 'user':'postgres', 'password':'abcabc'},
echo=True, echo_uow=True,
convert_unicode=True)
objectstore.LOG = True
states_table = Table('states', engine,
Column('state_id', Integer,
primary_key=True),
Column('state_name', String),
schema='public')
items_table = Table('items', engine,
Column('item_id', Integer,
primary_key=True),
Column('state_id', Integer,
ForeignKey(states_table.c.state_id)),
Column('details', Integer),
schema='public')
events_table = Table('events', engine,
Column('event_id', Integer,
primary_key=True),
Column('item_id', Integer,
ForeignKey(items_table.c.item_id)),
Column('new_state_id',
Integer, ForeignKey(states_table.c.state_id)),
Column('ts', Integer),
schema='public')
states_table.create()
items_table.create()
events_table.create()
states_table.insert().execute(
{'state_id':10000, 'state_name':'Purple 10k'},
{'state_id':20000, 'state_name':'Orange 20k'},
{'state_id':30000, 'state_name':'Violet 30k'},
{'state_id':40000, 'state_name':'Yellow 40k'})
items_table.insert().execute(
{'item_id':1, 'state_id':10000, 'details':100},
{'item_id':2, 'state_id':10000, 'details':200},
{'item_id':3, 'state_id':30000, 'details':300},
{'item_id':4, 'state_id':30000, 'details':400},
{'item_id':5, 'state_id':30000, 'details':500})
events_table.insert().execute(
{'event_id':5000, 'new_state_id':20000, 'item_id':2,
'ts':'5522'},
{'event_id':6000, 'new_state_id':20000, 'item_id':3,
'ts':'6633'},
{'event_id':7000, 'new_state_id':20000, 'item_id':5,
'ts':'7755'},
{'event_id':8000, 'new_state_id':20000, 'item_id':1,
'ts':'8811'},
{'event_id':9000, 'new_state_id':40000, 'item_id':3,
'ts':'9933'})
engine = None
def TestMappedSelect(convert_unicode):
try:
engine = create_engine('postgres', {'database':'test',
'host':'localhost', 'user':'postgres', 'password':'abcabc'},
echo=True,
echo_uow=True, convert_unicode=convert_unicode)
objectstore.LOG = True
class State(object):
pass
class Item(object):
pass
class Event(object):
pass
class EventItem(object):
pass
states_table = Table('states', engine, autoload=True,
schema='public')
items_table = Table('items', engine, autoload=True,
schema='public')
events_table = Table('events', engine, autoload=True,
schema='public')
event_items_view = select([events_table.c.new_state_id,
events_table.c.ts, items_table.c.item_id,
items_table.c.state_id, items_table.c.details],
from_obj=[outerjoin(items_table, events_table)],
distinct=[items_table.c.item_id],
order_by=[items_table.c.item_id,
desc(events_table.c.ts)]).alias('event_items_view')
assign_mapper(State, states_table)
assign_mapper(Item, items_table)
assign_mapper(Event, events_table)
assign_mapper(EventItem, event_items_view)
foo = EventItem.select()
print 'ID', 'TS ', 'Details'
for i in foo:
print i.item_id, '', i.ts, i.details
finally:
events_table.drop()
items_table.drop()
states_table.drop()
CreateTables()
TestMappedSelect(False)
CreateTables()
TestMappedSelect(True)
Windows XP SP2 Cygwin SQLAlchemy 1368 modified with MOD patch and DISCRETE ON() patch. PostgreSQL 8.x
