import new
-class MethodDescriptor(object):
+class hybrid(object):
def __init__(self, func):
self.func = func
def __get__(self, instance, owner):
else:
return new.instancemethod(self.func, instance, owner)
-class PropertyDescriptor(object):
- def __init__(self, fget, fset, fdel):
+class hybrid_property(object):
+ def __init__(self, fget, fset=None, fdel=None):
self.fget = fget
self.fset = fset
self.fdel = fdel
self.fset(instance, value)
def __delete__(self, instance):
self.fdel(instance)
-
-def hybrid(func):
- return MethodDescriptor(func)
-
-def hybrid_property(fget, fset=None, fdel=None):
- return PropertyDescriptor(fget, fset, fdel)
+
+ def setter(self, fset):
+ self.fset = fset
+ return self
### Example code
from sqlalchemy import MetaData, Table, Column, Integer
-from sqlalchemy.orm import mapper, create_session
+from sqlalchemy.orm import mapper, sessionmaker
metadata = MetaData('sqlite://')
metadata.bind.echo = True
print "Create the data"
-session = create_session()
+session = sessionmaker()()
intervals = [Interval1(1,4), Interval1(3,15), Interval1(11,16)]
session.add(interval)
session.add(Interval2(interval.start, interval.length))
-session.flush()
-
-print "Clear the cache and do some queries"
+session.commit()
-session.expunge_all()
for Interval in (Interval1, Interval2):
print "Querying using interval class %s" % Interval.__name__
print
print '-- length less than 10'
- print [(i, i.length) for i in session.query(Interval).filter(Interval.length < 10).all()]
+ print [(i, i.length) for i in
+ session.query(Interval).filter(Interval.length < 10).all()]
print
print '-- contains 12'
print
print '-- intersects 2..10'
other = Interval1(2,10)
- result = session.query(Interval).filter(Interval.intersects(other)).order_by(Interval.length).all()
+ result = session.query(Interval).\
+ filter(Interval.intersects(other)).\
+ order_by(Interval.length).all()
print [(interval, interval.intersects(other)) for interval in result]
-from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
-from sqlalchemy.orm import mapper, create_session, polymorphic_union
+from sqlalchemy import create_engine, MetaData, Table, Column, Integer, \
+ String
+from sqlalchemy.orm import mapper, sessionmaker, polymorphic_union
metadata = MetaData()
self.name = name
self.manager_data = manager_data
def __repr__(self):
- return self.__class__.__name__ + " " + self.name + " " + self.manager_data
+ return self.__class__.__name__ + " " + \
+ self.name + " " + self.manager_data
class Engineer(Employee):
def __init__(self, name, engineer_info):
self.name = name
self.engineer_info = engineer_info
def __repr__(self):
- return self.__class__.__name__ + " " + self.name + " " + self.engineer_info
+ return self.__class__.__name__ + " " + \
+ self.name + " " + self.engineer_info
pjoin = polymorphic_union({
}, 'type', 'pjoin')
employee_mapper = mapper(Employee, pjoin, polymorphic_on=pjoin.c.type)
-manager_mapper = mapper(Manager, managers_table, inherits=employee_mapper, concrete=True, polymorphic_identity='manager')
-engineer_mapper = mapper(Engineer, engineers_table, inherits=employee_mapper, concrete=True, polymorphic_identity='engineer')
+manager_mapper = mapper(Manager, managers_table,
+ inherits=employee_mapper, concrete=True,
+ polymorphic_identity='manager')
+engineer_mapper = mapper(Engineer, engineers_table,
+ inherits=employee_mapper, concrete=True,
+ polymorphic_identity='engineer')
-session = create_session(bind=engine)
+session = sessionmaker(engine)()
m1 = Manager("pointy haired boss", "manager1")
e1 = Engineer("wally", "engineer1")
session.add(m1)
session.add(e1)
session.add(e2)
-session.flush()
+session.commit()
-employees = session.query(Employee)
-print [e for e in employees]
+print session.query(Employee).all()
-from sqlalchemy import MetaData, Table, Column, Integer, String, ForeignKey
-from sqlalchemy.orm import mapper, relationship, create_session
-import sets
+from sqlalchemy import MetaData, Table, Column, Integer, String, \
+ ForeignKey, create_engine
+from sqlalchemy.orm import mapper, relationship, sessionmaker
+
# this example illustrates a polymorphic load of two classes
-metadata = MetaData('sqlite://')
-metadata.bind.echo = True
+metadata = MetaData()
# a table to store companies
companies = Table('companies', metadata,
Column('company_id', Integer, primary_key=True),
Column('name', String(50)))
-# we will define an inheritance relationship between the table "people" and "engineers",
-# and a second inheritance relationship between the table "people" and "managers"
+# we will define an inheritance relationship between the table "people" and
+# "engineers", and a second inheritance relationship between the table
+# "people" and "managers"
people = Table('people', metadata,
Column('person_id', Integer, primary_key=True),
Column('company_id', Integer, ForeignKey('companies.company_id')),
Column('type', String(30)))
engineers = Table('engineers', metadata,
- Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
+ Column('person_id', Integer, ForeignKey('people.person_id'),
+ primary_key=True),
Column('status', String(30)),
Column('engineer_name', String(50)),
Column('primary_language', String(50)),
)
managers = Table('managers', metadata,
- Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
+ Column('person_id', Integer, ForeignKey('people.person_id'),
+ primary_key=True),
Column('status', String(30)),
Column('manager_name', String(50))
)
-metadata.create_all()
-
# create our classes. The Engineer and Manager classes extend from Person.
class Person(object):
def __init__(self, **kwargs):
return "Ordinary person %s" % self.name
class Engineer(Person):
def __repr__(self):
- return "Engineer %s, status %s, engineer_name %s, primary_language %s" % (self.name, self.status, self.engineer_name, self.primary_language)
+ return "Engineer %s, status %s, engineer_name %s, "\
+ "primary_language %s" % \
+ (self.name, self.status,
+ self.engineer_name, self.primary_language)
class Manager(Person):
def __repr__(self):
- return "Manager %s, status %s, manager_name %s" % (self.name, self.status, self.manager_name)
+ return "Manager %s, status %s, manager_name %s" % \
+ (self.name, self.status, self.manager_name)
class Company(object):
def __init__(self, **kwargs):
for key, value in kwargs.iteritems():
return "Company %s" % self.name
-person_mapper = mapper(Person, people, polymorphic_on=people.c.type, polymorphic_identity='person')
-mapper(Engineer, engineers, inherits=person_mapper, polymorphic_identity='engineer')
-mapper(Manager, managers, inherits=person_mapper, polymorphic_identity='manager')
+person_mapper = mapper(Person, people, polymorphic_on=people.c.type,
+ polymorphic_identity='person')
+mapper(Engineer, engineers, inherits=person_mapper,
+ polymorphic_identity='engineer')
+mapper(Manager, managers, inherits=person_mapper,
+ polymorphic_identity='manager')
+
+mapper(Company, companies, properties={'employees'
+ : relationship(Person, lazy='joined', backref='company',
+ cascade='all, delete-orphan')})
+
+engine = create_engine('sqlite://', echo=True)
+
+metadata.create_all(engine)
-mapper(Company, companies, properties={
- 'employees': relationship(Person, lazy='joined', backref='company', cascade="all, delete-orphan")
-})
+session = sessionmaker(engine)()
-session = create_session()
c = Company(name='company1')
-c.employees.append(Manager(name='pointy haired boss', status='AAB', manager_name='manager1'))
-c.employees.append(Engineer(name='dilbert', status='BBA', engineer_name='engineer1', primary_language='java'))
+c.employees.append(Manager(name='pointy haired boss', status='AAB',
+ manager_name='manager1'))
+c.employees.append(Engineer(name='dilbert', status='BBA',
+ engineer_name='engineer1', primary_language='java'))
c.employees.append(Person(name='joesmith', status='HHH'))
-c.employees.append(Engineer(name='wally', status='CGG', engineer_name='engineer2', primary_language='python'))
-c.employees.append(Manager(name='jsmith', status='ABA', manager_name='manager2'))
+c.employees.append(Engineer(name='wally', status='CGG',
+ engineer_name='engineer2', primary_language='python'
+ ))
+c.employees.append(Manager(name='jsmith', status='ABA',
+ manager_name='manager2'))
session.add(c)
-session.flush()
-session.expunge_all()
+session.commit()
c = session.query(Company).get(1)
for e in c.employees:
print e, e._sa_instance_state.key, e.company
-assert sets.Set([e.name for e in c.employees]) == sets.Set(['pointy haired boss', 'dilbert', 'joesmith', 'wally', 'jsmith'])
+assert set([e.name for e in c.employees]) == set(['pointy haired boss',
+ 'dilbert', 'joesmith', 'wally', 'jsmith'])
print "\n"
dilbert = session.query(Person).filter_by(name='dilbert').one()
dilbert.engineer_name = 'hes dibert!'
-session.flush()
-session.expunge_all()
+session.commit()
c = session.query(Company).get(1)
for e in c.employees:
print e
session.delete(c)
-session.flush()
+session.commit()
-metadata.drop_all()
-from sqlalchemy import MetaData, Table, Column, Integer, String, ForeignKey
-from sqlalchemy.orm import mapper, relationship, create_session
+from sqlalchemy import MetaData, Table, Column, Integer, String, \
+ ForeignKey, create_engine
+from sqlalchemy.orm import mapper, relationship, sessionmaker
-metadata = MetaData('sqlite://')
-metadata.bind.echo = 'debug'
+metadata = MetaData()
# a table to store companies
companies = Table('companies', metadata,
Column('manager_name', String(50))
)
-metadata.create_all()
class Person(object):
def __init__(self, **kwargs):
return "Ordinary person %s" % self.name
class Engineer(Person):
def __repr__(self):
- return "Engineer %s, status %s, engineer_name %s, primary_language %s" % (self.name, self.status, self.engineer_name, self.primary_language)
+ return "Engineer %s, status %s, engineer_name %s, "\
+ "primary_language %s" % \
+ (self.name, self.status,
+ self.engineer_name, self.primary_language)
class Manager(Person):
def __repr__(self):
- return "Manager %s, status %s, manager_name %s" % (self.name, self.status, self.manager_name)
+ return "Manager %s, status %s, manager_name %s" % \
+ (self.name, self.status, self.manager_name)
class Company(object):
def __init__(self, **kwargs):
for key, value in kwargs.iteritems():
def __repr__(self):
return "Company %s" % self.name
-person_mapper = mapper(Person, employees_table, polymorphic_on=employees_table.c.type, polymorphic_identity='person')
-manager_mapper = mapper(Manager, inherits=person_mapper, polymorphic_identity='manager')
-engineer_mapper = mapper(Engineer, inherits=person_mapper, polymorphic_identity='engineer')
+person_mapper = mapper(Person, employees_table,
+ polymorphic_on=employees_table.c.type,
+ polymorphic_identity='person')
+manager_mapper = mapper(Manager, inherits=person_mapper,
+ polymorphic_identity='manager')
+engineer_mapper = mapper(Engineer, inherits=person_mapper,
+ polymorphic_identity='engineer')
mapper(Company, companies, properties={
'employees': relationship(Person, lazy=True, backref='company')
})
-session = create_session()
+
+engine = create_engine('sqlite:///', echo=True)
+
+metadata.create_all(engine)
+
+session = sessionmaker(engine)()
+
c = Company(name='company1')
-c.employees.append(Manager(name='pointy haired boss', status='AAB', manager_name='manager1'))
-c.employees.append(Engineer(name='dilbert', status='BBA', engineer_name='engineer1', primary_language='java'))
+c.employees.append(Manager(name='pointy haired boss', status='AAB',
+ manager_name='manager1'))
+c.employees.append(Engineer(name='dilbert', status='BBA',
+ engineer_name='engineer1', primary_language='java'))
c.employees.append(Person(name='joesmith', status='HHH'))
-c.employees.append(Engineer(name='wally', status='CGG', engineer_name='engineer2', primary_language='python'))
-c.employees.append(Manager(name='jsmith', status='ABA', manager_name='manager2'))
+c.employees.append(Engineer(name='wally', status='CGG',
+ engineer_name='engineer2', primary_language='python'
+ ))
+c.employees.append(Manager(name='jsmith', status='ABA',
+ manager_name='manager2'))
session.add(c)
-session.flush()
-
-session.expunge_all()
+session.commit()
c = session.query(Company).get(1)
for e in c.employees:
print e
session.delete(c)
-session.flush()
-
-metadata.drop_all()
+session.commit()
"""
"polymorphic" associations, ala ActiveRecord.
-In this example, we are specifically targeting this ActiveRecord functionality:
+In this example, we are specifically targeting this ActiveRecord
+functionality:
http://wiki.rubyonrails.org/rails/pages/UnderstandingPolymorphicAssociations
-The term "polymorphic" here means "object X can be referenced by objects A, B, and C,
-along a common line of association".
+The term "polymorphic" here means "object X can be referenced by objects A, B,
+and C, along a common line of association".
-In this example we illustrate the relationship in both directions.
-A little bit of property magic is used to smooth the edges.
+In this example we illustrate the relationship in both directions. A little
+bit of property magic is used to smooth the edges.
-AR creates this relationship in such a way that disallows
-any foreign key constraint from existing on the association.
-For a different way of doing this, see
-poly_assoc_fks.py. The interface is the same, the efficiency is more or less the same,
-but foreign key constraints may be used. That example also better separates
-the associated target object from those which associate with it.
+AR creates this relationship in such a way that disallows any foreign key
+constraint from existing on the association. For a different way of doing
+this, see poly_assoc_fks.py. The interface is the same, the efficiency is more
+or less the same, but foreign key constraints may be used. That example also
+better separates the associated target object from those which associate with
+it.
"""
from sqlalchemy import MetaData, Table, Column, Integer, String, and_
-from sqlalchemy.orm import (mapper, relationship, create_session, class_mapper,
- backref)
+from sqlalchemy.orm import mapper, relationship, sessionmaker, \
+ class_mapper, backref
metadata = MetaData('sqlite://')
class Address(object):
def __init__(self, type):
self.addressable_type = type
- member = property(lambda self: getattr(self, '_backref_%s' % self.addressable_type))
+ @property
+ def member(self):
+ return getattr(self, '_backref_%s' % self.addressable_type)
def addressable(cls, name, uselist=True):
"""addressable 'interface'.
- if you really wanted to make a "generic" version of this function, it's straightforward.
+ if you really wanted to make a "generic" version of this function, it's
+ straightforward.
"""
# create_address function, imitaes the rails example.
foreign_keys = [addresses.c.addressable_id]
mapper.add_property(name, relationship(
Address,
- primaryjoin=primaryjoin, uselist=uselist, foreign_keys=foreign_keys,
- backref=backref('_backref_%s' % table.name, primaryjoin=list(table.primary_key)[0] == addresses.c.addressable_id, foreign_keys=foreign_keys)
+ primaryjoin=primaryjoin,
+ uselist=uselist,
+ foreign_keys=foreign_keys,
+ backref=backref('_backref_%s' % table.name,
+ primaryjoin=list(table.primary_key)[0] ==\
+ addresses.c.addressable_id,
+ foreign_keys=foreign_keys)
)
)
a3 = o1.create_address()
a3.street = '444 park ave.'
-sess = create_session()
+sess = sessionmaker()()
sess.add(u1)
sess.add(o1)
-sess.flush()
-sess.expunge_all()
+sess.commit()
# query objects, get their addresses
"""
"polymorphic" associations, ala SQLAlchemy.
-See "poly_assoc.py" for an imitation of this functionality as implemented
-in ActiveRecord.
-
-Here, we build off the previous example, adding an association table
-that allows the relationship to be expressed as a many-to-one from the
-"model" object to its "association", so that each model table bears the foreign
-key constraint. This allows the same functionality via traditional
-normalized form with full constraints. It also isolates the target
-associated object from its method of being associated, allowing greater
-flexibility in its usage.
-
-As in the previous example, a little bit of property magic is used
-to smooth the edges.
-
-For a more genericized version of this example, see
-poly_assoc_generic.py.
+See "poly_assoc.py" for an imitation of this functionality as implemented in
+ActiveRecord.
+
+Here, we build off the previous example, adding an association table that
+allows the relationship to be expressed as a many-to-one from the "model"
+object to its "association", so that each model table bears the foreign key
+constraint. This allows the same functionality via traditional normalized form
+with full constraints. It also isolates the target associated object from its
+method of being associated, allowing greater flexibility in its usage.
+
+As in the previous example, a little bit of property magic is used to smooth
+the edges.
+
+For a more genericized version of this example, see poly_assoc_generic.py.
"""
-from sqlalchemy import MetaData, Table, Column, Integer, String, ForeignKey
-from sqlalchemy.orm import mapper, relationship, create_session, class_mapper
+from sqlalchemy import MetaData, Table, Column, Integer, String, \
+ ForeignKey
+from sqlalchemy.orm import mapper, relationship, sessionmaker, \
+ class_mapper
metadata = MetaData('sqlite://')
)
class Address(object):
- member = property(lambda self: getattr(self.association, '_backref_%s' % self.association.type))
+
+ @property
+ def member(self):
+ return getattr(self.association, '_backref_%s'
+ % self.association.type)
class AddressAssoc(object):
def __init__(self, name):
"""
mapper = class_mapper(cls)
table = mapper.local_table
- mapper.add_property('address_rel', relationship(AddressAssoc, backref='_backref_%s' % table.name))
+ mapper.add_property('address_rel',
+ relationship(AddressAssoc,
+ backref='_backref_%s' % table.name)
+ )
if uselist:
# list based property decorator
Column('id', Integer, primary_key=True),
Column('name', String(50), nullable=False),
# this column ties the users table into the address association
- Column('assoc_id', None, ForeignKey('address_associations.assoc_id'))
+ Column('assoc_id', Integer, ForeignKey('address_associations.assoc_id'))
)
class User(object):
Column('id', Integer, primary_key=True),
Column('description', String(50), nullable=False),
# this column ties the orders table into the address association
- Column('assoc_id', None, ForeignKey('address_associations.assoc_id'))
+ Column('assoc_id', Integer, ForeignKey('address_associations.assoc_id'))
)
class Order(object):
o1.address = Address()
o1.address.street = '444 park ave.'
-sess = create_session()
+sess = sessionmaker()()
sess.add(u1)
sess.add(o1)
-sess.flush()
-sess.expunge_all()
+sess.commit()
# query objects, get their addresses
bob = sess.query(User).filter_by(name='bob').one()
-assert [s.street for s in bob.addresses] == ['123 anywhere street', '345 orchard ave']
+assert [s.street for s in bob.addresses] == \
+ ['123 anywhere street', '345 orchard ave']
order = sess.query(Order).filter_by(description='order 1').one()
assert order.address.street == '444 park ave.'
"""
"polymorphic" associations, ala SQLAlchemy.
-This example generalizes the function in poly_assoc_pk.py into a
-function "association" which creates a new polymorphic association
-"interface".
+This example generalizes the function in poly_assoc_pk.py into a function
+"association" which creates a new polymorphic association "interface".
"""
-from sqlalchemy import MetaData, Table, Column, Integer, String, ForeignKey
-from sqlalchemy.orm import mapper, relationship, create_session, class_mapper
+from sqlalchemy import MetaData, Table, Column, Integer, String, \
+ ForeignKey
+from sqlalchemy.orm import mapper, relationship, sessionmaker, \
+ class_mapper
metadata = MetaData('sqlite://')
mapper = class_mapper(cls)
table = mapper.local_table
- mapper.add_property(attr_name, relationship(GenericAssoc, backref='_backref_%s' % table.name))
+ mapper.add_property(attr_name,
+ relationship(GenericAssoc,
+ backref='_backref_%s' % table.name)
+ )
if uselist:
# list based property decorator
setattr(self, attr_name, GenericAssoc(table.name))
getattr(self, attr_name).targets = [value]
setattr(cls, name, property(get, set))
-
- setattr(cls, 'member', property(lambda self: getattr(self.association, '_backref_%s' % self.association.type)))
+
+ @property
+ def member(self):
+ return getattr(self.association,
+ '_backref_%s' % self.association.type)
+
+ setattr(cls, 'member', member)
mapper(GenericAssoc, association_table, properties={
'targets':relationship(cls, backref='association'),
addresses = Table("addresses", metadata,
Column('id', Integer, primary_key=True),
- Column('assoc_id', None, ForeignKey('addresses_associations.assoc_id')),
+ Column('assoc_id', Integer, ForeignKey('addresses_associations.assoc_id')),
Column('street', String(100)),
Column('city', String(50)),
Column('country', String(50))
users = Table("users", metadata,
Column('id', Integer, primary_key=True),
Column('name', String(50), nullable=False),
- Column('assoc_id', None, ForeignKey('addresses_associations.assoc_id'))
+ Column('assoc_id', Integer, ForeignKey('addresses_associations.assoc_id'))
)
class User(object):
orders = Table("orders", metadata,
Column('id', Integer, primary_key=True),
Column('description', String(50), nullable=False),
- Column('assoc_id', None, ForeignKey('addresses_associations.assoc_id'))
+ Column('assoc_id', Integer, ForeignKey('addresses_associations.assoc_id'))
)
class Order(object):
o1.address = Address()
o1.address.street = '444 park ave.'
-sess = create_session()
+sess = sessionmaker()()
sess.add(u1)
sess.add(o1)
-sess.flush()
-
-sess.expunge_all()
+sess.commit()
# query objects, get their addresses
bob = sess.query(User).filter_by(name='bob').one()
-assert [s.street for s in bob.addresses] == ['123 anywhere street', '345 orchard ave']
+assert [s.street for s in bob.addresses] == \
+ ['123 anywhere street', '345 orchard ave']
order = sess.query(Order).filter_by(description='order 1').one()
assert order.address.street == '444 park ave.'
"""
-Illustrates an extension which creates version tables for entities and stores records for each change. The same idea as Elixir's versioned extension, but more efficient (uses attribute API to get history) and handles class inheritance. The given extensions generate an anonymous "history" class which represents historical versions of the target object.
+Illustrates an extension which creates version tables for entities and stores
+records for each change. The same idea as Elixir's versioned extension, but
+more efficient (uses attribute API to get history) and handles class
+inheritance. The given extensions generate an anonymous "history" class which
+represents historical versions of the target object.
-Usage is illustrated via a unit test module ``test_versioning.py``, which can be run via nose::
+Usage is illustrated via a unit test module ``test_versioning.py``, which can
+be run via nose::
nosetests -w examples/versioning/
all() \\
== [SomeClassHistory(version=1, name='sc1')]
-To apply ``VersionedMeta`` to a subset of classes (probably more typical), the metaclass can be applied on a per-class basis::
+To apply ``VersionedMeta`` to a subset of classes (probably more typical), the
+metaclass can be applied on a per-class basis::
from history_meta import VersionedMeta, VersionedListener
# ...
-The ``VersionedMeta`` is a declarative metaclass - to use the extension with plain mappers, the ``_history_mapper`` function can be applied::
+The ``VersionedMeta`` is a declarative metaclass - to use the extension with
+plain mappers, the ``_history_mapper`` function can be applied::
from history_meta import _history_mapper
def _history_mapper(local_mapper):
cls = local_mapper.class_
- # SLIGHT SQLA HACK #1 - set the "active_history" flag
+ # set the "active_history" flag
# on on column-mapped attributes so that the old version
# of the info is always loaded (currently sets it on all attributes)
for prop in local_mapper.iterate_properties:
obj_col = om.local_table.c[hist_col.key]
- # SLIGHT SQLA HACK #3 - get the value of the
+ # get the value of the
# attribute based on the MapperProperty related to the
# mapped column. this will allow usage of MapperProperties
# that have a different keyname than that of the mapped column.
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
