===================================================================
# -*- coding: cp1251 -*-
from sqlalchemy import *
import datetime
class SqlStrMixing( object ):
def __str__( self ):
s = [ self.__class__.__name__ + ': ' ]
for c in self.c:
s.append( u'%s=%s ' % ( c.key, getattr(self, c.key) ) )
result = ''.join(s).encode( 'cp866' )
return result
# Sqlite and postgres (psycopg2) give different results
engine_uri = 'sqlite:///'
#engine_uri = 'postgres://billing:[EMAIL PROTECTED]:5432/billing'
metadata = BoundMetaData(engine_uri,
convert_unicode=True)
people = Table('people', metadata,
Column('person_id', Integer, primary_key=True),
Column('name', String(50)),
Column('date', DateTime ),
Column('type', String(30)))
engineers = Table('engineers', metadata,
Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
Column('engineer_info', String(50)),
)
managers = Table('managers', metadata,
Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
Column('manager_data', String(50)),
)
addresses = Table( 'addresses', metadata,
Column('address_id', Integer, primary_key=True),
Column('person_id', Integer, ForeignKey('people.person_id') ),
Column('address', String(150)),
)
metadata.drop_all()
metadata.create_all()
person_join = polymorphic_union(
{
'engineer': people.join(engineers),
'manager': people.join(managers),
'person': people.select(people.c.type=='person'),
}, None, 'pjoin')
class Person(SqlStrMixing):
def __init__(self, **kwargs):
for key, value in kwargs.iteritems():
setattr(self, key, value)
self.date = datetime.datetime.today()
def __repr__(self):
return self.__class__.__name__ + " " + self.name
class Manager(Person):
def __repr__(self):
return self.__class__.__name__ + " " + self.name + " " + self.manager_data
class Engineer(Person):
def __repr__(self):
return self.__class__.__name__ + " " + self.name + " " + self.engineer_info
class Address(SqlStrMixing):
def __init__(self, **kwargs):
for key, value in kwargs.iteritems():
setattr(self, key, value)
mapper( Address, addresses )
person_mapper = mapper(Person, people, select_table=person_join, \
polymorphic_on=person_join.c.type, polymorphic_identity='person',
properties= \
{
'addresses' : relation( Address, backref='person' )
})
mapper( Engineer, engineers, inherits=person_mapper, \
polymorphic_identity='engineer' )
mapper( Manager, managers, inherits=person_mapper, \
polymorphic_identity='manager' )
session = create_session(echo_uow=False)
p1 = Engineer( name=u'John', engineer_info=u'some info' )
p2 = Manager( name=u'Jack', manager_data=u'some_data' )
session.save(p1)
session.save(p2)
session.flush()
session.close()
session = create_session(echo_uow=False)
person_list = session.query(Person).select()
for p in person_list:
print type(p.name)
print p.name
try:
print type(p.engineer_info)
print p.engineer_info
except:
pass
try:
print type(p.manager_data)
print p.manager_data
except:
pass
metadata.drop_all()
