from sqlalchemy import (create_engine, MetaData, Table, Column, Integer,
String, DateTime, Numeric, ForeignKey, and_)
-from sqlalchemy.orm import mapper, relationship, create_session
+from sqlalchemy.orm import mapper, relationship, Session
# Uncomment these to watch database activity.
#import logging
'item': relationship(Item, lazy='joined')
})
-session = create_session()
+session = Session()
# create our catalog
session.add(Item('SA T-Shirt', 10.99))
session.add(Item('SA Mug', 6.50))
session.add(Item('SA Hat', 8.99))
session.add(Item('MySQL Crowbar', 16.99))
-session.flush()
+session.commit()
# function to return items from the DB
def item(name):
order.order_items.append(OrderItem(item('MySQL Crowbar'), 10.99))
order.order_items.append(OrderItem(item('SA Hat')))
session.add(order)
-session.flush()
-
-session.expunge_all()
+session.commit()
# query the order, print items
order = session.query(Order).filter_by(customer_name='john smith').one()
from datetime import datetime
from sqlalchemy import (create_engine, MetaData, Table, Column, Integer,
String, DateTime, Float, ForeignKey, and_)
-from sqlalchemy.orm import mapper, relationship, create_session
+from sqlalchemy.orm import mapper, relationship, Session
from sqlalchemy.ext.associationproxy import AssociationProxy
engine = create_engine('sqlite://')
'item':relationship(Item, lazy='joined')
})
-session = create_session()
+session = Session()
# create our catalog
session.add_all([Item('SA T-Shirt', 10.99),
Item('SA Mug', 6.50),
Item('SA Hat', 8.99),
Item('MySQL Crowbar', 16.99)])
-session.flush()
+session.commit()
# function to return items
def item(name):
order.items.append(item('SA Hat'))
session.add(order)
-session.flush()
-
-session.expunge_all()
+session.commit()
# query the order, print items
order = session.query(Order).filter_by(customer_name='john smith').one()
-"""this example illustrates how to replace SQLAlchemy's class descriptors with a user-defined system.
+"""this example illustrates how to replace SQLAlchemy's class descriptors with
+a user-defined system.
-This sort of thing is appropriate for integration with frameworks that redefine class behaviors
-in their own way, such that SQLA's default instrumentation is not compatible.
+This sort of thing is appropriate for integration with frameworks that
+redefine class behaviors in their own way, such that SQLA's default
+instrumentation is not compatible.
-The example illustrates redefinition of instrumentation at the class level as well as the collection
-level, and redefines the storage of the class to store state within "instance._goofy_dict" instead
-of "instance.__dict__". Note that the default collection implementations can be used
-with a custom attribute system as well.
+The example illustrates redefinition of instrumentation at the class level as
+well as the collection level, and redefines the storage of the class to store
+state within "instance._goofy_dict" instead of "instance.__dict__". Note that
+the default collection implementations can be used with a custom attribute
+system as well.
"""
-from sqlalchemy import (create_engine, MetaData, Table, Column, Integer, Text,
- ForeignKey)
-from sqlalchemy.orm import (mapper, relationship, create_session,
- InstrumentationManager)
+from sqlalchemy import create_engine, MetaData, Table, Column, Integer, Text,\
+ ForeignKey
+from sqlalchemy.orm import mapper, relationship, Session,\
+ InstrumentationManager
-from sqlalchemy.orm.attributes import set_attribute, get_attribute, del_attribute, is_instrumented
+from sqlalchemy.orm.attributes import set_attribute, get_attribute, \
+ del_attribute, is_instrumented
from sqlalchemy.orm.collections import collection_adapter
def fire_append_event(self, item, initiator=None):
if initiator is not False and item is not None:
- self.state.get_impl(self.key).fire_append_event(self.state, self.state.dict, item,
- initiator)
+ self.state.get_impl(self.key).\
+ fire_append_event(self.state, self.state.dict, item,
+ initiator)
def fire_remove_event(self, item, initiator=None):
if initiator is not False and item is not None:
- self.state.get_impl(self.key).fire_remove_event(self.state, self.state.dict, item,
- initiator)
+ self.state.get_impl(self.key).\
+ fire_remove_event(self.state, self.state.dict, item,
+ initiator)
def fire_pre_remove_event(self, initiator=None):
- self.state.get_impl(self.key).fire_pre_remove_event(self.state, self.state.dict,
- initiator)
+ self.state.get_impl(self.key).\
+ fire_pre_remove_event(self.state, self.state.dict,
+ initiator)
class MyCollection(object):
def __init__(self):
if __name__ == '__main__':
meta = MetaData(create_engine('sqlite://'))
- table1 = Table('table1', meta, Column('id', Integer, primary_key=True), Column('name', Text))
- table2 = Table('table2', meta, Column('id', Integer, primary_key=True), Column('name', Text), Column('t1id', Integer, ForeignKey('table1.id')))
+ table1 = Table('table1', meta,
+ Column('id', Integer, primary_key=True),
+ Column('name', Text))
+ table2 = Table('table2', meta,
+ Column('id', Integer, primary_key=True),
+ Column('name', Text),
+ Column('t1id', Integer, ForeignKey('table1.id')))
meta.create_all()
class A(MyClass):
assert a1.bs[0].name == 'b1'
assert isinstance(a1.bs, MyCollection)
- sess = create_session()
+ sess = Session()
sess.add(a1)
- sess.flush()
- sess.expunge_all()
+ sess.commit()
a1 = sess.query(A).get(a1.id)
a1.bs.remove(a1.bs[0])
- sess.flush()
- sess.expunge_all()
+ sess.commit()
a1 = sess.query(A).get(a1.id)
assert len(a1.bs) == 1
sess = create_session()
for c in [tokyo, newyork, toronto, london, dublin, brasilia, quito]:
sess.add(c)
-sess.flush()
-
-sess.expunge_all()
+sess.commit()
t = sess.query(WeatherLocation).get(tokyo.id)
assert t.city == tokyo.city
from sqlalchemy import (MetaData, Table, Column, Integer, Unicode,
ForeignKey, UnicodeText, and_, not_, or_, String, Boolean, cast, text,
null, case)
- from sqlalchemy.orm import mapper, relationship, create_session
+ from sqlalchemy.orm import mapper, relationship, Session
from sqlalchemy.orm.collections import attribute_mapped_collection
metadata = MetaData()
metadata.bind = 'sqlite:///'
metadata.create_all()
- session = create_session()
+ session = Session()
stoat = Animal(u'stoat')
stoat[u'color'] = u'red'
stoat[u'weasel-like'] = True
session.add(stoat)
- session.flush()
- session.expunge_all()
+ session.commit()
critter = session.query(Animal).filter(Animal.name == u'stoat').one()
print critter[u'color']
critter[u'cuteness'] = u'very cute'
metadata.bind.echo = True
- session.flush()
+ session.commit()
metadata.bind.echo = False
marten = Animal(u'marten')
shrew[u'poisonous'] = True
session.add(shrew)
- session.flush()
+ session.commit()
q = (session.query(Animal).
filter(Animal.facts.any(
if __name__ == '__main__':
from sqlalchemy import (MetaData, Table, Column, Integer, Unicode,
ForeignKey, UnicodeText, and_, not_)
- from sqlalchemy.orm import mapper, relationship, create_session
+ from sqlalchemy.orm import mapper, relationship, Session
from sqlalchemy.orm.collections import attribute_mapped_collection
metadata = MetaData()
metadata.bind = 'sqlite:///'
metadata.create_all()
- session = create_session()
+ session = Session()
stoat = Animal(u'stoat')
stoat[u'color'] = u'reddish'
print stoat.facts[u'color']
session.add(stoat)
- session.flush()
- session.expunge_all()
+ session.commit()
critter = session.query(Animal).filter(Animal.name == u'stoat').one()
print critter[u'color']
print 'changing cuteness:'
metadata.bind.echo = True
- session.flush()
+ session.commit()
metadata.bind.echo = False
marten = Animal(u'marten')
loris[u'cuteness'] = u'fairly'
loris[u'poisonous-part'] = u'elbows'
session.add(loris)
- session.flush()
+ session.commit()
q = (session.query(Animal).
filter(Animal.facts.any(
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
