- Session now has tranasaction API built into it, SessionTransaction not part of the usual API now (but still does the work)
- integrated nested transactions/SAVEPOINT with Session transactional system
- Session gets "transactional" mode, it always begins a transaction and reopens after commit/rollback
- Session gets "autoflush=True" flag, implies "transactional", issues a flush before the execution of every
query so that query results include whatever is currently in the session
- various unit test fixes
information, and also check out the unit tests for examples of stepped
numbering, alphabetical and Fibonacci numbering.
-### threadlocal
-
-**Author:** Mike Bayer and Daniel Miller
-
-`threadlocal` is an extension that was created primarily to provide backwards compatibility with the older SQLAlchemy 0.1 series. It uses three features which SQLAlchemy 0.2 and above provide as distinct features: `SessionContext`, `assign_mapper`, and the `TLEngine`, which is the `Engine` used with the threadlocal `create_engine()` strategy. It is **strongly** recommended that these three features are understood individually before using threadlocal.
-
-In SQLAlchemy 0.1, users never dealt with explcit connections and didn't have a very explicit `Session` interface, instead relying upon a more magical global object called `objectstore`. The `objectstore` idea was wildly popular with about half of SA's users, and completely unpopular with the other half. The threadlocal mod basically brings back `objectstore`, which is in fact just a `SessionContext` where you can call `Session` methods directly off of it, instead of saying `context.current`. For `threadlocal` to faithfully produce 0.1 behavior, it is invoked as a *mod* which globally installs the objectstore's mapper extension, such that all `Mapper`s will automatically assign all new instances of mapped classes to the objectstore's contextual `Session`. Additionally, it also changes the default engine strategy used by `create_engine` to be the "threadlocal" strategy, which in normal practice does not affect much.
-
-When you import threadlocal, what you get is:
-
-* the "objectstore" session context object is now added to the `sqlalchemy` namespace.
-* a global `MapperExtension` is set up for all mappers which assigns "objectstore"'s session as the default session context, used by new instances as well as `Query` objects (see the section [plugins_sessioncontext_sessioncontextext](rel:plugins_sessioncontext_sessioncontextext)).
-* a new function "assign_mapper" is added to the `sqlalchemy` namespace, which calls the `assignmapper` mapper function using the new "objectstore" context.
-* the `create_engine` function is modified so that "threadlocal", and not "plain", is the default engine strategy.
-
-So an important point to understand is, **don't use the threadlocal mod unless you explcitly are looking for that behavior**. Unfortunately, the easy import of the "threadlocal" mod has found its way into several tutorials on external websites, which produces application-wide behavior that is in conflict with the SQLAlchemy tutorial and data mapping documentation.
-
-While "threadlocal" is only about 10 lines of code, it is strongly advised that users instead make usage of `SessionContext` and `assign_mapper` explictly to eliminate confusion. Additionally, the "threadlocal" strategy on `create_engine()` also exists primarily to provide patterns used in 0.1 and is probably not worth using either, unless you specifically need those patterns.
-
-Basic usage of threadlocal involves importing the mod, *before* any usage of the `sqlalchemy` namespace, since threadlocal is going to add the "objectstore" and "assign_mapper" keywords to "sqlalchemy".
-
-To use `objectstore`:
-
- {python}
- import sqlalchemy.mods.threadlocal
- from sqlalchemy import *
-
- metadata = MetaData('sqlite:///')
- user_table = Table('users', metadata,
- Column('user_id', Integer, primary_key=True),
- Column('user_name', String(50), nullable=False)
- )
-
- class User(object):
- pass
- mapper(User, user_table)
-
- # "user" object is added to the session automatically
- user = User()
-
- # flush the contextual session
- objectstore.flush()
-
-The actual `Session` is available as:
-
- {python}
- objectstore.get_session()
-
-To use `assign_mapper`:
-
- {python}
- import sqlalchemy.mods.threadlocal
- from sqlalchemy import *
-
- metadata = MetaData('sqlite:///')
- user_table = Table('users', metadata,
- Column('user_id', Integer, primary_key=True),
- Column('user_name', String(50), nullable=False)
- )
-
- class User(object):
- pass
-
- # note that no "context" object is needed
- assign_mapper(User, user_table)
-
- # call up a user
- user = User.selectfirst(user_table.c.user_id==7)
-
- # call 'delete' on the user
- user.delete()
-
- # flush
- objectstore.flush()
-
### ActiveMapper
Full SqlSoup documentation is on the [SQLAlchemy Wiki](http://www.sqlalchemy.org/trac/wiki/SqlSoup).
-### ProxyEngine
-
-This extension is temporarily removed from the 0.4 series of SQLAlchemy. A newer method of SQL instrumentation will eventually be re-introduced.
As you might have guessed, we have just defined a table named `users` which has three columns: `user_id` (which is a primary key column), `user_name` and `password`. Currently it is just an object that doesn't necessarily correspond to an existing table in our database. To actually create the table, we use the `create()` method. To make it interesting, we will have SQLAlchemy echo the SQL statements it sends to the database, by setting the `echo` flag on the `Engine` associated with our `MetaData`:
{python}
- >>> metadata.engine.echo = True
+ >>> metadata.bind.echo = True
>>> users_table.create() # doctest:+ELLIPSIS,+NORMALIZE_WHITESPACE
CREATE TABLE users (
user_id INTEGER NOT NULL,
{python}
>>> i = users_table.insert()
>>> i # doctest:+ELLIPSIS
- <sqlalchemy.sql._Insert object at 0x...>
+ <sqlalchemy.sql.Insert object at 0x...>
>>> # the string form of the Insert object is a generic SQL representation
>>> print i
INSERT INTO users (user_id, user_name, password) VALUES (?, ?, ?)
{python}
>>> row.keys()
- ['user_id', 'user_name', 'password']
+ [u'user_id', u'user_name', u'password']
>>> row['user_id'], row[1], row[users_table.c.password]
(4, u'Harry', None)
{python}
>>> query = session.query(User)
>>> print query.filter_by(user_name='Harry').all()
- SELECT users.user_name AS users_user_name, users.password AS users_password, users.user_id AS users_user_id
+ SELECT users.user_id AS users_user_id, users.user_name AS users_user_name, users.password AS users_password
FROM users
WHERE users.user_name = ? ORDER BY users.oid
['Harry']
Lets turn off the database echoing for a moment, and try out a few methods on `Query`. The two methods used to narrow results are `filter()` and `filter_by()`, and the two most common methods used to load results are `all()` and `first()`. The `get()` method is used for a quick lookup by primary key. `filter_by()` works with keyword arguments, and `filter()` works with `ClauseElement` objects, which are constructed by using `Column` objects inside of Python expressions, in the same way as we did with our SQL select example in the previous section of this tutorial. Using `ClauseElement` structures to query objects is more verbose but more flexible:
{python}
- >>> metadata.engine.echo = False
+ >>> metadata.bind.echo = False
>>> print query.filter(User.c.user_id==3).all()
[User(u'Fred',None)]
>>> print query.get(2)
Then to send all of our changes to the database, we `flush()` the Session. Lets turn echo back on to see this happen!:
{python}
- >>> metadata.engine.echo = True
+ >>> metadata.bind.echo = True
>>> session.flush()
BEGIN
UPDATE users SET password=? WHERE users.user_id = ?
... })
<sqlalchemy.orm.mapper.Mapper object at 0x...>
+Since we've made new mappers, we have to throw away the old `Query` object and get a new one:
+
+ >>> query = session.query(User)
+
The `relation()` function takes either a class or a Mapper as its first argument, and has many options to further control its behavior. When this mapping relationship is used, each new `User` instance will contain an attribute called `addresses`. SQLAlchemy will automatically determine that this relationship is a one-to-many relationship, and will subsequently create `addresses` as a list. When a new `User` is created, this list will begin as empty.
The order in which the mapping definitions for `User` and `Address` is created is *not significant*. When the `mapper()` function is called, it creates an *uncompiled* mapping record corresponding to the given class/table combination. When the mappers are first used, the entire collection of mappers created up until that point will be compiled, which involves the establishment of class instrumentation as well as the resolution of all mapping relationships.
{python}
>>> mary = query.filter_by(user_name='Mary').first() # doctest: +NORMALIZE_WHITESPACE
- SELECT users.user_name AS users_user_name, users.password AS users_password, users.user_id AS users_user_id
+ SELECT users.user_id AS users_user_id, users.user_name AS users_user_name, users.password AS users_password
FROM users
WHERE users.user_name = ? ORDER BY users.oid
LIMIT 1 OFFSET 0
['Mary']
>>> print [a for a in mary.addresses]
- SELECT email_addresses.user_id AS email_addresses_user_id, email_addresses.address_id AS email_addresses_address_id, email_addresses.email_address AS email_addresses_email_address
+ SELECT email_addresses.address_id AS email_addresses_address_id, email_addresses.email_address AS email_addresses_email_address, email_addresses.user_id AS email_addresses_user_id
FROM email_addresses
WHERE ? = email_addresses.user_id ORDER BY email_addresses.oid
[1]
### Transactions
-You may have noticed from the example above that when we say `session.flush()`, SQLAlchemy indicates the names `BEGIN` and `COMMIT` to indicate a transaction with the database. The `flush()` method, since it may execute many statements in a row, will automatically use a transaction in order to execute these instructions. But what if we want to use `flush()` inside of a larger transaction? This is performed via the `SessionTransaction` object, which we can establish using `session.create_transaction()`. Below, we will perform a more complicated `SELECT` statement, make several changes to our collection of users and email addresess, and then create a new user with two email addresses, within the context of a transaction. We will perform a `flush()` in the middle of it to write the changes we have so far, and then allow the remaining changes to be written when we finally `commit()` the transaction. We enclose our operations within a `try/except` block to ensure that resources are properly freed:
+You may have noticed from the example above that when we say `session.flush()`, SQLAlchemy indicates the names `BEGIN` and `COMMIT` to indicate a transaction with the database. The `flush()` method, since it may execute many statements in a row, will automatically use a transaction in order to execute these instructions. But what if we want to use `flush()` inside of a larger transaction? The easiest way is to use a "transactional" session; that is, when the session is created, you're automatically in a transaction which you can commit or rollback at any time. As a bonus, it offers the ability to call `flush()` for you, whenever a query is issued; that way whatever changes you've made can be returned right back (and since its all in a transaction, nothing gets committed until you tell it so).
+
+Below, we create a session with `autoflush=True`, which implies that it's transactional. We can query for things as soon as they are created without the need for calling `flush()`. At the end, we call `commit()` to persist everything permanently.
{python}
- >>> transaction = session.create_transaction()
- >>> try: # doctest: +NORMALIZE_WHITESPACE
- ... (ed, harry, mary) = session.query(User).filter(
+ >>> metadata.bind.echo = False
+ >>> session = create_session(autoflush=True)
+ >>> (ed, harry, mary) = session.query(User).filter(
... User.c.user_name.in_('Ed', 'Harry', 'Mary')
- ... ).order_by(User.c.user_name).all()
- ... del mary.addresses[1]
- ... harry.addresses.append(Address('harry2@gmail.com'))
- ... session.flush()
- ... print "***flushed the session***"
- ... fred = User()
- ... fred.user_name = 'fred_again'
- ... fred.addresses.append(Address('fred@fred.com'))
- ... fred.addresses.append(Address('fredsnewemail@fred.com'))
- ... session.save(fred)
- ... transaction.commit()
- ... except:
- ... transaction.rollback()
- ... raise
- BEGIN
- SELECT users.user_name AS users_user_name, users.password AS users_password, users.user_id AS users_user_id
- FROM users
- WHERE users.user_name IN (?, ?, ?) ORDER BY users.user_name
- ['Ed', 'Harry', 'Mary']
- SELECT email_addresses.user_id AS email_addresses_user_id, email_addresses.address_id AS email_addresses_address_id, email_addresses.email_address AS email_addresses_email_address
- FROM email_addresses
- WHERE ? = email_addresses.user_id ORDER BY email_addresses.oid
- [4]
- UPDATE email_addresses SET user_id=? WHERE email_addresses.address_id = ?
- [None, 3]
- INSERT INTO email_addresses (email_address, user_id) VALUES (?, ?)
- ['harry2@gmail.com', 4]
- ***flushed the session***
- INSERT INTO users (user_name, password) VALUES (?, ?)
- ['fred_again', None]
- INSERT INTO email_addresses (email_address, user_id) VALUES (?, ?)
- ['fred@fred.com', 6]
- INSERT INTO email_addresses (email_address, user_id) VALUES (?, ?)
- ['fredsnewemail@fred.com', 6]
- COMMIT
-
-The `SessionTransaction` process above is due to be greatly simplified in version 0.4 of SQLAlchemy, where the `Session` will be able to wrap its whole lifespan in a transaction automatically.
+ ... ).order_by(User.c.user_name).all() # doctest: +NORMALIZE_WHITESPACE
+ >>> del mary.addresses[1]
+ >>> harry_address = Address('harry2@gmail.com')
+ >>> harry.addresses.append(harry_address)
+ >>> session.query(User).join('addresses').filter_by(email_address='harry2@gmail.com').first() # doctest: +NORMALIZE_WHITESPACE
+ User(u'Harry',u'harrysnewpassword')
+ >>> session.commit()
Main documentation: [unitofwork](rel:unitofwork)
def persistent_id(self, obj):
if getattr(obj, "id", None) is None:
sess = MyPickler.sessions.current
- newsess = create_session(bind_to=sess.connection(class_mapper(Bar)))
+ newsess = create_session(bind=sess.connection(class_mapper(Bar)))
newsess.save(obj)
newsess.flush()
key = "%s:%s" % (type(obj).__name__, obj.id)
def _autorollback(self):
if not self.in_transaction():
self._rollback_impl()
-
+
def close(self):
try:
c = self.__connection
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
+"""Provide default implementations of per-dialect sqlalchemy.engine classes"""
from sqlalchemy import schema, exceptions, util, sql, types
import StringIO, sys, re
from sqlalchemy.engine import base
-"""Provide default implementations of the engine interfaces"""
-
class DefaultDialect(base.Dialect):
"""Default implementation of Dialect"""
one which is the "plain" strategy.
New strategies can be added via constructing a new EngineStrategy
-object which will add itself to the list of available strategies here,
-or replace one of the existing name. this can be accomplished via a
-mod; see the sqlalchemy/mods package for details.
+object which will add itself to the list of available strategies.
"""
import StringIO, sys, re
from sqlalchemy.engine import base, default
-"""Provide a thread-local transactional wrapper around the basic ComposedSQLEngine.
+"""Provide a thread-local transactional wrapper around the root Engine class.
Multiple calls to engine.connect() will return the same connection for
the same thread. also provides begin/commit methods on the engine
-import re
-import cgi
-import sys
-import urllib
+"""Provide the URL object as well as the make_url parsing function."""
+
+import re, cgi, sys, urllib
from sqlalchemy import exceptions
-"""Provide the URL object as well as the make_url parsing function."""
class URL(object):
"""Represent the components of a URL used to connect to a database.
+++ /dev/null
-"""A plugin that emulates 0.1 Session behavior."""
-
-import sqlalchemy.orm.objectstore as objectstore
-import sqlalchemy.orm.unitofwork as unitofwork
-import sqlalchemy.util as util
-import sqlalchemy
-
-import sqlalchemy.mods.threadlocal
-
-class LegacySession(objectstore.Session):
- def __init__(self, nest_on=None, hash_key=None, **kwargs):
- super(LegacySession, self).__init__(**kwargs)
- self.parent_uow = None
- self.begin_count = 0
- self.nest_on = util.to_list(nest_on)
- self.__pushed_count = 0
-
- def was_pushed(self):
- if self.nest_on is None:
- return
- self.__pushed_count += 1
- if self.__pushed_count == 1:
- for n in self.nest_on:
- n.push_session()
-
- def was_popped(self):
- if self.nest_on is None or self.__pushed_count == 0:
- return
- self.__pushed_count -= 1
- if self.__pushed_count == 0:
- for n in self.nest_on:
- n.pop_session()
-
- class SessionTrans(object):
- """Returned by ``Session.begin()``, denotes a
- transactionalized UnitOfWork instance. Call ``commit()`
- on this to commit the transaction.
- """
-
- def __init__(self, parent, uow, isactive):
- self.__parent = parent
- self.__isactive = isactive
- self.__uow = uow
-
- isactive = property(lambda s:s.__isactive, doc="True if this SessionTrans is the 'active' transaction marker, else its a no-op.")
- parent = property(lambda s:s.__parent, doc="The parent Session of this SessionTrans object.")
- uow = property(lambda s:s.__uow, doc="The parent UnitOfWork corresponding to this transaction.")
-
- def begin(self):
- """Call ``begin()`` on the underlying ``Session`` object,
- returning a new no-op ``SessionTrans`` object.
- """
-
- if self.parent.uow is not self.uow:
- raise InvalidRequestError("This SessionTrans is no longer valid")
- return self.parent.begin()
-
- def commit(self):
- """Commit the transaction noted by this ``SessionTrans`` object."""
-
- self.__parent._trans_commit(self)
- self.__isactive = False
-
- def rollback(self):
- """Roll back the current UnitOfWork transaction, in the
- case that ``begin()`` has been called.
-
- The changes logged since the begin() call are discarded.
- """
-
- self.__parent._trans_rollback(self)
- self.__isactive = False
-
- def begin(self):
- """Begin a new UnitOfWork transaction and return a
- transaction-holding object.
-
- ``commit()`` or ``rollback()`` should be called on the returned object.
-
- ``commit()`` on the ``Session`` will do nothing while a
- transaction is pending, and further calls to ``begin()`` will
- return no-op transactional objects.
- """
-
- if self.parent_uow is not None:
- return LegacySession.SessionTrans(self, self.uow, False)
- self.parent_uow = self.uow
- self.uow = unitofwork.UnitOfWork(identity_map = self.uow.identity_map)
- return LegacySession.SessionTrans(self, self.uow, True)
-
- def commit(self, *objects):
- """Commit the current UnitOfWork transaction.
-
- Called with no arguments, this is only used for *implicit*
- transactions when there was no ``begin()``.
-
- If individual objects are submitted, then only those objects
- are committed, and the begin/commit cycle is not affected.
- """
-
- # if an object list is given, commit just those but dont
- # change begin/commit status
- if len(objects):
- self._commit_uow(*objects)
- self.uow.flush(self, *objects)
- return
- if self.parent_uow is None:
- self._commit_uow()
-
- def _trans_commit(self, trans):
- if trans.uow is self.uow and trans.isactive:
- try:
- self._commit_uow()
- finally:
- self.uow = self.parent_uow
- self.parent_uow = None
-
- def _trans_rollback(self, trans):
- if trans.uow is self.uow:
- self.uow = self.parent_uow
- self.parent_uow = None
-
- def _commit_uow(self, *obj):
- self.was_pushed()
- try:
- self.uow.flush(self, *obj)
- finally:
- self.was_popped()
-
-def begin():
- """Deprecated. Use ``s = Session(new_imap=False)``."""
-
- return objectstore.get_session().begin()
-
-def commit(*obj):
- """Deprecated. Use ``flush(*obj)``."""
-
- objectstore.get_session().flush(*obj)
-
-def uow():
- return objectstore.get_session()
-
-def push_session(sess):
- old = get_session()
- if getattr(sess, '_previous', None) is not None:
- raise InvalidRequestError("Given Session is already pushed onto some thread's stack")
- sess._previous = old
- session_registry.set(sess)
- sess.was_pushed()
-
-def pop_session():
- sess = get_session()
- old = sess._previous
- sess._previous = None
- session_registry.set(old)
- sess.was_popped()
- return old
-
-def using_session(sess, func):
- push_session(sess)
- try:
- return func()
- finally:
- pop_session()
-
-def install_plugin():
- objectstore.Session = LegacySession
- objectstore.session_registry = util.ScopedRegistry(objectstore.Session)
- objectstore.begin = begin
- objectstore.commit = commit
- objectstore.uow = uow
- objectstore.push_session = push_session
- objectstore.pop_session = pop_session
- objectstore.using_session = using_session
-
-install_plugin()
+++ /dev/null
-"""This plugin installs thread-local behavior at the ``Engine`` and ``Session`` level.
-
-The default ``Engine`` strategy will be *threadlocal*, producing
-``TLocalEngine`` instances for create_engine by default.
-
-With this engine, ``connect()`` method will return the same connection
-on the same thread, if it is already checked out from the pool. This
-greatly helps functions that call multiple statements to be able to
-easily use just one connection without explicit ``close`` statements
-on result handles.
-
-On the ``Session`` side, module-level methods will be installed within
-the objectstore module, such as ``flush()``, ``delete()``, etc. which
-call this method on the thread-local session.
-
-Note: this mod creates a global, thread-local session context named
-``sqlalchemy.objectstore``. All mappers created while this mod is
-installed will reference this global context when creating new mapped
-object instances.
-"""
-
-from sqlalchemy import util, engine
-from sqlalchemy.ext.sessioncontext import SessionContext
-import sqlalchemy.ext.assignmapper as assignmapper
-from sqlalchemy.orm.mapper import global_extensions
-from sqlalchemy.orm.session import Session
-import sqlalchemy
-import sys, types
-
-__all__ = ['Objectstore', 'assign_mapper']
-
-class Objectstore(object):
- def __init__(self, *args, **kwargs):
- self.context = SessionContext(*args, **kwargs)
- def __getattr__(self, name):
- return getattr(self.context.current, name)
- session = property(lambda s:s.context.current)
-
-def assign_mapper(class_, *args, **kwargs):
- assignmapper.assign_mapper(objectstore.context, class_, *args, **kwargs)
-
-objectstore = Objectstore(Session)
-def install_plugin():
- sqlalchemy.objectstore = objectstore
- global_extensions.append(objectstore.context.mapper_extension)
- engine.default_strategy = 'threadlocal'
- sqlalchemy.assign_mapper = assign_mapper
-
-def uninstall_plugin():
- engine.default_strategy = 'plain'
- global_extensions.remove(objectstore.context.mapper_extension)
-
-install_plugin()
detect changes on the parent instance
comparator
- a sql.Comparator to which compare/math events will be sent
+ a sql.Comparator to which class-level compare/math events will be sent
"""
"""Bridge the ``PropertyLoader`` (i.e. a ``relation()``) and the
-``UOWTransaction`` together to allow processing of scalar- and
-list-based dependencies at flush time.
+``UOWTransaction`` together to allow processing of relation()-based
+ dependencies at flush time.
"""
from sqlalchemy.orm import sync
def populate_instance(self, selectcontext, instance, row, ispostselect=None, **flags):
"""populate an instance from a result row."""
+
populators = selectcontext.attributes.get(('instance_populators', self, ispostselect), None)
if populators is None:
populators = []
def __iter__(self):
statement = self.compile()
statement.use_labels = True
+ if self.session.autoflush:
+ self.session.flush()
result = self.session.execute(self.mapper, statement, params=self._params)
try:
return iter(self.instances(result))
The SessionTransaction object is **not** threadsafe.
"""
- def __init__(self, session, parent=None, autoflush=True):
+ def __init__(self, session, parent=None, autoflush=True, nested=False):
self.session = session
- self.connections = {}
- self.parent = parent
+ self.__connections = {}
+ self.__parent = parent
self.autoflush = autoflush
+ self.nested = nested
def connection(self, mapper_or_class, entity_name=None):
if isinstance(mapper_or_class, type):
engine = self.session.get_bind(mapper_or_class)
return self.get_or_add(engine)
- def _begin(self):
- return SessionTransaction(self.session, self)
+ def _begin(self, **kwargs):
+ return SessionTransaction(self.session, self, **kwargs)
def add(self, bind):
- if self.parent is not None:
- return self.parent.add(bind)
+ if self.__parent is not None:
+ return self.__parent.add(bind)
- if self.connections.has_key(bind.engine):
- raise exceptions.InvalidRequestError("Session already has a Connection associated for the given Connection's Engine")
+ if self.__connections.has_key(bind.engine):
+ raise exceptions.InvalidRequestError("Session already has a Connection associated for the given %sEngine" % (isinstance(bind, engine.Connection) and "Connection's " or ""))
return self.get_or_add(bind)
+ def _connection_dict(self):
+ if self.__parent is not None and not self.nested:
+ return self.__parent._connection_dict()
+ else:
+ return self.__connections
+
def get_or_add(self, bind):
- if self.parent is not None:
- return self.parent.get_or_add(bind)
+ if self.__parent is not None:
+ if not self.nested:
+ return self.__parent.get_or_add(bind)
+
+ if self.__connections.has_key(bind):
+ return self.__connections[bind][0]
+
+ if bind in self.__parent._connection_dict():
+ (conn, trans, autoclose) = self.__parent.__connections[bind]
+ self.__connections[conn] = self.__connections[bind.engine] = (conn, conn.begin_nested(), autoclose)
+ return conn
+ elif self.__connections.has_key(bind):
+ return self.__connections[bind][0]
- if self.connections.has_key(bind):
- return self.connections[bind][0]
-
if not isinstance(bind, engine.Connection):
e = bind
c = bind.contextual_connect()
else:
e = bind.engine
c = bind
-
- self.connections[bind] = self.connections[e] = (c, c.begin(), c is not bind)
- return self.connections[bind][0]
+ if e in self.__connections:
+ raise exceptions.InvalidRequestError("Session already has a Connection associated for the given Connection's Engine")
+ if self.nested:
+ trans = c.begin_nested()
+ else:
+ trans = c.begin()
+ self.__connections[c] = self.__connections[e] = (c, trans, c is not bind)
+ return self.__connections[c][0]
def commit(self):
- if self.parent is not None:
- return
+ if self.__parent is not None and not self.nested:
+ return self.__parent
if self.autoflush:
self.session.flush()
- for t in util.Set(self.connections.values()):
+ for t in util.Set(self.__connections.values()):
t[1].commit()
self.close()
+ return self.__parent
def rollback(self):
- if self.parent is not None:
- self.parent.rollback()
- return
- for k, t in self.connections.iteritems():
+ if self.__parent is not None and not self.nested:
+ return self.__parent.rollback()
+ for t in util.Set(self.__connections.values()):
t[1].rollback()
self.close()
-
+ return self.__parent
+
def close(self):
- if self.parent is not None:
+ if self.__parent is not None:
return
- for t in self.connections.values():
+ for t in util.Set(self.__connections.values()):
if t[2]:
t[0].close()
self.session.transaction = None
of Sessions, see the ``sqlalchemy.ext.sessioncontext`` module.
"""
- def __init__(self, bind=None, hash_key=None, import_session=None, echo_uow=False, weak_identity_map=False):
- if import_session is not None:
- self.uow = unitofwork.UnitOfWork(identity_map=import_session.uow.identity_map, weak_identity_map=weak_identity_map)
- else:
- self.uow = unitofwork.UnitOfWork(weak_identity_map=weak_identity_map)
+ def __init__(self, bind=None, autoflush=False, transactional=False, echo_uow=False, weak_identity_map=False):
+ self.uow = unitofwork.UnitOfWork(weak_identity_map=weak_identity_map)
self.bind = bind
self.binds = {}
self.echo_uow = echo_uow
self.weak_identity_map = weak_identity_map
self.transaction = None
- if hash_key is None:
- self.hash_key = id(self)
- else:
- self.hash_key = hash_key
+ self.hash_key = id(self)
+ self.autoflush = autoflush
+ self.transactional = transactional or autoflush
+ if self.transactional:
+ self.begin()
_sessions[self.hash_key] = self
-
+
def _get_echo_uow(self):
return self.uow.echo
self.uow.echo = value
echo_uow = property(_get_echo_uow,_set_echo_uow)
- def create_transaction(self, **kwargs):
- """Return a new ``SessionTransaction`` corresponding to an
- existing or new transaction.
-
- If the transaction is new, the returned ``SessionTransaction``
- will have commit control over the underlying transaction, else
- will have rollback control only.
- """
+ def begin(self, **kwargs):
+ """Begin a transaction on this Session."""
if self.transaction is not None:
- return self.transaction._begin()
+ self.transaction = self.transaction._begin(**kwargs)
else:
self.transaction = SessionTransaction(self, **kwargs)
- return self.transaction
+ create_transaction = begin
+ def begin_nested(self):
+ return self.begin(nested=True)
+
+ def rollback(self):
+ if self.transaction is None:
+ raise exceptions.InvalidRequestError("No transaction is begun.")
+ else:
+ self.transaction = self.transaction.rollback()
+ if self.transaction is None and self.transactional:
+ self.begin()
+
+ def commit(self):
+ if self.transaction is None:
+ raise exceptions.InvalidRequestError("No transaction is begun.")
+ else:
+ self.transaction = self.transaction.commit()
+ if self.transaction is None and self.transactional:
+ self.begin()
+
def connect(self, mapper=None, **kwargs):
"""Return a unique connection corresponding to the given mapper.
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
-
-
-from sqlalchemy import sql, schema, exceptions
-from sqlalchemy import logging
-from sqlalchemy.orm import util as mapperutil
-
"""Contains the ClauseSynchronizer class, which is used to map
attributes between two objects in a manner corresponding to a SQL
clause that compares column values.
"""
+from sqlalchemy import sql, schema, exceptions
+from sqlalchemy import logging
+from sqlalchemy.orm import util as mapperutil
+
ONETOMANY = 0
MANYTOONE = 1
MANYTOMANY = 2
for obj in self.deleted.intersection(objset).difference(processed):
flush_context.register_object(obj, isdelete=True)
- trans = session.create_transaction(autoflush=False)
- flush_context.transaction = trans
+ session.create_transaction(autoflush=False)
+ flush_context.transaction = session.transaction
try:
flush_context.execute()
except:
- trans.rollback()
+ session.rollback()
raise
- trans.commit()
+ session.commit()
flush_context.post_exec()
import tables
db = testbase.db
from sqlalchemy import *
+from testbase import Table, Column
class BindTest(testbase.PersistTest):
def test_create_drop_explicit(self):
def test_implicit_execution(self):
metadata = MetaData()
table = Table('test_table', metadata,
- Column('foo', Integer))
+ Column('foo', Integer),
+ test_needs_acid=True,
+ )
conn = testbase.db.connect()
metadata.create_all(bind=conn)
try:
table.insert().execute(foo=7)
trans.rollback()
metadata.bind = None
- assert testbase.db.execute("select count(1) from test_table").scalar() == 0
+ assert conn.execute("select count(1) from test_table").scalar() == 0
finally:
metadata.drop_all(bind=conn)
l = q.add_column("count").add_column("concat").from_statement(s).all()
assert l == expected
+ assert [user8] == create_session().query(User).select_from(s).filter(s.c.count==3).all()
+
q = create_session().query(User).add_column(func.count(addresses.c.id))\
.add_column(("Name:" + users.c.name)).select_from(users.outerjoin(addresses))\
.group_by([c for c in users.c]).order_by(users.c.id)
s.user_name = 'some other user'
s.flush()
+ def test_close_two(self):
+ c = testbase.db.connect()
+ try:
+ class User(object):pass
+ mapper(User, users)
+ s = create_session(bind=c)
+ s.begin()
+ tran = s.transaction
+ s.save(User())
+ s.flush()
+ c.execute("select * from users")
+ u = User()
+ s.save(u)
+ s.user_name = 'some user'
+ s.flush()
+ u = User()
+ s.save(u)
+ s.user_name = 'some other user'
+ s.flush()
+ assert s.transaction is tran
+ tran.close()
+ finally:
+ c.close()
+
def test_expunge_cascade(self):
tables.data()
mapper(Address, addresses)
# then see if expunge fails
session.expunge(u)
-
+
+ @testbase.unsupported('sqlite')
def test_transaction(self):
class User(object):pass
mapper(User, users)
- sess = create_session()
- transaction = sess.create_transaction()
+ conn1 = testbase.db.connect()
+ conn2 = testbase.db.connect()
+
+ sess = create_session(transactional=True, bind=conn1)
+ u = User()
+ sess.save(u)
+ sess.flush()
+ assert conn1.execute("select count(1) from users").scalar() == 1
+ assert conn2.execute("select count(1) from users").scalar() == 0
+ sess.commit()
+ assert conn1.execute("select count(1) from users").scalar() == 1
+ assert testbase.db.connect().execute("select count(1) from users").scalar() == 1
+
+ def test_autoflush(self):
+ class User(object):pass
+ mapper(User, users)
+ conn1 = testbase.db.connect()
+ conn2 = testbase.db.connect()
+
+ sess = create_session(autoflush=True, bind=conn1)
+ u = User()
+ u.user_name='ed'
+ sess.save(u)
+ u2 = sess.query(User).filter_by(user_name='ed').one()
+ assert u2 is u
+ assert conn1.execute("select count(1) from users").scalar() == 1
+ assert conn2.execute("select count(1) from users").scalar() == 0
+ sess.commit()
+ assert conn1.execute("select count(1) from users").scalar() == 1
+ assert testbase.db.connect().execute("select count(1) from users").scalar() == 1
+
+ def test_autoflush_2(self):
+ class User(object):pass
+ mapper(User, users)
+ conn1 = testbase.db.connect()
+ conn2 = testbase.db.connect()
+
+ sess = create_session(autoflush=True, bind=conn1)
+ u = User()
+ u.user_name='ed'
+ sess.save(u)
+ sess.commit()
+ assert conn1.execute("select count(1) from users").scalar() == 1
+ assert testbase.db.connect().execute("select count(1) from users").scalar() == 1
+
+ def test_external_joined_transaction(self):
+ class User(object):pass
+ mapper(User, users)
+ conn = testbase.db.connect()
+ trans = conn.begin()
+ sess = create_session(bind=conn)
+ sess.begin()
+ u = User()
+ sess.save(u)
+ sess.flush()
+ sess.commit() # commit does nothing
+ trans.rollback() # rolls back
+ assert len(sess.query(User).select()) == 0
+
+ @testbase.supported('postgres', 'mysql')
+ def test_external_nested_transaction(self):
+ class User(object):pass
+ mapper(User, users)
try:
- u = User()
- sess.save(u)
+ conn = testbase.db.connect()
+ trans = conn.begin()
+ sess = create_session(bind=conn)
+ u1 = User()
+ sess.save(u1)
sess.flush()
- sess.delete(u)
- sess.save(User())
+
+ sess.begin_nested()
+ u2 = User()
+ sess.save(u2)
sess.flush()
- # TODO: assertion ?
- transaction.commit()
+ sess.rollback()
+
+ trans.commit()
+ assert len(sess.query(User).select()) == 1
except:
- transaction.rollback()
+ conn.close()
+ raise
+
+ def test_joined_transaction(self):
+ class User(object):pass
+ mapper(User, users)
+ sess = create_session()
+ sess.begin()
+ sess.begin()
+ u = User()
+ sess.save(u)
+ sess.flush()
+ sess.commit() # commit does nothing
+ sess.rollback() # rolls back
+ assert len(sess.query(User).select()) == 0
+ @testbase.supported('postgres', 'mysql')
def test_nested_transaction(self):
class User(object):pass
mapper(User, users)
sess = create_session()
- transaction = sess.create_transaction()
- trans2 = sess.create_transaction()
+ sess.begin()
+
u = User()
sess.save(u)
sess.flush()
- trans2.commit()
- transaction.rollback()
- assert len(sess.query(User).select()) == 0
+
+ sess.begin_nested() # nested transaction
+
+ u2 = User()
+ sess.save(u2)
+ sess.flush()
+
+ sess.rollback()
+
+ sess.commit()
+ assert len(sess.query(User).select()) == 1
+
+ @testbase.supported('postgres', 'mysql')
+ def test_nested_autotrans(self):
+ class User(object):pass
+ mapper(User, users)
+ sess = create_session(transactional=True)
+ u = User()
+ sess.save(u)
+ sess.flush()
+
+ sess.begin_nested() # nested transaction
+
+ u2 = User()
+ sess.save(u2)
+ sess.flush()
+
+ sess.rollback()
+
+ sess.commit()
+ assert len(sess.query(User).select()) == 1
def test_bound_connection(self):
class User(object):pass
mapper(User, users)
c = testbase.db.connect()
sess = create_session(bind=c)
- transaction = sess.create_transaction()
- trans2 = sess.create_transaction()
+ sess.create_transaction()
+ transaction = sess.transaction
u = User()
sess.save(u)
sess.flush()
- assert transaction.get_or_add(testbase.db) is trans2.get_or_add(testbase.db) #is transaction.get_or_add(c) is trans2.get_or_add(c) is c
- trans2.commit()
- transaction.rollback()
- assert len(sess.query(User).select()) == 0
+ assert transaction.get_or_add(testbase.db) is transaction.get_or_add(c) is c
- def test_close_two(self):
- c = testbase.db.connect()
try:
- class User(object):pass
- mapper(User, users)
- s = create_session(bind=c)
- tran = s.create_transaction()
- s.save(User())
- s.flush()
- c.execute("select * from users")
- u = User()
- s.save(u)
- s.user_name = 'some user'
- s.flush()
- u = User()
- s.save(u)
- s.user_name = 'some other user'
- s.flush()
- assert s.transaction is tran
- tran.close()
- finally:
- c.close()
+ transaction.add(testbase.db.connect())
+ assert False
+ except exceptions.InvalidRequestError, e:
+ assert str(e) == "Session already has a Connection associated for the given Connection's Engine"
+
+ try:
+ transaction.get_or_add(testbase.db.connect())
+ assert False
+ except exceptions.InvalidRequestError, e:
+ assert str(e) == "Session already has a Connection associated for the given Connection's Engine"
+
+ try:
+ transaction.add(testbase.db)
+ assert False
+ except exceptions.InvalidRequestError, e:
+ assert str(e) == "Session already has a Connection associated for the given Engine"
+ transaction.rollback()
+ assert len(sess.query(User).select()) == 0
+
def test_update(self):
"""test that the update() method functions and doesnet blow away changes"""
tables.delete()
user = s.query(User).selectone()
assert user.user_name == 'fred'
- # insure its not dirty if no changes occur
+ # ensure its not dirty if no changes occur
s.clear()
assert user not in s
s.update(user)
mytable = Table('mytable', metadata,
Column('id', Integer, primary_key=True),
Column('data', String(30)),
- test_need_fk=True,
+ test_needs_fk=True,
)
myothertable = Table('myothertable', metadata,
Column('parent_id', Integer),
Column('data', String(30)),
ForeignKeyConstraint(['parent_id'],['mytable.id'], ondelete="CASCADE"),
- test_need_fk=True,
+ test_needs_fk=True,
)
metadata.create_all()
class ZBlogTest(AssertMixin):
def create_tables(self):
- tables.metadata.create_all(connectable=db)
+ tables.metadata.create_all(bind=db)
def drop_tables(self):
- tables.metadata.drop_all(connectable=db)
+ tables.metadata.drop_all(bind=db)
def setUpAll(self):
self.create_tables()
super(SavePostTest, self).setUpAll()
mappers.zblog_mappers()
global blog_id, user_id
- s = create_session(bind_to=db)
+ s = create_session(bind=db)
user = User('zbloguser', "Zblog User", "hello", group=administrator)
blog = Blog(owner=user)
blog.name = "this is a blog"
"""test that a transient/pending instance has proper bi-directional behavior.
this requires that lazy loaders do not fire off for a transient/pending instance."""
- s = create_session(bind_to=db)
+ s = create_session(bind=db)
- trans = s.create_transaction()
+ s.begin()
try:
blog = s.query(Blog).get(blog_id)
post = Post(headline="asdf asdf", summary="asdfasfd")
post.blog = blog
assert post in blog.posts
finally:
- trans.rollback()
+ s.rollback()
def testoptimisticorphans(self):
"""test that instances in the session with un-loaded parents will not
get marked as "orphans" and then deleted """
- s = create_session(bind_to=db)
+ s = create_session(bind=db)
- trans = s.create_transaction()
+ s.begin()
try:
blog = s.query(Blog).get(blog_id)
post = Post(headline="asdf asdf", summary="asdfasfd")
assert s.query(Post).get(post.id) is not None
finally:
- trans.rollback()
+ s.rollback()
if __name__ == "__main__":
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
