dinislage/CouchPotatoServer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update sqlalchemy

Browse Source
This commit is contained in:
Ruud
2012-05-05 01:13:50 +02:00
parent dcaca7294c
commit 710c2aa05f
38 changed files with 2292 additions and 1025 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/sqlalchemy/orm/collections.py
+49 -15
View File
@@ -112,12 +112,32 @@ from sqlalchemy.sql import expression
from sqlalchemy import schema, util, exc as sa_exc
__all__ = ['collection', 'collection_adapter',
'mapped_collection', 'column_mapped_collection',
'attribute_mapped_collection']
__instrumentation_mutex = util.threading.Lock()
class _SerializableColumnGetter(object):
def __init__(self, colkeys):
self.colkeys = colkeys
self.composite = len(colkeys) > 1
def __reduce__(self):
return _SerializableColumnGetter, (self.colkeys,)
def __call__(self, value):
state = instance_state(value)
m = _state_mapper(state)
key = [m._get_state_attr_by_column(
state, state.dict,
m.mapped_table.columns[k])
for k in self.colkeys]
if self.composite:
return tuple(key)
else:
return key[0]
def column_mapped_collection(mapping_spec):
"""A dictionary-based collection type with column-based keying.
@@ -131,25 +151,27 @@ def column_mapped_collection(mapping_spec):
after a session flush.
"""
global _state_mapper, instance_state
from sqlalchemy.orm.util import _state_mapper
from sqlalchemy.orm.attributes import instance_state
cols = [expression._only_column_elements(q, "mapping_spec")
for q in util.to_list(mapping_spec)]
if len(cols) == 1:
def keyfunc(value):
state = instance_state(value)
m = _state_mapper(state)
return m._get_state_attr_by_column(state, state.dict, cols[0])
else:
mapping_spec = tuple(cols)
def keyfunc(value):
state = instance_state(value)
m = _state_mapper(state)
return tuple(m._get_state_attr_by_column(state, state.dict, c)
for c in mapping_spec)
cols = [c.key for c in [
expression._only_column_elements(q, "mapping_spec")
for q in util.to_list(mapping_spec)]]
keyfunc = _SerializableColumnGetter(cols)
return lambda: MappedCollection(keyfunc)
class _SerializableAttrGetter(object):
def __init__(self, name):
self.name = name
self.getter = operator.attrgetter(name)
def __call__(self, target):
return self.getter(target)
def __reduce__(self):
return _SerializableAttrGetter, (self.name, )
def attribute_mapped_collection(attr_name):
"""A dictionary-based collection type with attribute-based keying.
@@ -163,7 +185,8 @@ def attribute_mapped_collection(attr_name):
after a session flush.
"""
return lambda: MappedCollection(operator.attrgetter(attr_name))
getter = _SerializableAttrGetter(attr_name)
return lambda: MappedCollection(getter)
def mapped_collection(keyfunc):
@@ -814,6 +837,7 @@ def _instrument_class(cls):
methods[name] = None, None, after
# apply ABC auto-decoration to methods that need it
for method, decorator in decorators.items():
fn = getattr(cls, method, None)
if (fn and method not in methods and
@@ -1465,3 +1489,13 @@ class MappedCollection(dict):
incoming_key, value, new_key))
yield value
_convert = collection.converter(_convert)
# ensure instrumentation is associated with
# these built-in classes; if a user-defined class
# subclasses these and uses @internally_instrumented,
# the superclass is otherwise not instrumented.
# see [ticket:2406].
_instrument_class(MappedCollection)
_instrument_class(InstrumentedList)
_instrument_class(InstrumentedSet)
libs/sqlalchemy/orm/mapper.py
+19 -630
View File
@@ -1452,12 +1452,19 @@ class Mapper(object):
return result
def _is_userland_descriptor(self, obj):
return not isinstance(obj,
(MapperProperty, attributes.QueryableAttribute)) and \
hasattr(obj, '__get__') and not \
isinstance(obj.__get__(None, obj),
attributes.QueryableAttribute)
if isinstance(obj, (MapperProperty,
attributes.QueryableAttribute)):
return False
elif not hasattr(obj, '__get__'):
return False
else:
obj = util.unbound_method_to_callable(obj)
if isinstance(
obj.__get__(None, obj),
attributes.QueryableAttribute
):
return False
return True
def _should_exclude(self, name, assigned_name, local, column):
"""determine whether a particular property should be implicitly
@@ -1875,501 +1882,6 @@ class Mapper(object):
self._memoized_values[key] = value = callable_()
return value
def _post_update(self, states, uowtransaction, post_update_cols):
"""Issue UPDATE statements on behalf of a relationship() which
specifies post_update.
"""
cached_connections = util.PopulateDict(
lambda conn:conn.execution_options(
compiled_cache=self._compiled_cache
))
# if session has a connection callable,
# organize individual states with the connection
# to use for update
if uowtransaction.session.connection_callable:
connection_callable = \
uowtransaction.session.connection_callable
else:
connection = uowtransaction.transaction.connection(self)
connection_callable = None
tups = []
for state in _sort_states(states):
if connection_callable:
conn = connection_callable(self, state.obj())
else:
conn = connection
mapper = _state_mapper(state)
tups.append((state, state.dict, mapper, conn))
table_to_mapper = self._sorted_tables
for table in table_to_mapper:
update = []
for state, state_dict, mapper, connection in tups:
if table not in mapper._pks_by_table:
continue
pks = mapper._pks_by_table[table]
params = {}
hasdata = False
for col in mapper._cols_by_table[table]:
if col in pks:
params[col._label] = \
mapper._get_state_attr_by_column(
state,
state_dict, col)
elif col in post_update_cols:
prop = mapper._columntoproperty[col]
history = attributes.get_state_history(
state, prop.key,
attributes.PASSIVE_NO_INITIALIZE)
if history.added:
value = history.added[0]
params[col.key] = value
hasdata = True
if hasdata:
update.append((state, state_dict, params, mapper,
connection))
if update:
mapper = table_to_mapper[table]
def update_stmt():
clause = sql.and_()
for col in mapper._pks_by_table[table]:
clause.clauses.append(col == sql.bindparam(col._label,
type_=col.type))
return table.update(clause)
statement = self._memo(('post_update', table), update_stmt)
# execute each UPDATE in the order according to the original
# list of states to guarantee row access order, but
# also group them into common (connection, cols) sets
# to support executemany().
for key, grouper in groupby(
update, lambda rec: (rec[4], rec[2].keys())
):
multiparams = [params for state, state_dict,
params, mapper, conn in grouper]
cached_connections[connection].\
execute(statement, multiparams)
def _save_obj(self, states, uowtransaction, single=False):
"""Issue ``INSERT`` and/or ``UPDATE`` statements for a list
of objects.
This is called within the context of a UOWTransaction during a
flush operation, given a list of states to be flushed. The
base mapper in an inheritance hierarchy handles the inserts/
updates for all descendant mappers.
"""
# if batch=false, call _save_obj separately for each object
if not single and not self.batch:
for state in _sort_states(states):
self._save_obj([state],
uowtransaction,
single=True)
return
# if session has a connection callable,
# organize individual states with the connection
# to use for insert/update
if uowtransaction.session.connection_callable:
connection_callable = \
uowtransaction.session.connection_callable
else:
connection = uowtransaction.transaction.connection(self)
connection_callable = None
tups = []
for state in _sort_states(states):
if connection_callable:
conn = connection_callable(self, state.obj())
else:
conn = connection
has_identity = bool(state.key)
mapper = _state_mapper(state)
instance_key = state.key or mapper._identity_key_from_state(state)
row_switch = None
# call before_XXX extensions
if not has_identity:
mapper.dispatch.before_insert(mapper, conn, state)
else:
mapper.dispatch.before_update(mapper, conn, state)
# detect if we have a "pending" instance (i.e. has
# no instance_key attached to it), and another instance
# with the same identity key already exists as persistent.
# convert to an UPDATE if so.
if not has_identity and \
instance_key in uowtransaction.session.identity_map:
instance = \
uowtransaction.session.identity_map[instance_key]
existing = attributes.instance_state(instance)
if not uowtransaction.is_deleted(existing):
raise orm_exc.FlushError(
"New instance %s with identity key %s conflicts "
"with persistent instance %s" %
(state_str(state), instance_key,
state_str(existing)))
self._log_debug(
"detected row switch for identity %s. "
"will update %s, remove %s from "
"transaction", instance_key,
state_str(state), state_str(existing))
# remove the "delete" flag from the existing element
uowtransaction.remove_state_actions(existing)
row_switch = existing
tups.append(
(state, state.dict, mapper, conn,
has_identity, instance_key, row_switch)
)
# dictionary of connection->connection_with_cache_options.
cached_connections = util.PopulateDict(
lambda conn:conn.execution_options(
compiled_cache=self._compiled_cache
))
table_to_mapper = self._sorted_tables
for table in table_to_mapper:
insert = []
update = []
for state, state_dict, mapper, connection, has_identity, \
instance_key, row_switch in tups:
if table not in mapper._pks_by_table:
continue
pks = mapper._pks_by_table[table]
isinsert = not has_identity and not row_switch
params = {}
value_params = {}
if isinsert:
has_all_pks = True
for col in mapper._cols_by_table[table]:
if col is mapper.version_id_col:
params[col.key] = \
mapper.version_id_generator(None)
else:
# pull straight from the dict for
# pending objects
prop = mapper._columntoproperty[col]
value = state_dict.get(prop.key, None)
if value is None:
if col in pks:
has_all_pks = False
elif col.default is None and \
col.server_default is None:
params[col.key] = value
elif isinstance(value, sql.ClauseElement):
value_params[col] = value
else:
params[col.key] = value
insert.append((state, state_dict, params, mapper,
connection, value_params, has_all_pks))
else:
hasdata = False
for col in mapper._cols_by_table[table]:
if col is mapper.version_id_col:
params[col._label] = \
mapper._get_committed_state_attr_by_column(
row_switch or state,
row_switch and row_switch.dict
or state_dict,
col)
prop = mapper._columntoproperty[col]
history = attributes.get_state_history(
state, prop.key,
attributes.PASSIVE_NO_INITIALIZE
)
if history.added:
params[col.key] = history.added[0]
hasdata = True
else:
params[col.key] = \
mapper.version_id_generator(
params[col._label])
# HACK: check for history, in case the
# history is only
# in a different table than the one
# where the version_id_col is.
for prop in mapper._columntoproperty.\
itervalues():
history = attributes.get_state_history(
state, prop.key,
attributes.PASSIVE_NO_INITIALIZE)
if history.added:
hasdata = True
else:
prop = mapper._columntoproperty[col]
history = attributes.get_state_history(
state, prop.key,
attributes.PASSIVE_NO_INITIALIZE)
if history.added:
if isinstance(history.added[0],
sql.ClauseElement):
value_params[col] = history.added[0]
else:
value = history.added[0]
params[col.key] = value
if col in pks:
if history.deleted and \
not row_switch:
# if passive_updates and sync detected
# this was a pk->pk sync, use the new
# value to locate the row, since the
# DB would already have set this
if ("pk_cascaded", state, col) in \
uowtransaction.\
attributes:
value = history.added[0]
params[col._label] = value
else:
# use the old value to
# locate the row
value = history.deleted[0]
params[col._label] = value
hasdata = True
else:
# row switch logic can reach us here
# remove the pk from the update params
# so the update doesn't
# attempt to include the pk in the
# update statement
del params[col.key]
value = history.added[0]
params[col._label] = value
if value is None and hasdata:
raise sa_exc.FlushError(
"Can't update table "
"using NULL for primary key "
"value")
else:
hasdata = True
elif col in pks:
value = state.manager[prop.key].\
impl.get(state, state_dict)
if value is None:
raise sa_exc.FlushError(
"Can't update table "
"using NULL for primary "
"key value")
params[col._label] = value
if hasdata:
update.append((state, state_dict, params, mapper,
connection, value_params))
if update:
mapper = table_to_mapper[table]
needs_version_id = mapper.version_id_col is not None and \
table.c.contains_column(mapper.version_id_col)
def update_stmt():
clause = sql.and_()
for col in mapper._pks_by_table[table]:
clause.clauses.append(col == sql.bindparam(col._label,
type_=col.type))
if needs_version_id:
clause.clauses.append(mapper.version_id_col ==\
sql.bindparam(mapper.version_id_col._label,
type_=col.type))
return table.update(clause)
statement = self._memo(('update', table), update_stmt)
rows = 0
for state, state_dict, params, mapper, \
connection, value_params in update:
if value_params:
c = connection.execute(
statement.values(value_params),
params)
else:
c = cached_connections[connection].\
execute(statement, params)
mapper._postfetch(
uowtransaction,
table,
state,
state_dict,
c.context.prefetch_cols,
c.context.postfetch_cols,
c.context.compiled_parameters[0],
value_params)
rows += c.rowcount
if connection.dialect.supports_sane_rowcount:
if rows != len(update):
raise orm_exc.StaleDataError(
"UPDATE statement on table '%s' expected to update %d row(s); "
"%d were matched." %
(table.description, len(update), rows))
elif needs_version_id:
util.warn("Dialect %s does not support updated rowcount "
"- versioning cannot be verified." %
c.dialect.dialect_description,
stacklevel=12)
if insert:
statement = self._memo(('insert', table), table.insert)
for (connection, pkeys, hasvalue, has_all_pks), \
records in groupby(insert,
lambda rec: (rec[4],
rec[2].keys(),
bool(rec[5]),
rec[6])
):
if has_all_pks and not hasvalue:
records = list(records)
multiparams = [rec[2] for rec in records]
c = cached_connections[connection].\
execute(statement, multiparams)
for (state, state_dict, params, mapper,
conn, value_params, has_all_pks), \
last_inserted_params in \
zip(records, c.context.compiled_parameters):
mapper._postfetch(
uowtransaction,
table,
state,
state_dict,
c.context.prefetch_cols,
c.context.postfetch_cols,
last_inserted_params,
value_params)
else:
for state, state_dict, params, mapper, \
connection, value_params, \
has_all_pks in records:
if value_params:
result = connection.execute(
statement.values(value_params),
params)
else:
result = cached_connections[connection].\
execute(statement, params)
primary_key = result.context.inserted_primary_key
if primary_key is not None:
# set primary key attributes
for pk, col in zip(primary_key,
mapper._pks_by_table[table]):
prop = mapper._columntoproperty[col]
if state_dict.get(prop.key) is None:
# TODO: would rather say:
#state_dict[prop.key] = pk
mapper._set_state_attr_by_column(
state,
state_dict,
col, pk)
mapper._postfetch(
uowtransaction,
table,
state,
state_dict,
result.context.prefetch_cols,
result.context.postfetch_cols,
result.context.compiled_parameters[0],
value_params)
for state, state_dict, mapper, connection, has_identity, \
instance_key, row_switch in tups:
if mapper._readonly_props:
readonly = state.unmodified_intersection(
[p.key for p in mapper._readonly_props
if p.expire_on_flush or p.key not in state.dict]
)
if readonly:
state.expire_attributes(state.dict, readonly)
# if eager_defaults option is enabled,
# refresh whatever has been expired.
if self.eager_defaults and state.unloaded:
state.key = self._identity_key_from_state(state)
uowtransaction.session.query(self)._load_on_ident(
state.key, refresh_state=state,
only_load_props=state.unloaded)
# call after_XXX extensions
if not has_identity:
mapper.dispatch.after_insert(mapper, connection, state)
else:
mapper.dispatch.after_update(mapper, connection, state)
def _postfetch(self, uowtransaction, table,
state, dict_, prefetch_cols, postfetch_cols,
params, value_params):
"""During a flush, expire attributes in need of newly
persisted database state."""
if self.version_id_col is not None:
prefetch_cols = list(prefetch_cols) + [self.version_id_col]
for c in prefetch_cols:
if c.key in params and c in self._columntoproperty:
self._set_state_attr_by_column(state, dict_, c, params[c.key])
if postfetch_cols:
state.expire_attributes(state.dict,
[self._columntoproperty[c].key
for c in postfetch_cols if c in
self._columntoproperty]
)
# synchronize newly inserted ids from one table to the next
# TODO: this still goes a little too often. would be nice to
# have definitive list of "columns that changed" here
for m, equated_pairs in self._table_to_equated[table]:
sync.populate(state, m, state, m,
equated_pairs,
uowtransaction,
self.passive_updates)
@util.memoized_property
def _table_to_equated(self):
"""memoized map of tables to collections of columns to be
@@ -2387,128 +1899,6 @@ class Mapper(object):
return result
def _delete_obj(self, states, uowtransaction):
"""Issue ``DELETE`` statements for a list of objects.
This is called within the context of a UOWTransaction during a
flush operation.
"""
if uowtransaction.session.connection_callable:
connection_callable = \
uowtransaction.session.connection_callable
else:
connection = uowtransaction.transaction.connection(self)
connection_callable = None
tups = []
cached_connections = util.PopulateDict(
lambda conn:conn.execution_options(
compiled_cache=self._compiled_cache
))
for state in _sort_states(states):
mapper = _state_mapper(state)
if connection_callable:
conn = connection_callable(self, state.obj())
else:
conn = connection
mapper.dispatch.before_delete(mapper, conn, state)
tups.append((state,
state.dict,
_state_mapper(state),
bool(state.key),
conn))
table_to_mapper = self._sorted_tables
for table in reversed(table_to_mapper.keys()):
delete = util.defaultdict(list)
for state, state_dict, mapper, has_identity, connection in tups:
if not has_identity or table not in mapper._pks_by_table:
continue
params = {}
delete[connection].append(params)
for col in mapper._pks_by_table[table]:
params[col.key] = \
value = \
mapper._get_state_attr_by_column(
state, state_dict, col)
if value is None:
raise sa_exc.FlushError(
"Can't delete from table "
"using NULL for primary "
"key value")
if mapper.version_id_col is not None and \
table.c.contains_column(mapper.version_id_col):
params[mapper.version_id_col.key] = \
mapper._get_committed_state_attr_by_column(
state, state_dict,
mapper.version_id_col)
mapper = table_to_mapper[table]
need_version_id = mapper.version_id_col is not None and \
table.c.contains_column(mapper.version_id_col)
def delete_stmt():
clause = sql.and_()
for col in mapper._pks_by_table[table]:
clause.clauses.append(
col == sql.bindparam(col.key, type_=col.type))
if need_version_id:
clause.clauses.append(
mapper.version_id_col ==
sql.bindparam(
mapper.version_id_col.key,
type_=mapper.version_id_col.type
)
)
return table.delete(clause)
for connection, del_objects in delete.iteritems():
statement = self._memo(('delete', table), delete_stmt)
rows = -1
connection = cached_connections[connection]
if need_version_id and \
not connection.dialect.supports_sane_multi_rowcount:
# TODO: need test coverage for this [ticket:1761]
if connection.dialect.supports_sane_rowcount:
rows = 0
# execute deletes individually so that versioned
# rows can be verified
for params in del_objects:
c = connection.execute(statement, params)
rows += c.rowcount
else:
util.warn(
"Dialect %s does not support deleted rowcount "
"- versioning cannot be verified." %
connection.dialect.dialect_description,
stacklevel=12)
connection.execute(statement, del_objects)
else:
c = connection.execute(statement, del_objects)
if connection.dialect.supports_sane_multi_rowcount:
rows = c.rowcount
if rows != -1 and rows != len(del_objects):
raise orm_exc.StaleDataError(
"DELETE statement on table '%s' expected to delete %d row(s); "
"%d were matched." %
(table.description, len(del_objects), c.rowcount)
)
for state, state_dict, mapper, has_identity, connection in tups:
mapper.dispatch.after_delete(mapper, connection, state)
def _instance_processor(self, context, path, reduced_path, adapter,
polymorphic_from=None,
@@ -2518,6 +1908,12 @@ class Mapper(object):
"""Produce a mapper level row processor callable
which processes rows into mapped instances."""
# note that this method, most of which exists in a closure
# called _instance(), resists being broken out, as
# attempts to do so tend to add significant function
# call overhead. _instance() is the most
# performance-critical section in the whole ORM.
pk_cols = self.primary_key
if polymorphic_from or refresh_state:
@@ -2961,13 +2357,6 @@ def _event_on_resurrect(state):
state, state.dict, col, val)
def _sort_states(states):
pending = set(states)
persistent = set(s for s in pending if s.key is not None)
pending.difference_update(persistent)
return sorted(pending, key=operator.attrgetter("insert_order")) + \
sorted(persistent, key=lambda q:q.key[1])
class _ColumnMapping(util.py25_dict):
"""Error reporting helper for mapper._columntoproperty."""
libs/sqlalchemy/orm/persistence.py
+777
View File
@@ -0,0 +1,777 @@
# orm/persistence.py
# Copyright (C) 2005-2012 the SQLAlchemy authors and contributors <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
"""private module containing functions used to emit INSERT, UPDATE
and DELETE statements on behalf of a :class:`.Mapper` and its descending
mappers.
The functions here are called only by the unit of work functions
in unitofwork.py.
"""
import operator
from itertools import groupby
from sqlalchemy import sql, util, exc as sa_exc
from sqlalchemy.orm import attributes, sync, \
exc as orm_exc
from sqlalchemy.orm.util import _state_mapper, state_str
def save_obj(base_mapper, states, uowtransaction, single=False):
"""Issue ``INSERT`` and/or ``UPDATE`` statements for a list
of objects.
This is called within the context of a UOWTransaction during a
flush operation, given a list of states to be flushed. The
base mapper in an inheritance hierarchy handles the inserts/
updates for all descendant mappers.
"""
# if batch=false, call _save_obj separately for each object
if not single and not base_mapper.batch:
for state in _sort_states(states):
save_obj(base_mapper, [state], uowtransaction, single=True)
return
states_to_insert, states_to_update = _organize_states_for_save(
base_mapper,
states,
uowtransaction)
cached_connections = _cached_connection_dict(base_mapper)
for table, mapper in base_mapper._sorted_tables.iteritems():
insert = _collect_insert_commands(base_mapper, uowtransaction,
table, states_to_insert)
update = _collect_update_commands(base_mapper, uowtransaction,
table, states_to_update)
if update:
_emit_update_statements(base_mapper, uowtransaction,
cached_connections,
mapper, table, update)
if insert:
_emit_insert_statements(base_mapper, uowtransaction,
cached_connections,
table, insert)
_finalize_insert_update_commands(base_mapper, uowtransaction,
states_to_insert, states_to_update)
def post_update(base_mapper, states, uowtransaction, post_update_cols):
"""Issue UPDATE statements on behalf of a relationship() which
specifies post_update.
"""
cached_connections = _cached_connection_dict(base_mapper)
states_to_update = _organize_states_for_post_update(
base_mapper,
states, uowtransaction)
for table, mapper in base_mapper._sorted_tables.iteritems():
update = _collect_post_update_commands(base_mapper, uowtransaction,
table, states_to_update,
post_update_cols)
if update:
_emit_post_update_statements(base_mapper, uowtransaction,
cached_connections,
mapper, table, update)
def delete_obj(base_mapper, states, uowtransaction):
"""Issue ``DELETE`` statements for a list of objects.
This is called within the context of a UOWTransaction during a
flush operation.
"""
cached_connections = _cached_connection_dict(base_mapper)
states_to_delete = _organize_states_for_delete(
base_mapper,
states,
uowtransaction)
table_to_mapper = base_mapper._sorted_tables
for table in reversed(table_to_mapper.keys()):
delete = _collect_delete_commands(base_mapper, uowtransaction,
table, states_to_delete)
mapper = table_to_mapper[table]
_emit_delete_statements(base_mapper, uowtransaction,
cached_connections, mapper, table, delete)
for state, state_dict, mapper, has_identity, connection \
in states_to_delete:
mapper.dispatch.after_delete(mapper, connection, state)
def _organize_states_for_save(base_mapper, states, uowtransaction):
"""Make an initial pass across a set of states for INSERT or
UPDATE.
This includes splitting out into distinct lists for
each, calling before_insert/before_update, obtaining
key information for each state including its dictionary,
mapper, the connection to use for the execution per state,
and the identity flag.
"""
states_to_insert = []
states_to_update = []
for state, dict_, mapper, connection in _connections_for_states(
base_mapper, uowtransaction,
states):
has_identity = bool(state.key)
instance_key = state.key or mapper._identity_key_from_state(state)
row_switch = None
# call before_XXX extensions
if not has_identity:
mapper.dispatch.before_insert(mapper, connection, state)
else:
mapper.dispatch.before_update(mapper, connection, state)
# detect if we have a "pending" instance (i.e. has
# no instance_key attached to it), and another instance
# with the same identity key already exists as persistent.
# convert to an UPDATE if so.
if not has_identity and \
instance_key in uowtransaction.session.identity_map:
instance = \
uowtransaction.session.identity_map[instance_key]
existing = attributes.instance_state(instance)
if not uowtransaction.is_deleted(existing):
raise orm_exc.FlushError(
"New instance %s with identity key %s conflicts "
"with persistent instance %s" %
(state_str(state), instance_key,
state_str(existing)))
base_mapper._log_debug(
"detected row switch for identity %s. "
"will update %s, remove %s from "
"transaction", instance_key,
state_str(state), state_str(existing))
# remove the "delete" flag from the existing element
uowtransaction.remove_state_actions(existing)
row_switch = existing
if not has_identity and not row_switch:
states_to_insert.append(
(state, dict_, mapper, connection,
has_identity, instance_key, row_switch)
)
else:
states_to_update.append(
(state, dict_, mapper, connection,
has_identity, instance_key, row_switch)
)
return states_to_insert, states_to_update
def _organize_states_for_post_update(base_mapper, states,
uowtransaction):
"""Make an initial pass across a set of states for UPDATE
corresponding to post_update.
This includes obtaining key information for each state
including its dictionary, mapper, the connection to use for
the execution per state.
"""
return list(_connections_for_states(base_mapper, uowtransaction,
states))
def _organize_states_for_delete(base_mapper, states, uowtransaction):
"""Make an initial pass across a set of states for DELETE.
This includes calling out before_delete and obtaining
key information for each state including its dictionary,
mapper, the connection to use for the execution per state.
"""
states_to_delete = []
for state, dict_, mapper, connection in _connections_for_states(
base_mapper, uowtransaction,
states):
mapper.dispatch.before_delete(mapper, connection, state)
states_to_delete.append((state, dict_, mapper,
bool(state.key), connection))
return states_to_delete
def _collect_insert_commands(base_mapper, uowtransaction, table,
states_to_insert):
"""Identify sets of values to use in INSERT statements for a
list of states.
"""
insert = []
for state, state_dict, mapper, connection, has_identity, \
instance_key, row_switch in states_to_insert:
if table not in mapper._pks_by_table:
continue
pks = mapper._pks_by_table[table]
params = {}
value_params = {}
has_all_pks = True
for col in mapper._cols_by_table[table]:
if col is mapper.version_id_col:
params[col.key] = mapper.version_id_generator(None)
else:
# pull straight from the dict for
# pending objects
prop = mapper._columntoproperty[col]
value = state_dict.get(prop.key, None)
if value is None:
if col in pks:
has_all_pks = False
elif col.default is None and \
col.server_default is None:
params[col.key] = value
elif isinstance(value, sql.ClauseElement):
value_params[col] = value
else:
params[col.key] = value
insert.append((state, state_dict, params, mapper,
connection, value_params, has_all_pks))
return insert
def _collect_update_commands(base_mapper, uowtransaction,
table, states_to_update):
"""Identify sets of values to use in UPDATE statements for a
list of states.
This function works intricately with the history system
to determine exactly what values should be updated
as well as how the row should be matched within an UPDATE
statement. Includes some tricky scenarios where the primary
key of an object might have been changed.
"""
update = []
for state, state_dict, mapper, connection, has_identity, \
instance_key, row_switch in states_to_update:
if table not in mapper._pks_by_table:
continue
pks = mapper._pks_by_table[table]
params = {}
value_params = {}
hasdata = hasnull = False
for col in mapper._cols_by_table[table]:
if col is mapper.version_id_col:
params[col._label] = \
mapper._get_committed_state_attr_by_column(
row_switch or state,
row_switch and row_switch.dict
or state_dict,
col)
prop = mapper._columntoproperty[col]
history = attributes.get_state_history(
state, prop.key,
attributes.PASSIVE_NO_INITIALIZE
)
if history.added:
params[col.key] = history.added[0]
hasdata = True
else:
params[col.key] = mapper.version_id_generator(
params[col._label])
# HACK: check for history, in case the
# history is only
# in a different table than the one
# where the version_id_col is.
for prop in mapper._columntoproperty.itervalues():
history = attributes.get_state_history(
state, prop.key,
attributes.PASSIVE_NO_INITIALIZE)
if history.added:
hasdata = True
else:
prop = mapper._columntoproperty[col]
history = attributes.get_state_history(
state, prop.key,
attributes.PASSIVE_NO_INITIALIZE)
if history.added:
if isinstance(history.added[0],
sql.ClauseElement):
value_params[col] = history.added[0]
else:
value = history.added[0]
params[col.key] = value
if col in pks:
if history.deleted and \
not row_switch:
# if passive_updates and sync detected
# this was a pk->pk sync, use the new
# value to locate the row, since the
# DB would already have set this
if ("pk_cascaded", state, col) in \
uowtransaction.attributes:
value = history.added[0]
params[col._label] = value
else:
# use the old value to
# locate the row
value = history.deleted[0]
params[col._label] = value
hasdata = True
else:
# row switch logic can reach us here
# remove the pk from the update params
# so the update doesn't
# attempt to include the pk in the
# update statement
del params[col.key]
value = history.added[0]
params[col._label] = value
if value is None:
hasnull = True
else:
hasdata = True
elif col in pks:
value = state.manager[prop.key].impl.get(
state, state_dict)
if value is None:
hasnull = True
params[col._label] = value
if hasdata:
if hasnull:
raise sa_exc.FlushError(
"Can't update table "
"using NULL for primary "
"key value")
update.append((state, state_dict, params, mapper,
connection, value_params))
return update
def _collect_post_update_commands(base_mapper, uowtransaction, table,
states_to_update, post_update_cols):
"""Identify sets of values to use in UPDATE statements for a
list of states within a post_update operation.
"""
update = []
for state, state_dict, mapper, connection in states_to_update:
if table not in mapper._pks_by_table:
continue
pks = mapper._pks_by_table[table]
params = {}
hasdata = False
for col in mapper._cols_by_table[table]:
if col in pks:
params[col._label] = \
mapper._get_state_attr_by_column(
state,
state_dict, col)
elif col in post_update_cols:
prop = mapper._columntoproperty[col]
history = attributes.get_state_history(
state, prop.key,
attributes.PASSIVE_NO_INITIALIZE)
if history.added:
value = history.added[0]
params[col.key] = value
hasdata = True
if hasdata:
update.append((state, state_dict, params, mapper,
connection))
return update
def _collect_delete_commands(base_mapper, uowtransaction, table,
states_to_delete):
"""Identify values to use in DELETE statements for a list of
states to be deleted."""
delete = util.defaultdict(list)
for state, state_dict, mapper, has_identity, connection \
in states_to_delete:
if not has_identity or table not in mapper._pks_by_table:
continue
params = {}
delete[connection].append(params)
for col in mapper._pks_by_table[table]:
params[col.key] = \
value = \
mapper._get_state_attr_by_column(
state, state_dict, col)
if value is None:
raise sa_exc.FlushError(
"Can't delete from table "
"using NULL for primary "
"key value")
if mapper.version_id_col is not None and \
table.c.contains_column(mapper.version_id_col):
params[mapper.version_id_col.key] = \
mapper._get_committed_state_attr_by_column(
state, state_dict,
mapper.version_id_col)
return delete
def _emit_update_statements(base_mapper, uowtransaction,
cached_connections, mapper, table, update):
"""Emit UPDATE statements corresponding to value lists collected
by _collect_update_commands()."""
needs_version_id = mapper.version_id_col is not None and \
table.c.contains_column(mapper.version_id_col)
def update_stmt():
clause = sql.and_()
for col in mapper._pks_by_table[table]:
clause.clauses.append(col == sql.bindparam(col._label,
type_=col.type))
if needs_version_id:
clause.clauses.append(mapper.version_id_col ==\
sql.bindparam(mapper.version_id_col._label,
type_=col.type))
return table.update(clause)
statement = base_mapper._memo(('update', table), update_stmt)
rows = 0
for state, state_dict, params, mapper, \
connection, value_params in update:
if value_params:
c = connection.execute(
statement.values(value_params),
params)
else:
c = cached_connections[connection].\
execute(statement, params)
_postfetch(
mapper,
uowtransaction,
table,
state,
state_dict,
c.context.prefetch_cols,
c.context.postfetch_cols,
c.context.compiled_parameters[0],
value_params)
rows += c.rowcount
if connection.dialect.supports_sane_rowcount:
if rows != len(update):
raise orm_exc.StaleDataError(
"UPDATE statement on table '%s' expected to "
"update %d row(s); %d were matched." %
(table.description, len(update), rows))
elif needs_version_id:
util.warn("Dialect %s does not support updated rowcount "
"- versioning cannot be verified." %
c.dialect.dialect_description,
stacklevel=12)
def _emit_insert_statements(base_mapper, uowtransaction,
cached_connections, table, insert):
"""Emit INSERT statements corresponding to value lists collected
by _collect_insert_commands()."""
statement = base_mapper._memo(('insert', table), table.insert)
for (connection, pkeys, hasvalue, has_all_pks), \
records in groupby(insert,
lambda rec: (rec[4],
rec[2].keys(),
bool(rec[5]),
rec[6])
):
if has_all_pks and not hasvalue:
records = list(records)
multiparams = [rec[2] for rec in records]
c = cached_connections[connection].\
execute(statement, multiparams)
for (state, state_dict, params, mapper,
conn, value_params, has_all_pks), \
last_inserted_params in \
zip(records, c.context.compiled_parameters):
_postfetch(
mapper,
uowtransaction,
table,
state,
state_dict,
c.context.prefetch_cols,
c.context.postfetch_cols,
last_inserted_params,
value_params)
else:
for state, state_dict, params, mapper, \
connection, value_params, \
has_all_pks in records:
if value_params:
result = connection.execute(
statement.values(value_params),
params)
else:
result = cached_connections[connection].\
execute(statement, params)
primary_key = result.context.inserted_primary_key
if primary_key is not None:
# set primary key attributes
for pk, col in zip(primary_key,
mapper._pks_by_table[table]):
prop = mapper._columntoproperty[col]
if state_dict.get(prop.key) is None:
# TODO: would rather say:
#state_dict[prop.key] = pk
mapper._set_state_attr_by_column(
state,
state_dict,
col, pk)
_postfetch(
mapper,
uowtransaction,
table,
state,
state_dict,
result.context.prefetch_cols,
result.context.postfetch_cols,
result.context.compiled_parameters[0],
value_params)
def _emit_post_update_statements(base_mapper, uowtransaction,
cached_connections, mapper, table, update):
"""Emit UPDATE statements corresponding to value lists collected
by _collect_post_update_commands()."""
def update_stmt():
clause = sql.and_()
for col in mapper._pks_by_table[table]:
clause.clauses.append(col == sql.bindparam(col._label,
type_=col.type))
return table.update(clause)
statement = base_mapper._memo(('post_update', table), update_stmt)
# execute each UPDATE in the order according to the original
# list of states to guarantee row access order, but
# also group them into common (connection, cols) sets
# to support executemany().
for key, grouper in groupby(
update, lambda rec: (rec[4], rec[2].keys())
):
connection = key[0]
multiparams = [params for state, state_dict,
params, mapper, conn in grouper]
cached_connections[connection].\
execute(statement, multiparams)
def _emit_delete_statements(base_mapper, uowtransaction, cached_connections,
mapper, table, delete):
"""Emit DELETE statements corresponding to value lists collected
by _collect_delete_commands()."""
need_version_id = mapper.version_id_col is not None and \
table.c.contains_column(mapper.version_id_col)
def delete_stmt():
clause = sql.and_()
for col in mapper._pks_by_table[table]:
clause.clauses.append(
col == sql.bindparam(col.key, type_=col.type))
if need_version_id:
clause.clauses.append(
mapper.version_id_col ==
sql.bindparam(
mapper.version_id_col.key,
type_=mapper.version_id_col.type
)
)
return table.delete(clause)
for connection, del_objects in delete.iteritems():
statement = base_mapper._memo(('delete', table), delete_stmt)
connection = cached_connections[connection]
if need_version_id:
# TODO: need test coverage for this [ticket:1761]
if connection.dialect.supports_sane_rowcount:
rows = 0
# execute deletes individually so that versioned
# rows can be verified
for params in del_objects:
c = connection.execute(statement, params)
rows += c.rowcount
if rows != len(del_objects):
raise orm_exc.StaleDataError(
"DELETE statement on table '%s' expected to "
"delete %d row(s); %d were matched." %
(table.description, len(del_objects), c.rowcount)
)
else:
util.warn(
"Dialect %s does not support deleted rowcount "
"- versioning cannot be verified." %
connection.dialect.dialect_description,
stacklevel=12)
connection.execute(statement, del_objects)
else:
connection.execute(statement, del_objects)
def _finalize_insert_update_commands(base_mapper, uowtransaction,
states_to_insert, states_to_update):
"""finalize state on states that have been inserted or updated,
including calling after_insert/after_update events.
"""
for state, state_dict, mapper, connection, has_identity, \
instance_key, row_switch in states_to_insert + \
states_to_update:
if mapper._readonly_props:
readonly = state.unmodified_intersection(
[p.key for p in mapper._readonly_props
if p.expire_on_flush or p.key not in state.dict]
)
if readonly:
state.expire_attributes(state.dict, readonly)
# if eager_defaults option is enabled,
# refresh whatever has been expired.
if base_mapper.eager_defaults and state.unloaded:
state.key = base_mapper._identity_key_from_state(state)
uowtransaction.session.query(base_mapper)._load_on_ident(
state.key, refresh_state=state,
only_load_props=state.unloaded)
# call after_XXX extensions
if not has_identity:
mapper.dispatch.after_insert(mapper, connection, state)
else:
mapper.dispatch.after_update(mapper, connection, state)
def _postfetch(mapper, uowtransaction, table,
state, dict_, prefetch_cols, postfetch_cols,
params, value_params):
"""Expire attributes in need of newly persisted database state,
after an INSERT or UPDATE statement has proceeded for that
state."""
if mapper.version_id_col is not None:
prefetch_cols = list(prefetch_cols) + [mapper.version_id_col]
for c in prefetch_cols:
if c.key in params and c in mapper._columntoproperty:
mapper._set_state_attr_by_column(state, dict_, c, params[c.key])
if postfetch_cols:
state.expire_attributes(state.dict,
[mapper._columntoproperty[c].key
for c in postfetch_cols if c in
mapper._columntoproperty]
)
# synchronize newly inserted ids from one table to the next
# TODO: this still goes a little too often. would be nice to
# have definitive list of "columns that changed" here
for m, equated_pairs in mapper._table_to_equated[table]:
sync.populate(state, m, state, m,
equated_pairs,
uowtransaction,
mapper.passive_updates)
def _connections_for_states(base_mapper, uowtransaction, states):
"""Return an iterator of (state, state.dict, mapper, connection).
The states are sorted according to _sort_states, then paired
with the connection they should be using for the given
unit of work transaction.
"""
# if session has a connection callable,
# organize individual states with the connection
# to use for update
if uowtransaction.session.connection_callable:
connection_callable = \
uowtransaction.session.connection_callable
else:
connection = uowtransaction.transaction.connection(
base_mapper)
connection_callable = None
for state in _sort_states(states):
if connection_callable:
connection = connection_callable(base_mapper, state.obj())
mapper = _state_mapper(state)
yield state, state.dict, mapper, connection
def _cached_connection_dict(base_mapper):
# dictionary of connection->connection_with_cache_options.
return util.PopulateDict(
lambda conn:conn.execution_options(
compiled_cache=base_mapper._compiled_cache
))
def _sort_states(states):
pending = set(states)
persistent = set(s for s in pending if s.key is not None)
pending.difference_update(persistent)
return sorted(pending, key=operator.attrgetter("insert_order")) + \
sorted(persistent, key=lambda q:q.key[1])
libs/sqlalchemy/orm/query.py
+123 -35
View File
@@ -133,7 +133,7 @@ class Query(object):
with_polymorphic = mapper._with_polymorphic_mappers
if mapper.mapped_table not in \
self._polymorphic_adapters:
self.__mapper_loads_polymorphically_with(mapper,
self._mapper_loads_polymorphically_with(mapper,
sql_util.ColumnAdapter(
selectable,
mapper._equivalent_columns))
@@ -150,7 +150,7 @@ class Query(object):
is_aliased_class, with_polymorphic)
ent.setup_entity(entity, *d[entity])
def __mapper_loads_polymorphically_with(self, mapper, adapter):
def _mapper_loads_polymorphically_with(self, mapper, adapter):
for m2 in mapper._with_polymorphic_mappers:
self._polymorphic_adapters[m2] = adapter
for m in m2.iterate_to_root():
@@ -174,10 +174,6 @@ class Query(object):
self._from_obj_alias = sql_util.ColumnAdapter(
self._from_obj[0], equivs)
def _get_polymorphic_adapter(self, entity, selectable):
self.__mapper_loads_polymorphically_with(entity.mapper,
sql_util.ColumnAdapter(selectable,
entity.mapper._equivalent_columns))
def _reset_polymorphic_adapter(self, mapper):
for m2 in mapper._with_polymorphic_mappers:
@@ -276,6 +272,7 @@ class Query(object):
return self._select_from_entity or \
self._entity_zero().entity_zero
@property
def _mapper_entities(self):
# TODO: this is wrong, its hardcoded to "primary entity" when
@@ -324,13 +321,6 @@ class Query(object):
)
return self._entity_zero()
def _generate_mapper_zero(self):
if not getattr(self._entities[0], 'primary_entity', False):
raise sa_exc.InvalidRequestError(
"No primary mapper set up for this Query.")
entity = self._entities[0]._clone()
self._entities = [entity] + self._entities[1:]
return entity
def __all_equivs(self):
equivs = {}
@@ -460,6 +450,62 @@ class Query(object):
"""
return self.enable_eagerloads(False).statement.alias(name=name)
def cte(self, name=None, recursive=False):
"""Return the full SELECT statement represented by this :class:`.Query`
represented as a common table expression (CTE).
The :meth:`.Query.cte` method is new in 0.7.6.
Parameters and usage are the same as those of the
:meth:`._SelectBase.cte` method; see that method for
further details.
Here is the `Postgresql WITH
RECURSIVE example <http://www.postgresql.org/docs/8.4/static/queries-with.html>`_.
Note that, in this example, the ``included_parts`` cte and the ``incl_alias`` alias
of it are Core selectables, which
means the columns are accessed via the ``.c.`` attribute. The ``parts_alias``
object is an :func:`.orm.aliased` instance of the ``Part`` entity, so column-mapped
attributes are available directly::
from sqlalchemy.orm import aliased
class Part(Base):
__tablename__ = 'part'
part = Column(String, primary_key=True)
sub_part = Column(String, primary_key=True)
quantity = Column(Integer)
included_parts = session.query(
Part.sub_part,
Part.part,
Part.quantity).\\
filter(Part.part=="our part").\\
cte(name="included_parts", recursive=True)
incl_alias = aliased(included_parts, name="pr")
parts_alias = aliased(Part, name="p")
included_parts = included_parts.union_all(
session.query(
parts_alias.part,
parts_alias.sub_part,
parts_alias.quantity).\\
filter(parts_alias.part==incl_alias.c.sub_part)
)
q = session.query(
included_parts.c.sub_part,
func.sum(included_parts.c.quantity).label('total_quantity')
).\\
group_by(included_parts.c.sub_part)
See also:
:meth:`._SelectBase.cte`
"""
return self.enable_eagerloads(False).statement.cte(name=name, recursive=recursive)
def label(self, name):
"""Return the full SELECT statement represented by this :class:`.Query`, converted
to a scalar subquery with a label of the given name.
@@ -601,7 +647,12 @@ class Query(object):
such as concrete table mappers.
"""
entity = self._generate_mapper_zero()
if not getattr(self._entities[0], 'primary_entity', False):
raise sa_exc.InvalidRequestError(
"No primary mapper set up for this Query.")
entity = self._entities[0]._clone()
self._entities = [entity] + self._entities[1:]
entity.set_with_polymorphic(self,
cls_or_mappers,
selectable=selectable,
@@ -1041,7 +1092,22 @@ class Query(object):
@_generative()
def with_lockmode(self, mode):
"""Return a new Query object with the specified locking mode."""
"""Return a new Query object with the specified locking mode.
:param mode: a string representing the desired locking mode. A
corresponding value is passed to the ``for_update`` parameter of
:meth:`~sqlalchemy.sql.expression.select` when the query is
executed. Valid values are:
``'update'`` - passes ``for_update=True``, which translates to
``FOR UPDATE`` (standard SQL, supported by most dialects)
``'update_nowait'`` - passes ``for_update='nowait'``, which
translates to ``FOR UPDATE NOWAIT`` (supported by Oracle)
``'read'`` - passes ``for_update='read'``, which translates to
``LOCK IN SHARE MODE`` (supported by MySQL).
"""
self._lockmode = mode
@@ -1583,7 +1649,6 @@ class Query(object):
consistent format with which to form the actual JOIN constructs.
"""
self._polymorphic_adapters = self._polymorphic_adapters.copy()
if not from_joinpoint:
self._reset_joinpoint()
@@ -1683,6 +1748,8 @@ class Query(object):
onclause, outerjoin, create_aliases, prop):
"""append a JOIN to the query's from clause."""
self._polymorphic_adapters = self._polymorphic_adapters.copy()
if left is None:
if self._from_obj:
left = self._from_obj[0]
@@ -1696,7 +1763,29 @@ class Query(object):
"are the same entity" %
(left, right))
left_mapper, left_selectable, left_is_aliased = _entity_info(left)
right, right_is_aliased, onclause = self._prepare_right_side(
right, onclause,
outerjoin, create_aliases,
prop)
# if joining on a MapperProperty path,
# track the path to prevent redundant joins
if not create_aliases and prop:
self._update_joinpoint({
'_joinpoint_entity':right,
'prev':((left, right, prop.key), self._joinpoint)
})
else:
self._joinpoint = {
'_joinpoint_entity':right
}
self._join_to_left(left, right,
right_is_aliased,
onclause, outerjoin)
def _prepare_right_side(self, right, onclause, outerjoin,
create_aliases, prop):
right_mapper, right_selectable, right_is_aliased = _entity_info(right)
if right_mapper:
@@ -1741,24 +1830,13 @@ class Query(object):
right = aliased(right)
need_adapter = True
# if joining on a MapperProperty path,
# track the path to prevent redundant joins
if not create_aliases and prop:
self._update_joinpoint({
'_joinpoint_entity':right,
'prev':((left, right, prop.key), self._joinpoint)
})
else:
self._joinpoint = {
'_joinpoint_entity':right
}
# if an alias() of the right side was generated here,
# apply an adapter to all subsequent filter() calls
# until reset_joinpoint() is called.
if need_adapter:
self._filter_aliases = ORMAdapter(right,
equivalents=right_mapper and right_mapper._equivalent_columns or {},
equivalents=right_mapper and
right_mapper._equivalent_columns or {},
chain_to=self._filter_aliases)
# if the onclause is a ClauseElement, adapt it with any
@@ -1771,7 +1849,7 @@ class Query(object):
# ensure that columns retrieved from this target in the result
# set are also adapted.
if aliased_entity and not create_aliases:
self.__mapper_loads_polymorphically_with(
self._mapper_loads_polymorphically_with(
right_mapper,
ORMAdapter(
right,
@@ -1779,6 +1857,11 @@ class Query(object):
)
)
return right, right_is_aliased, onclause
def _join_to_left(self, left, right, right_is_aliased, onclause, outerjoin):
left_mapper, left_selectable, left_is_aliased = _entity_info(left)
# this is an overly broad assumption here, but there's a
# very wide variety of situations where we rely upon orm.join's
# adaption to glue clauses together, with joined-table inheritance's
@@ -2959,7 +3042,9 @@ class _MapperEntity(_QueryEntity):
# with_polymorphic() can be applied to aliases
if not self.is_aliased_class:
self.selectable = from_obj
self.adapter = query._get_polymorphic_adapter(self, from_obj)
query._mapper_loads_polymorphically_with(self.mapper,
sql_util.ColumnAdapter(from_obj,
self.mapper._equivalent_columns))
filter_fn = id
@@ -3086,8 +3171,9 @@ class _MapperEntity(_QueryEntity):
class _ColumnEntity(_QueryEntity):
"""Column/expression based entity."""
def __init__(self, query, column):
def __init__(self, query, column, namespace=None):
self.expr = column
self.namespace = namespace
if isinstance(column, basestring):
column = sql.literal_column(column)
@@ -3106,7 +3192,7 @@ class _ColumnEntity(_QueryEntity):
for c in column._select_iterable:
if c is column:
break
_ColumnEntity(query, c)
_ColumnEntity(query, c, namespace=column)
if c is not column:
return
@@ -3147,12 +3233,14 @@ class _ColumnEntity(_QueryEntity):
if self.entities:
self.entity_zero = list(self.entities)[0]
elif self.namespace is not None:
self.entity_zero = self.namespace
else:
self.entity_zero = None
@property
def entity_zero_or_selectable(self):
if self.entity_zero:
if self.entity_zero is not None:
return self.entity_zero
elif self.actual_froms:
return list(self.actual_froms)[0]
libs/sqlalchemy/orm/scoping.py
+7 -5
View File
@@ -41,8 +41,9 @@ class ScopedSession(object):
scope = kwargs.pop('scope', False)
if scope is not None:
if self.registry.has():
raise sa_exc.InvalidRequestError("Scoped session is already present; "
"no new arguments may be specified.")
raise sa_exc.InvalidRequestError(
"Scoped session is already present; "
"no new arguments may be specified.")
else:
sess = self.session_factory(**kwargs)
self.registry.set(sess)
@@ -70,8 +71,8 @@ class ScopedSession(object):
self.session_factory.configure(**kwargs)
def query_property(self, query_cls=None):
"""return a class property which produces a `Query` object against the
class when called.
"""return a class property which produces a `Query` object
against the class when called.
e.g.::
@@ -121,7 +122,8 @@ def makeprop(name):
def get(self):
return getattr(self.registry(), name)
return property(get, set)
for prop in ('bind', 'dirty', 'deleted', 'new', 'identity_map', 'is_active', 'autoflush'):
for prop in ('bind', 'dirty', 'deleted', 'new', 'identity_map',
'is_active', 'autoflush', 'no_autoflush'):
setattr(ScopedSession, prop, makeprop(prop))
def clslevel(name):
libs/sqlalchemy/orm/session.py
+42 -1
View File
@@ -99,7 +99,7 @@ def sessionmaker(bind=None, class_=None, autoflush=True, autocommit=False,
kwargs.update(new_kwargs)
return type("Session", (Sess, class_), {})
return type("SessionMaker", (Sess, class_), {})
class SessionTransaction(object):
@@ -978,6 +978,34 @@ class Session(object):
return self._query_cls(entities, self, **kwargs)
@property
@util.contextmanager
def no_autoflush(self):
"""Return a context manager that disables autoflush.
e.g.::
with session.no_autoflush:
some_object = SomeClass()
session.add(some_object)
# won't autoflush
some_object.related_thing = session.query(SomeRelated).first()
Operations that proceed within the ``with:`` block
will not be subject to flushes occurring upon query
access. This is useful when initializing a series
of objects which involve existing database queries,
where the uncompleted object should not yet be flushed.
New in 0.7.6.
"""
autoflush = self.autoflush
self.autoflush = False
yield self
self.autoflush = autoflush
def _autoflush(self):
if self.autoflush and not self._flushing:
self.flush()
@@ -1772,6 +1800,19 @@ class Session(object):
return self.transaction and self.transaction.is_active
identity_map = None
"""A mapping of object identities to objects themselves.
Iterating through ``Session.identity_map.values()`` provides
access to the full set of persistent objects (i.e., those
that have row identity) currently in the session.
See also:
:func:`.identity_key` - operations involving identity keys.
"""
@property
def _dirty_states(self):
"""The set of all persistent states considered dirty.
libs/sqlalchemy/orm/sync.py
+1
View File
@@ -6,6 +6,7 @@
"""private module containing functions used for copying data
between instances based on join conditions.
"""
from sqlalchemy.orm import exc, util as mapperutil, attributes
libs/sqlalchemy/orm/unitofwork.py
+6 -6
View File
@@ -14,7 +14,7 @@ organizes them in order of dependency, and executes.
from sqlalchemy import util, event
from sqlalchemy.util import topological
from sqlalchemy.orm import attributes, interfaces
from sqlalchemy.orm import attributes, interfaces, persistence
from sqlalchemy.orm import util as mapperutil
session = util.importlater("sqlalchemy.orm", "session")
@@ -462,7 +462,7 @@ class IssuePostUpdate(PostSortRec):
states, cols = uow.post_update_states[self.mapper]
states = [s for s in states if uow.states[s][0] == self.isdelete]
self.mapper._post_update(states, uow, cols)
persistence.post_update(self.mapper, states, uow, cols)
class SaveUpdateAll(PostSortRec):
def __init__(self, uow, mapper):
@@ -470,7 +470,7 @@ class SaveUpdateAll(PostSortRec):
assert mapper is mapper.base_mapper
def execute(self, uow):
self.mapper._save_obj(
persistence.save_obj(self.mapper,
uow.states_for_mapper_hierarchy(self.mapper, False, False),
uow
)
@@ -493,7 +493,7 @@ class DeleteAll(PostSortRec):
assert mapper is mapper.base_mapper
def execute(self, uow):
self.mapper._delete_obj(
persistence.delete_obj(self.mapper,
uow.states_for_mapper_hierarchy(self.mapper, True, False),
uow
)
@@ -551,7 +551,7 @@ class SaveUpdateState(PostSortRec):
if r.__class__ is cls_ and
r.mapper is mapper]
recs.difference_update(our_recs)
mapper._save_obj(
persistence.save_obj(mapper,
[self.state] +
[r.state for r in our_recs],
uow)
@@ -575,7 +575,7 @@ class DeleteState(PostSortRec):
r.mapper is mapper]
recs.difference_update(our_recs)
states = [self.state] + [r.state for r in our_recs]
mapper._delete_obj(
persistence.delete_obj(mapper,
[s for s in states if uow.states[s][0]],
uow)
libs/sqlalchemy/orm/util.py
+47 -27
View File
@@ -11,6 +11,7 @@ from sqlalchemy.orm.interfaces import MapperExtension, EXT_CONTINUE,\
PropComparator, MapperProperty
from sqlalchemy.orm import attributes, exc
import operator
import re
mapperlib = util.importlater("sqlalchemy.orm", "mapperlib")
@@ -20,38 +21,52 @@ all_cascades = frozenset(("delete", "delete-orphan", "all", "merge",
_INSTRUMENTOR = ('mapper', 'instrumentor')
class CascadeOptions(dict):
class CascadeOptions(frozenset):
"""Keeps track of the options sent to relationship().cascade"""
def __init__(self, arg=""):
if not arg:
values = set()
else:
values = set(c.strip() for c in arg.split(','))
_add_w_all_cascades = all_cascades.difference([
'all', 'none', 'delete-orphan'])
_allowed_cascades = all_cascades
for name in ['save-update', 'delete', 'refresh-expire',
'merge', 'expunge']:
boolean = name in values or 'all' in values
setattr(self, name.replace('-', '_'), boolean)
if boolean:
self[name] = True
def __new__(cls, arg):
values = set([
c for c
in re.split('\s*,\s*', arg or "")
if c
])
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(
values.difference(cls._allowed_cascades)
)])
)
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan:
self['delete-orphan'] = True
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade option requires "
"'delete'.")
for x in values:
if x not in all_cascades:
raise sa_exc.ArgumentError("Invalid cascade option '%s'" % x)
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def __repr__(self):
return "CascadeOptions(%s)" % repr(",".join(
[x for x in ['delete', 'save_update', 'merge', 'expunge',
'delete_orphan', 'refresh-expire']
if getattr(self, x, False) is True]))
return "CascadeOptions(%r)" % (
",".join([x for x in sorted(self)])
)
def _validator_events(desc, key, validator):
"""Runs a validation method on an attribute value to be set or appended."""
@@ -557,15 +572,20 @@ def _entity_descriptor(entity, key):
attribute.
"""
if not isinstance(entity, (AliasedClass, type)):
entity = entity.class_
if isinstance(entity, expression.FromClause):
description = entity
entity = entity.c
elif not isinstance(entity, (AliasedClass, type)):
description = entity = entity.class_
else:
description = entity
try:
return getattr(entity, key)
except AttributeError:
raise sa_exc.InvalidRequestError(
"Entity '%s' has no property '%s'" %
(entity, key)
(description, key)
)
def _orm_columns(entity):