- SADeprecationWarning has moved to exceptions. An alias remains in logging until 0.5.
0.4.2p4
------
- orm
- - proper error message is raised when trying to
- access expired instance attributes with no session
- present
-
+ - proper error message is raised when trying to access
+ expired instance attributes with no session present
+
- added a mapper() flag "eager_defaults"; when set to
- True, defaults that are generated during an INSERT
- or UPDATE operation are post-fetched immediately,
- instead of being deferred until later. This mimics
- the old 0.3 behavior.
-
+ True, defaults that are generated during an INSERT or
+ UPDATE operation are post-fetched immediately, instead
+ of being deferred until later. This mimics the old 0.3
+ behavior.
+
+- general
+ - warnings are now issued as type exceptions.SAWarning.
+
- dialects
- - finally added PGMacAddr type to postgres
- [ticket:580]
-
+ - finally added PGMacAddr type to postgres [ticket:580]
+
0.4.2p3
------
- general
- sub version numbering scheme changed to suite
setuptools version number rules; easy_install -u
should now get this version over 0.4.2.
-
+
- sql
- Text type is properly exported now and does not
raise a warning on DDL create; String types with no
- new UnicodeText type is added, to specify an
encoded, unlengthed Text type
-
+
- fixed bug in union() so that select() statements
which don't derive from FromClause objects can be
unioned
- orm
- fixed bug with session.dirty when using "mutable
scalars" (such as PickleTypes)
-
+
- added a more descriptive error message when flushing
on a relation() that has non-locally-mapped columns
in its primary or secondary join condition
-
+
- dialects
- Fixed reflection of mysql empty string column
defaults.
import datetime
-import warnings
from sqlalchemy import exceptions, schema, types as sqltypes, sql, util
from sqlalchemy.engine import base, default
# get the data types and lengths
coltype = ischema_names.get(row['ftype'].rstrip())
if coltype is None:
- warnings.warn(RuntimeWarning("Did not recognize type '%s' of column '%s'" % (str(row['ftype']), name)))
+ util.warn("Did not recognize type '%s' of column '%s'" %
+ (str(row['ftype']), name))
coltype = sqltypes.NULLTYPE
else:
coltype = coltype(row)
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
-import datetime, warnings
+import datetime
from sqlalchemy import sql, schema, exceptions, pool
from sqlalchemy.sql import compiler
try:
coltype = ischema_names[coltype]
except KeyError:
- warnings.warn(RuntimeWarning("Did not recognize type '%s' of column '%s'" % (coltype, name)))
+ util.warn("Did not recognize type '%s' of column '%s'" %
+ (coltype, name))
coltype = sqltypes.NULLTYPE
colargs = []
this.
"""
-import datetime, itertools, re, warnings
+import datetime, itertools, re
from sqlalchemy import exceptions, schema, sql, util
from sqlalchemy.sql import operators as sql_operators, expression as sql_expr
type_cls = ischema_names[col_type.lower()]
type_instance = type_cls(*type_args, **type_kw)
except KeyError:
- warnings.warn(RuntimeWarning(
- "Did not recognize type '%s' of column '%s'" %
- (col_type, name)))
+ util.warn("Did not recognize type '%s' of column '%s'" %
+ (col_type, name))
type_instance = sqltypes.NullType
col_kw = {'autoincrement': False}
"""
-import datetime, random, warnings, re, sys, operator
+import datetime, operator, random, re, sys
from sqlalchemy import sql, schema, exceptions, util
from sqlalchemy.sql import compiler, expression, operators as sqlops
coltype = MSText()
else:
if coltype is None:
- warnings.warn(RuntimeWarning("Did not recognize type '%s' of column '%s'" % (type, name)))
+ util.warn("Did not recognize type '%s' of column '%s'" %
+ (type, name))
coltype = sqltypes.NULLTYPE
elif coltype in (MSNVarchar, AdoMSNVarchar) and charlen == -1:
timely information affecting MySQL in SQLAlchemy.
"""
-import re, datetime, inspect, warnings, sys
+import datetime, inspect, re, sys
from array import array as _array
from sqlalchemy import exceptions, logging, schema, sql, util
if 'character_set' in opts:
return opts['character_set']
else:
- warnings.warn(RuntimeWarning(
+ util.warn(
"Could not detect the connection character set with this "
"combination of MySQL server and MySQL-python. "
- "MySQL-python >= 1.2.2 is recommended. Assuming latin1."))
+ "MySQL-python >= 1.2.2 is recommended. Assuming latin1.")
return 'latin1'
def _detect_casing(self, connection, charset=None):
else:
type_, spec = self.parse_constraints(line)
if type_ is None:
- warnings.warn(
- RuntimeWarning("Unknown schema content: %s" %
- repr(line)))
+ util.warn("Unknown schema content: %r" % line)
elif type_ == 'key':
keys.append(spec)
elif type_ == 'constraint':
def _add_column(self, table, line, charset, only=None):
spec = self.parse_column(line)
if not spec:
- warnings.warn(RuntimeWarning(
- "Unknown column definition %s" % line))
+ util.warn("Unknown column definition %r" % line)
return
if not spec['full']:
- warnings.warn(RuntimeWarning(
- "Incomplete reflection of column definition %s" % line))
+ util.warn("Incomplete reflection of column definition %r" % line)
name, type_, args, notnull = \
spec['name'], spec['coltype'], spec['arg'], spec['notnull']
try:
col_type = ischema_names[type_]
except KeyError:
- warnings.warn(RuntimeWarning(
- "Did not recognize type '%s' of column '%s'" %
- (type_, name)))
+ util.warn("Did not recognize type '%s' of column '%s'" %
+ (type_, name))
col_type = sqltypes.NullType
# Column type positional arguments eg. varchar(32)
# the MIT License: http://www.opensource.org/licenses/mit-license.php
-import re, warnings, random
+import datetime, random, re
from sqlalchemy import util, sql, schema, exceptions, logging
from sqlalchemy.engine import default, base
from sqlalchemy.sql import operators as sql_operators
from sqlalchemy import types as sqltypes
-import datetime
-
class OracleNumeric(sqltypes.Numeric):
def get_col_spec(self):
try:
coltype = ischema_names[coltype]
except KeyError:
- warnings.warn(RuntimeWarning("Did not recognize type '%s' of column '%s'" % (coltype, colname)))
+ util.warn("Did not recognize type '%s' of column '%s'" %
+ (coltype, colname))
coltype = sqltypes.NULLTYPE
colargs = []
fks[cons_name] = fk
if remote_table is None:
# ticket 363
- warnings.warn("Got 'None' querying 'table_name' from all_cons_columns%(dblink)s - does the user have proper rights to the table?" % {'dblink':dblink})
+ util.warn(
+ ("Got 'None' querying 'table_name' from "
+ "all_cons_columns%(dblink)s - does the user have "
+ "proper rights to the table?") % {'dblink':dblink})
continue
refspec = ".".join([remote_table, remote_column])
schema.Table(remote_table, table.metadata, autoload=True, autoload_with=connection, owner=remote_owner)
postgres_returning=[empl.c.id, empl.c.salary]).execute().fetchall()
"""
-import re, random, warnings, string
+import random, re, string
from sqlalchemy import sql, schema, exceptions, util
from sqlalchemy.engine import base, default
if is_array:
coltype = PGArray(coltype)
else:
- warnings.warn(RuntimeWarning("Did not recognize type '%s' of column '%s'" % (attype, name)))
+ util.warn("Did not recognize type '%s' of column '%s'" %
+ (attype, name))
coltype = sqltypes.NULLTYPE
colargs= []
# the MIT License: http://www.opensource.org/licenses/mit-license.php
-import re
+import datetime, re, time
from sqlalchemy import schema, exceptions, pool, PassiveDefault
from sqlalchemy.engine import default
import sqlalchemy.types as sqltypes
-import datetime,time, warnings
import sqlalchemy.util as util
from sqlalchemy.sql import compiler
if self.dbapi is not None:
sqlite_ver = self.dbapi.version_info
if sqlite_ver < (2,1,'3'):
- warnings.warn(RuntimeWarning("The installed version of pysqlite2 (%s) is out-dated, and will cause errors in some cases. Version 2.1.3 or greater is recommended." % '.'.join([str(subver) for subver in sqlite_ver])))
+ util.warn(
+ ("The installed version of pysqlite2 (%s) is out-dated "
+ "and will cause errors in some cases. Version 2.1.3 "
+ "or greater is recommended.") %
+ '.'.join([str(subver) for subver in sqlite_ver]))
self.supports_cast = (self.dbapi is None or vers(self.dbapi.sqlite_version) >= vers("3.2.3"))
def dbapi(cls):
try:
coltype = ischema_names[coltype]
except KeyError:
- warnings.warn(RuntimeWarning("Did not recognize type '%s' of column '%s'" % (coltype, name)))
+ util.warn("Did not recognize type '%s' of column '%s'" %
+ (coltype, name))
coltype = sqltypes.NullType
if args is not None:
* Tested on 'Adaptive Server Anywhere 9' (version 9.0.1.1751)
"""
-import datetime, random, warnings, operator
+import datetime, operator, random
from sqlalchemy import util, sql, schema, exceptions
from sqlalchemy.sql import compiler, expression
coltype = SybaseText()
else:
if coltype is None:
- warnings.warn(RuntimeWarning("Did not recognize type '%s' of column '%s'" % (type, name)))
+ util.warn("Did not recognize type '%s' of column '%s'" %
+ (type, name))
coltype = sqltypes.NULLTYPE
coltype = coltype(*args)
colargs= []
class CompileError(SQLAlchemyError):
"""Raised when an error occurs during SQL compilation"""
-
+
class TimeoutError(SQLAlchemyError):
"""Raised when a connection pool times out on getting a connection."""
class CircularDependencyError(SQLAlchemyError):
"""Raised by topological sorts when a circular dependency is detected"""
-
+
class FlushError(SQLAlchemyError):
"""Raised when an invalid condition is detected upon a ``flush()``."""
"""
class UnmappedColumnError(InvalidRequestError):
- """A mapper was asked to return mapped information about a column
+ """A mapper was asked to return mapped information about a column
which it does not map"""
-
+
class NoSuchTableError(InvalidRequestError):
"""SQLAlchemy was asked to load a table's definition from the
database, but the table doesn't exist.
class DisconnectionError(SQLAlchemyError):
"""Raised within ``Pool`` when a disconnect is detected on a raw DB-API connection.
-
+
This error is consumed internally by a connection pool. It can be raised by
a ``PoolListener`` so that the host pool forces a disconnect.
"""
the exception object in the ``statement`` and ``params`` attributes.
The wrapped exception object is available in the ``orig`` attribute.
- Its type and properties are DB-API implementation specific.
+ Its type and properties are DB-API implementation specific.
"""
def instance(cls, statement, params, orig, connection_invalidated=False):
# DBAPIError didn't exist.
if isinstance(orig, (KeyboardInterrupt, SystemExit)):
return orig
-
+
if orig is not None:
name, glob = orig.__class__.__name__, globals()
if name in glob and issubclass(glob[name], DBAPIError):
cls = glob[name]
-
+
return cls(statement, params, orig, connection_invalidated)
instance = classmethod(instance)
-
+
def __init__(self, statement, params, orig, connection_invalidated=False):
try:
text = str(orig)
class NotSupportedError(DatabaseError):
"""Wraps a DB-API NotSupportedError."""
+
+# Warnings
+class SADeprecationWarning(DeprecationWarning):
+ """Issued once per usage of a deprecated API."""
+
+class SAWarning(RuntimeWarning):
+ """Issued at runtime."""
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
-"""Provides a few functions used by instances to turn on/off their
-logging, including support for the usual "echo" parameter.
+"""Logging control and utilities.
-Control of logging for SA can be performed from the regular python
-logging module. The regular dotted module namespace is used, starting
-at 'sqlalchemy'. For class-level logging, the class name is appended,
-and for instance-level logging, the hex id of the instance is
-appended.
+Provides a few functions used by instances to turn on/off their logging,
+including support for the usual "echo" parameter.
-The "echo" keyword parameter which is available on some SA objects
-corresponds to an instance-level logger for that instance.
+Control of logging for SA can be performed from the regular python logging
+module. The regular dotted module namespace is used, starting at
+'sqlalchemy'. For class-level logging, the class name is appended, and for
+instance-level logging, the hex id of the instance is appended.
+
+The "echo" keyword parameter which is available on some SA objects corresponds
+to an instance-level logger for that instance.
E.g.::
is equivalent to::
import logging
- logging.getLogger('sqlalchemy.engine.Engine.%s' % hex(id(engine))).setLevel(logging.DEBUG)
+ logger = logging.getLogger('sqlalchemy.engine.Engine.%s' % hex(id(engine)))
+ logger.setLevel(logging.DEBUG)
"""
import sys, warnings
+import sqlalchemy.exceptions as sa_exc
# py2.5 absolute imports will fix....
logging = __import__('logging')
-# why is this in the "logging" module?
-class SADeprecationWarning(DeprecationWarning):
- pass
-
+# moved to sqlalchemy.exceptions. this alias will be removed in 0.5.
+SADeprecationWarning = sa_exc.SADeprecationWarning
+
rootlogger = logging.getLogger('sqlalchemy')
rootlogger.setLevel(logging.WARN)
-warnings.filterwarnings("once", category=SADeprecationWarning)
+warnings.filterwarnings("once", category=sa_exc.SADeprecationWarning)
default_enabled = False
def default_logging(name):
if not default_enabled:
default_enabled = True
handler = logging.StreamHandler(sys.stdout)
- handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(name)s %(message)s'))
+ handler.setFormatter(logging.Formatter(
+ '%(asctime)s %(levelname)s %(name)s %(message)s'))
rootlogger.addHandler(handler)
def _get_instance_name(instance):
- # since getLogger() does not have any way of removing logger objects from memory,
- # instance logging displays the instance id as a modulus of 16 to prevent endless memory growth
- # also speeds performance as logger initialization is apparently slow
- return instance.__class__.__module__ + "." + instance.__class__.__name__ + ".0x.." + hex(id(instance))[-2:]
+ # since getLogger() does not have any way of removing logger objects from
+ # memory, instance logging displays the instance id as a modulus of 16 to
+ # prevent endless memory growth also speeds performance as logger
+ # initialization is apparently slow
+ return "%s.%s.0x..%s" % (instance.__class__.__module__,
+ instance.__class__.__name__,
+ hex(id(instance))[-2:])
+ return (instance.__class__.__module__ + "." + instance.__class__.__name__ +
+ ".0x.." + hex(id(instance))[-2:])
def class_logger(cls):
return logging.getLogger(cls.__module__ + "." + cls.__name__)
return l
class echo_property(object):
- __doc__ = """when ``True``, enable log output for this element.
+ __doc__ = """\
+ When ``True``, enable log output for this element.
+
- This has the effect of setting the Python logging level for the
- namespace of this element's class and object reference. A value
- of boolean ``True`` indicates that the loglevel ``logging.INFO`` will be
- set for the logger, whereas the string value ``debug`` will set the loglevel
- to ``logging.DEBUG``.
+ This has the effect of setting the Python logging level for the namespace
+ of this element's class and object reference. A value of boolean ``True``
+ indicates that the loglevel ``logging.INFO`` will be set for the logger,
+ whereas the string value ``debug`` will set the loglevel to
+ ``logging.DEBUG``.
"""
def __get__(self, instance, owner):
"""Defines the [sqlalchemy.orm.mapper#Mapper] class, the central configurational
unit which associates a class with a database table.
-This is a semi-private module; the main configurational API of the ORM is
+This is a semi-private module; the main configurational API of the ORM is
avaiable in [sqlalchemy.orm#].
"""
-import weakref, warnings
+import weakref
from itertools import chain
from sqlalchemy import sql, util, exceptions, logging
from sqlalchemy.sql import expression, visitors, operators, util as sqlutil
eager_defaults=False):
"""Construct a new mapper.
- Mappers are normally constructed via the [sqlalchemy.orm#mapper()]
+ Mappers are normally constructed via the [sqlalchemy.orm#mapper()]
function. See for details.
"""
self._eager_loaders = util.Set()
self._row_translators = {}
self._dependency_processors = []
-
+
# our 'polymorphic identity', a string name that when located in a result set row
# indicates this Mapper should be used to construct the object instance for that row.
self.polymorphic_identity = polymorphic_identity
self.polymorphic_fetch = (self.select_table is None) and 'select' or 'union'
else:
self.polymorphic_fetch = polymorphic_fetch
-
+
# a dictionary of 'polymorphic identity' names, associating those names with
# Mappers that will be used to construct object instances upon a select operation.
if _polymorphic_map is None:
self.__should_log_info = logging.is_info_enabled(self.logger)
self.__should_log_debug = logging.is_debug_enabled(self.logger)
-
+
self._compile_class()
self._compile_inheritance()
self._compile_extensions()
def get_property(self, key, resolve_synonyms=False, raiseerr=True):
"""return a MapperProperty associated with the given key."""
-
+
self.compile()
return self._get_property(key, resolve_synonyms=resolve_synonyms, raiseerr=raiseerr)
def _get_property(self, key, resolve_synonyms=False, raiseerr=True):
"""private in-compilation version of get_property()."""
-
+
prop = self.__props.get(key, None)
if resolve_synonyms:
while isinstance(prop, SynonymProperty):
if prop is None and raiseerr:
raise exceptions.InvalidRequestError("Mapper '%s' has no property '%s'" % (str(self), key))
return prop
-
+
def iterate_properties(self):
self.compile()
return self.__props.itervalues()
iterate_properties = property(iterate_properties, doc="returns an iterator of all MapperProperty objects.")
-
+
def properties(self):
raise NotImplementedError("Public collection of MapperProperty objects is provided by the get_property() and iterate_properties accessors.")
properties = property(properties)
-
+
compiled = property(lambda self:self.__props_init, doc="return True if this mapper is compiled")
-
+
def dispose(self):
# disaable any attribute-based compilation
self.__props_init = True
del self._class_state.mappers[self.entity_name]
if not self._class_state.mappers:
attributes.unregister_class(self.class_)
-
+
def compile(self):
"""Compile this mapper into its final internal format.
"""
# double-check inside mutex
if self.__props_init:
return self
-
+
# initialize properties on all mappers
for mapper in list(_mapper_registry):
if not mapper.__props_init:
for ext_obj in util.to_list(extension):
# local MapperExtensions have already instrumented the class
extlist.add(ext_obj)
-
+
if self.inherits is not None:
for ext in self.inherits.extension:
if ext not in extlist:
if ext not in extlist:
extlist.add(ext)
ext.instrument_class(self, self.class_)
-
+
self.extension = ExtensionCarrier()
for ext in extlist:
self.extension.append(ext)
-
+
def _compile_inheritance(self):
"""Determine if this Mapper inherits from another mapper, and
if so calculates the mapped_table for this Mapper taking the
self._identity_class = self.inherits._identity_class
else:
self._identity_class = self.class_
-
+
if self.version_id_col is None:
self.version_id_col = self.inherits.version_id_col
-
+
if self.order_by is False:
self.order_by = self.inherits.order_by
self.polymorphic_map = self.inherits.polymorphic_map
raise exceptions.ArgumentError("Mapper '%s' specifies a polymorphic_identity of '%s', but no mapper in it's hierarchy specifies the 'polymorphic_on' column argument" % (str(self), self.polymorphic_identity))
self.polymorphic_map[self.polymorphic_identity] = self
self._identity_class = self.class_
-
+
if self.mapped_table is None:
raise exceptions.ArgumentError("Mapper '%s' does not have a mapped_table specified. (Are you using the return value of table.create()? It no longer has a return value.)" % str(self))
self._pks_by_table = {}
self._cols_by_table = {}
-
+
all_cols = util.Set(chain(*[c2 for c2 in [col.proxy_set for col in [c for c in self._columntoproperty]]]))
pk_cols = util.Set([c for c in all_cols if c.primary_key])
-
+
for t in util.Set(self.tables + [self.mapped_table]):
self._all_tables.add(t)
if t.primary_key and pk_cols.issuperset(t.primary_key):
# ordering is important since it determines the ordering of mapper.primary_key (and therefore query.get())
self._pks_by_table[t] = util.OrderedSet(t.primary_key).intersection(pk_cols)
self._cols_by_table[t] = util.OrderedSet(t.c).intersection(all_cols)
-
+
if self.primary_key_argument:
for k in self.primary_key_argument:
if k.table not in self._pks_by_table:
self._pks_by_table[k.table] = util.OrderedSet()
self._pks_by_table[k.table].add(k)
-
+
if self.mapped_table not in self._pks_by_table or len(self._pks_by_table[self.mapped_table]) == 0:
raise exceptions.ArgumentError("Could not assemble any primary key columns for mapped table '%s'" % (self.mapped_table.name))
# multiple columns that all reference a common parent column. it will also resolve the column
# against the "mapped_table" of this mapper.
self._equivalent_columns = self._get_equivalent_columns()
-
+
primary_key = expression.ColumnSet()
for col in (self.primary_key_argument or self._pks_by_table[self.mapped_table]):
else:
break
primary_key.add(c)
-
+
if len(primary_key) == 0:
raise exceptions.ArgumentError("Could not assemble any primary key columns for mapped table '%s'" % (self.mapped_table.name))
one another either by an established foreign key relationship
or by a joined-table inheritance join.
- This is used to determine the minimal set of primary key
- columns for the mapper, as well as when relating
+ This is used to determine the minimal set of primary key
+ columns for the mapper, as well as when relating
columns to those of a polymorphic selectable (i.e. a UNION of
several mapped tables), as that selectable usually only contains
one column in its columns clause out of a group of several which
to lists of equivalent columns, i.e.
{
- tablea.col1:
+ tablea.col1:
set([tableb.col1, tablec.col1]),
tablea.col2:
set([tabled.col2])
}
-
+
"""
result = {}
result[fk.column] = util.Set()
result[fk.column].add(equiv)
equivs(fk.column, recursive, col)
-
+
for column in (self.primary_key_argument or self._pks_by_table[self.mapped_table]):
for col in column.proxy_set:
if not col.foreign_keys:
result[col].add(col)
else:
equivs(col, util.Set(), col)
-
+
return result
-
+
class _CompileOnAttr(PropComparator):
"""placeholder class attribute which fires mapper compilation on access"""
self.class_ = class_
self.key = key
self.existing_prop = getattr(class_, key, None)
-
+
def __getattribute__(self, key):
cls = object.__getattribute__(self, 'class_')
clskey = object.__getattribute__(self, 'key')
return object.__getattribute__(self, key)
class_mapper(cls)
-
+
if cls.__dict__.get(clskey) is self:
- # FIXME: there should not be any scenarios where
- # a mapper compile leaves this CompileOnAttr in
- # place.
- warnings.warn(RuntimeWarning("Attribute '%s' on class '%s' was not replaced during mapper compilation operation" % (clskey, cls.__name__)))
+ # FIXME: there should not be any scenarios where
+ # a mapper compile leaves this CompileOnAttr in
+ # place.
+ util.warn(
+ ("Attribute '%s' on class '%s' was not replaced during "
+ "mapper compilation operation") % (clskey, cls.__name__))
# clean us up explicitly
delattr(cls, clskey)
-
+
return getattr(getattr(cls, clskey), key)
-
+
def _compile_properties(self):
# object attribute names mapped to MapperProperty objects
column.key not in self.include_properties):
self.__log("not including property %s" % (column.key))
continue
-
+
if (self.exclude_properties is not None and
column.key in self.exclude_properties):
self.__log("excluding property %s" % (column.key))
continue
column_key = (self.column_prefix or '') + column.key
-
+
self._compile_property(column_key, column, init=False, setparent=True)
def _adapt_inherited_property(self, key, prop):
column = columns[0]
if not expression.is_column(column):
raise exceptions.ArgumentError("%s=%r is not an instance of MapperProperty or Column" % (key, prop))
-
+
prop = self.__props.get(key, None)
if isinstance(prop, ColumnProperty):
- # TODO: the "property already exists" case is still not well defined here.
+ # TODO: the "property already exists" case is still not well defined here.
# assuming single-column, etc.
-
+
if prop.parent is not self:
# existing ColumnProperty from an inheriting mapper.
# make a copy and append our column to it
if prop.map_column:
if not key in self.mapped_table.c:
raise exceptions.ArgumentError("Can't compile synonym '%s': no column on table '%s' named '%s'" % (prop.name, self.mapped_table.description, key))
- self._compile_property(prop.name, ColumnProperty(self.mapped_table.c[key]), init=init, setparent=setparent)
+ self._compile_property(prop.name, ColumnProperty(self.mapped_table.c[key]), init=init, setparent=setparent)
self.__props[key] = prop
if setparent:
self.compile()
if 'init_instance' in self.extension.methods:
self.extension.init_instance(self, class_, oldinit, instance, args, kwargs)
-
+
def on_exception(class_, oldinit, instance, args, kwargs):
util.warn_exception(self.extension.init_failed, self, class_, oldinit, instance, args, kwargs)
attributes.register_class(self.class_, extra_init=extra_init, on_exception=on_exception, deferred_scalar_loader=_load_scalar_attributes)
-
+
self._class_state = self.class_._class_state
_mapper_registry[self] = True
Raise ``InvalidRequestError`` if a session cannot be retrieved
from the extension chain.
"""
-
+
if 'get_session' in self.extension.methods:
s = self.extension.get_session()
if s is not EXT_CONTINUE:
return s
raise exceptions.InvalidRequestError("No contextual Session is established.")
-
+
def instances(self, cursor, session, *mappers, **kwargs):
"""Return a list of mapped instances corresponding to the rows
in a given ResultProxy.
-
+
DEPRECATED.
"""
raise exceptions.UnmappedColumnError("Column '%s.%s' is not available, due to conflicting property '%s':%s" % (column.table.name, column.name, column.key, repr(prop)))
else:
raise exceptions.UnmappedColumnError("No column %s.%s is configured on mapper %s..." % (column.table.name, column.name, str(self)))
-
+
def _get_state_attr_by_column(self, state, column):
return self._get_col_to_prop(column).getattr(state, column)
-
+
def _set_state_attr_by_column(self, state, column, value):
return self._get_col_to_prop(column).setattr(state, value, column)
-
+
def _get_attr_by_column(self, obj, column):
return self._get_col_to_prop(column).getattr(obj._state, column)
def _get_committed_attr_by_column(self, obj, column):
return self._get_col_to_prop(column).getcommitted(obj._state, column)
-
+
def _set_attr_by_column(self, obj, column, value):
self._get_col_to_prop(column).setattr(obj._state, column, value)
self._save_obj([state], uowtransaction, postupdate=postupdate, post_update_cols=post_update_cols, single=True)
return
- # if session has a connection callable,
+ # if session has a connection callable,
# organize individual states with the connection to use for insert/update
if 'connection_callable' in uowtransaction.mapper_flush_opts:
connection_callable = uowtransaction.mapper_flush_opts['connection_callable']
else:
connection = uowtransaction.transaction.connection(self)
tups = [(state, connection, _state_has_identity(state)) for state in states]
-
+
if not postupdate:
# call before_XXX extensions
for state, connection, has_identity in tups:
if col in pks:
params[col._label] = mapper._get_state_attr_by_column(state, col)
continue
-
+
prop = mapper._columntoproperty[col]
(added, unchanged, deleted) = attributes.get_history(state, prop.key, passive=True)
if added:
if update:
mapper = table_to_mapper[table]
clause = sql.and_()
-
+
for col in mapper._pks_by_table[table]:
clause.clauses.append(col == sql.bindparam(col._label, type_=col.type))
-
+
if mapper.version_id_col is not None and table.c.contains_column(mapper.version_id_col):
clause.clauses.append(mapper.version_id_col == sql.bindparam(mapper.version_id_col._label, type_=col.type))
-
+
statement = table.update(clause)
pks = mapper._pks_by_table[table]
def comparator(a, b):
return x
return 0
update.sort(comparator)
-
+
rows = 0
for rec in update:
(state, params, mapper, connection, value_params) = rec
else:
if 'after_update' in mapper.extension.methods:
mapper.extension.after_update(mapper, connection, state.obj())
-
+
def _postfetch(self, uowtransaction, connection, table, state, resultproxy, params, value_params):
"""After an ``INSERT`` or ``UPDATE``, assemble newly generated
values on an instance. For columns which are marked as being generated
- on the database side, set up a group-based "deferred" loader
+ on the database side, set up a group-based "deferred" loader
which will populate those attributes in one query when next accessed.
"""
- postfetch_cols = util.Set(resultproxy.postfetch_cols()).union(util.Set(value_params.keys()))
+ postfetch_cols = util.Set(resultproxy.postfetch_cols()).union(util.Set(value_params.keys()))
deferred_props = []
for c in self._cols_by_table[table]:
deferred_props.append(prop.key)
elif not c.primary_key and c.key in params and self._get_state_attr_by_column(state, c) != params[c.key]:
self._set_state_attr_by_column(state, c, params[c.key])
-
+
if deferred_props:
if self.eager_defaults:
_instance_key = self._identity_key_from_state(state)
prop.register_dependencies(uowcommit)
for dep in self._dependency_processors:
dep.register_dependencies(uowcommit)
-
+
def cascade_iterator(self, type, state, recursive=None, halt_on=None):
- """Iterate each element and its mapper in an object graph,
+ """Iterate each element and its mapper in an object graph,
for all relations that meet the given cascade rule.
type
recursive
Used by the function for internal context during recursive
calls, leave as None.
-
+
the return value are object instances; this provides a strong
reference so that they don't fall out of scope immediately.
"""
def _instance(self, context, row, result=None, skip_polymorphic=False, extension=None, only_load_props=None, refresh_instance=None):
if not extension:
extension = self.extension
-
+
if 'translate_row' in extension.methods:
ret = extension.translate_row(self, context, row)
if ret is not EXT_CONTINUE:
context.attributes[('polymorphic_fetch', mapper)] = (self, [t for t in mapper.tables if t not in self.tables])
row = self.translate_row(mapper, row)
return mapper._instance(context, row, result=result, skip_polymorphic=True)
-
- # determine identity key
+
+ # determine identity key
if refresh_instance:
identitykey = refresh_instance.dict['_instance_key']
else:
identitykey = self.identity_key_from_row(row)
-
+
session_identity_map = context.session.identity_map
if identitykey in session_identity_map:
isnew = state.runid != context.runid
currentload = not isnew
-
+
if not currentload and context.version_check and self.version_id_col and self._get_attr_by_column(instance, self.version_id_col) != row[self.version_id_col]:
raise exceptions.ConcurrentModificationError("Instance '%s' version of %s does not match %s" % (instance, self._get_attr_by_column(instance, self.version_id_col), row[self.version_id_col]))
elif refresh_instance:
else:
if self.__should_log_debug:
self.__log_debug("_instance(): identity key %s not in session" % str(identitykey))
-
+
if self.allow_null_pks:
for x in identitykey[1]:
if x is not None:
attributes.manage(instance)
else:
instance = attributes.new_instance(self.class_)
-
+
if self.__should_log_debug:
self.__log_debug("_instance(): created new instance %s identity %s" % (instance_str(instance), str(identitykey)))
-
- state = instance._state
+
+ state = instance._state
instance._entity_name = self.entity_name
instance._instance_key = identitykey
instance._sa_session_id = context.session.hash_key
session_identity_map[identitykey] = instance
-
+
if currentload or context.populate_existing or self.always_refresh:
if isnew:
state.runid = context.runid
context.progress.add(state)
-
+
if 'populate_instance' not in extension.methods or extension.populate_instance(self, context, row, instance, only_load_props=only_load_props, instancekey=identitykey, isnew=isnew) is EXT_CONTINUE:
self.populate_instance(context, instance, row, only_load_props=only_load_props, instancekey=identitykey, isnew=isnew)
# if 'populate_instance' not in extension.methods or extension.populate_instance(self, context, row, instance, only_load_props=attrs, instancekey=identitykey, isnew=isnew) is EXT_CONTINUE:
# self.populate_instance(context, instance, row, only_load_props=attrs, instancekey=identitykey, isnew=isnew)
# context.partials.add((state, attrs)) <-- allow query.instances to commit the subset of attrs
-
+
if result is not None and ('append_result' not in extension.methods or extension.append_result(self, context, row, instance, result, instancekey=identitykey, isnew=isnew) is EXT_CONTINUE):
result.append(instance)
-
+
return instance
def translate_row(self, tomapper, row):
This can be used in conjunction with populate_instance to
populate an instance using an alternate mapper.
"""
-
+
if tomapper in self._row_translators:
# row translators are cached based on target mapper
return self._row_translators[tomapper](row)
snapshot = selectcontext.path + (self,)
# retrieve a set of "row population" functions derived from the MapperProperties attached
- # to this Mapper. These are keyed in the select context based primarily off the
+ # to this Mapper. These are keyed in the select context based primarily off the
# "snapshot" of the stack, which represents a path from the lead mapper in the query to this one,
# including relation() names. the key also includes "self", and allows us to distinguish between
# other mappers within our inheritance hierarchy
existing_populators.append((prop.key, existingpop))
if post_proc is not None:
post_processors.append(post_proc)
-
+
# install a post processor for immediate post-load of joined-table inheriting mappers
poly_select_loader = self._get_poly_select_loader(selectcontext, row)
if poly_select_loader is not None:
post_processors.append(poly_select_loader)
-
+
selectcontext.attributes[('populators', self, snapshot, ispostselect)] = (new_populators, existing_populators)
selectcontext.attributes[('post_processors', self, ispostselect)] = post_processors
if only_load_props:
populators = [p for p in populators if p[0] in only_load_props]
-
+
for (key, populator) in populators:
selectcontext.exec_with_path(self, key, populator, instance, row, ispostselect=ispostselect, isnew=isnew, **flags)
-
+
if self.non_primary:
selectcontext.attributes[('populating_mapper', instance._state)] = self
-
+
def _post_instance(self, selectcontext, state):
post_processors = selectcontext.attributes[('post_processors', self, None)]
for p in post_processors:
def _get_poly_select_loader(self, selectcontext, row):
"""set up attribute loaders for 'select' and 'deferred' polymorphic loading.
-
+
this loading uses a second SELECT statement to load additional tables,
either immediately after loading the main table or via a deferred attribute trigger.
"""
-
+
(hosted_mapper, needs_tables) = selectcontext.attributes.get(('polymorphic_fetch', self), (None, None))
-
+
if hosted_mapper is None or not needs_tables:
return
-
+
cond, param_names = self._deferred_inheritance_condition(hosted_mapper, needs_tables)
statement = sql.select(needs_tables, cond, use_labels=True)
-
+
if hosted_mapper.polymorphic_fetch == 'select':
def post_execute(instance, **flags):
if self.__should_log_debug:
return post_execute
elif hosted_mapper.polymorphic_fetch == 'deferred':
from sqlalchemy.orm.strategies import DeferredColumnLoader
-
+
def post_execute(instance, **flags):
def create_statement(instance):
params = {}
# use the "committed" (database) version to get query column values
params[bind] = self._get_committed_attr_by_column(instance, c)
return (statement, params)
-
+
props = [prop for prop in [self._get_col_to_prop(col) for col in statement.inner_columns] if prop.key not in instance.__dict__]
keys = [p.key for p in props]
for prop in props:
def _deferred_inheritance_condition(self, base_mapper, needs_tables):
base_mapper = base_mapper.primary_mapper()
-
+
def visit_binary(binary):
leftcol = binary.left
rightcol = binary.right
allconds.append(visitors.traverse(mapper.inherit_condition, clone=True, visit_binary=visit_binary))
return sql.and_(*allconds), param_names
-
+
Mapper.logger = logging.class_logger(Mapper)
def _state_has_identity(state):
return '_instance_key' in state.dict
-
+
def has_mapper(object):
"""Return True if the given object has had a mapper association
set up, either through loading, or via insertion in a session.
session = mapper.get_session()
except exceptions.InvalidRequestError:
raise exceptions.InvalidRequestError("Instance %s is not bound to a Session, and no contextual session is established; attribute refresh operation cannot proceed" % (instance.__class__))
-
+
if session.query(mapper)._get(instance._instance_key, refresh_instance=instance._state, only_load_props=attribute_names) is None:
raise exceptions.InvalidRequestError("Could not refresh instance '%s'" % instance_str(instance))
def object_mapper(object, entity_name=None, raiseerror=True):
"""Given an object, return the primary Mapper associated with the object instance.
-
+
object
The object instance.
-
+
entity_name
- Entity name of the mapper to retrieve, if the given instance is
- transient. Otherwise uses the entity name already associated
+ Entity name of the mapper to retrieve, if the given instance is
+ transient. Otherwise uses the entity name already associated
with the instance.
-
+
raiseerror
Defaults to True: raise an ``InvalidRequestError`` if no mapper can
be located. If False, return None.
-
+
"""
try:
def class_mapper(class_, entity_name=None, compile=True):
"""Given a class and optional entity_name, return the primary Mapper associated with the key.
-
+
If no mapper can be located, raises ``InvalidRequestError``.
"""
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
-"""MapperProperty implementations.
+"""MapperProperty implementations.
This is a private module which defines the behavior of
invidual ORM-mapped attributes.
from sqlalchemy.orm.util import CascadeOptions
from sqlalchemy.orm.interfaces import StrategizedProperty, PropComparator, MapperProperty
from sqlalchemy.exceptions import ArgumentError
-import warnings
-__all__ = ['ColumnProperty', 'CompositeProperty', 'SynonymProperty', 'PropertyLoader', 'BackRef']
+__all__ = ('ColumnProperty', 'CompositeProperty', 'SynonymProperty',
+ 'PropertyLoader', 'BackRef')
+
class ColumnProperty(StrategizedProperty):
"""Describes an object attribute that corresponds to a table column."""
for col in columns:
if not isinstance(col, ColumnElement):
raise ArgumentError('column_property() must be given a ColumnElement as its argument. Try .label() or .as_scalar() for Selectables to fix this.')
-
+
def create_strategy(self):
if self.deferred:
return strategies.DeferredColumnLoader(self)
def do_init(self):
super(ColumnProperty, self).do_init()
if len(self.columns) > 1 and self.parent.primary_key.issuperset(self.columns):
- warnings.warn(RuntimeWarning("On mapper %s, primary key column '%s' is being combined with distinct primary key column '%s' in attribute '%s'. Use explicit properties to give each column its own mapped attribute name." % (str(self.parent), str(self.columns[1]), str(self.columns[0]), self.key)))
-
+ util.warn(
+ ("On mapper %s, primary key column '%s' is being combined "
+ "with distinct primary key column '%s' in attribute '%s'. "
+ "Use explicit properties to give each column its own mapped "
+ "attribute name.") % (str(self.parent), str(self.columns[1]),
+ str(self.columns[0]), self.key))
+
def copy(self):
return ColumnProperty(deferred=self.deferred, group=self.group, *self.columns)
-
+
def getattr(self, state, column):
return getattr(state.class_, self.key).impl.get(state)
def setattr(self, state, value, column):
getattr(state.class_, self.key).impl.set(state, value, None)
-
+
def merge(self, session, source, dest, dont_load, _recursive):
value = attributes.get_as_list(source._state, self.key, passive=True)
if value:
else:
# TODO: lazy callable should merge to the new instance
dest._state.expire_attributes([self.key])
-
+
def get_col_value(self, column, value):
return value
class ColumnComparator(PropComparator):
def clause_element(self):
return self.prop.columns[0]
-
+
def operate(self, op, *other):
return op(self.prop.columns[0], *other)
class CompositeProperty(ColumnProperty):
"""subclasses ColumnProperty to provide composite type support."""
-
+
def __init__(self, class_, *columns, **kwargs):
super(CompositeProperty, self).__init__(*columns, **kwargs)
self.composite_class = class_
def do_init(self):
super(ColumnProperty, self).do_init()
# TODO: similar PK check as ColumnProperty does ?
-
+
def copy(self):
return CompositeProperty(deferred=self.deferred, group=self.group, composite_class=self.composite_class, *self.columns)
if obj is None:
obj = self.composite_class(*[None for c in self.columns])
getattr(state.class_, self.key).impl.set(state, obj, None)
-
+
for a, b in zip(self.columns, value.__composite_values__()):
if a is column:
setattr(obj, b, value)
-
+
def get_col_value(self, column, value):
for a, b in zip(self.columns, value.__composite_values__()):
if a is column:
self.name = name
self.map_column=map_column
self.instrument = None
-
+
def setup(self, querycontext, **kwargs):
pass
return s
return getattr(obj, self.name)
self.instrument = SynonymProp()
-
+
sessionlib.register_attribute(class_, self.key, uselist=False, proxy_property=self.instrument, useobject=False, comparator=comparator)
def merge(self, session, source, dest, _recursive):
self.cascade = CascadeOptions("all, delete-orphan")
else:
self.cascade = CascadeOptions("save-update, merge")
-
+
if self.passive_deletes == 'all' and ("delete" in self.cascade or "delete-orphan" in self.cascade):
raise exceptions.ArgumentError("Can't set passive_deletes='all' in conjunction with 'delete' or 'delete-orphan' cascade")
sql.exists([1], j & sql.and_(*[x==y for (x, y) in zip(self.prop.mapper.primary_key, self.prop.mapper.primary_key_from_instance(o))]))
)
return sql.and_(*clauses)
- else:
+ else:
return self.prop._optimized_compare(other)
-
+
def any(self, criterion=None, **kwargs):
if not self.prop.uselist:
raise exceptions.InvalidRequestError("'any()' not implemented for scalar attributes. Use has().")
else:
criterion = criterion & crit
return sql.exists([1], j & criterion)
-
+
def has(self, criterion=None, **kwargs):
if self.prop.uselist:
raise exceptions.InvalidRequestError("'has()' not implemented for collections. Use any().")
else:
criterion = criterion & crit
return sql.exists([1], j & criterion)
-
+
def contains(self, other):
if not self.prop.uselist:
raise exceptions.InvalidRequestError("'contains' not implemented for scalar attributes. Use ==")
def __ne__(self, other):
if self.prop.uselist and not hasattr(other, '__iter__'):
raise exceptions.InvalidRequestError("Can only compare a collection to an iterable object")
-
+
j = self.prop.primaryjoin
if self.prop.secondaryjoin:
j = j & self.prop.secondaryjoin
return ~sql.exists([1], j & sql.and_(*[x==y for (x, y) in zip(self.prop.mapper.primary_key, self.prop.mapper.primary_key_from_instance(other))]))
-
+
def compare(self, op, value, value_is_parent=False):
if op == operators.eq:
if value is None:
return self._optimized_compare(value, value_is_parent=value_is_parent)
else:
return op(self.comparator, value)
-
+
def _optimized_compare(self, value, value_is_parent=False):
return self._get_strategy(strategies.LazyLoader).lazy_clause(value, reverse_direction=not value_is_parent)
-
+
def private(self):
return self.cascade.delete_orphan
private = property(private)
recursive.add(c)
# cascade using the mapper local to this object, so that its individual properties are located
- instance_mapper = object_mapper(c, entity_name=mapper.entity_name)
+ instance_mapper = object_mapper(c, entity_name=mapper.entity_name)
yield (c, instance_mapper)
for (c2, m) in instance_mapper.cascade_iterator(type, c._state, recursive):
yield (c2, m)
if not self.parent.concrete:
for inheriting in self.parent.iterate_to_root():
if inheriting is not self.parent and inheriting._get_property(self.key, raiseerr=False):
- warnings.warn(RuntimeWarning("Warning: relation '%s' on mapper '%s' supercedes the same relation on inherited mapper '%s'; this can cause dependency issues during flush" % (self.key, self.parent, inheriting)))
+ util.warn(
+ ("Warning: relation '%s' on mapper '%s' supercedes "
+ "the same relation on inherited mapper '%s'; this "
+ "can cause dependency issues during flush") %
+ (self.key, self.parent, inheriting))
if self.association is not None:
if isinstance(self.association, type):
if self.secondaryjoin is not None and self.secondary is None:
raise exceptions.ArgumentError("Property '" + self.key + "' specified with secondary join condition but no secondary argument")
# if join conditions were not specified, figure them out based on foreign keys
-
+
def _search_for_join(mapper, table):
"""find a join between the given mapper's mapped table and the given table.
- will try the mapper's local table first for more specificity, then if not
+ will try the mapper's local table first for more specificity, then if not
found will try the more general mapped table, which in the case of inheritance
is a join."""
try:
return sql.join(mapper.local_table, table)
except exceptions.ArgumentError, e:
return sql.join(mapper.mapped_table, table)
-
+
try:
if self.secondary is not None:
if self.secondaryjoin is None:
# or clauses related to those external tables dealt with. see orm.relationships.ViewOnlyTest
if not col_is_part_of_mappings(binary.left) or not col_is_part_of_mappings(binary.right):
return
-
+
for f in binary.left.foreign_keys:
if f.references(binary.right.table):
self.foreign_keys.add(binary.left)
for c in list(self.remote_side):
if self.secondary and self.secondary.columns.contains_column(c):
continue
- for equiv in [c] + (c in target_equivalents and list(target_equivalents[c]) or []):
+ for equiv in [c] + (c in target_equivalents and list(target_equivalents[c]) or []):
corr = self.mapper.select_table.corresponding_column(equiv)
if corr:
self.remote_side.add(corr)
def get_join(self, parent, primary=True, secondary=True, polymorphic_parent=True):
"""return a join condition from the given parent mapper to this PropertyLoader's mapper.
-
+
The resulting ClauseElement object is cached and should not be modified directly.
-
+
parent
- a mapper which has a relation() to this PropertyLoader. A PropertyLoader can
+ a mapper which has a relation() to this PropertyLoader. A PropertyLoader can
have multiple "parents" when its actual parent mapper has inheriting mappers.
-
+
primary
include the primary join condition in the resulting join.
-
+
secondary
include the secondary join condition in the resulting join. If both primary
and secondary are returned, they are joined via AND.
-
+
polymorphic_parent
if True, use the parent's 'select_table' instead of its 'mapped_table' to produce the join.
"""
-
+
try:
return self._parent_join_cache[(parent, primary, secondary, polymorphic_parent)]
except KeyError:
class BackRef(object):
"""Attached to a PropertyLoader to indicate a complementary reverse relationship.
-
+
Can optionally create the complementing PropertyLoader if one does not exist already."""
def __init__(self, key, _prop=None, **kwargs):
def compile(self, prop):
if self.prop:
return
-
+
self.prop = prop
-
+
mapper = prop.mapper.primary_mapper()
if mapper._get_property(self.key, raiseerr=False) is None:
pj = self.kwargs.pop('primaryjoin', None)
parent = prop.parent.primary_mapper()
self.kwargs.setdefault('viewonly', prop.viewonly)
self.kwargs.setdefault('post_update', prop.post_update)
-
+
relation = PropertyLoader(parent, prop.secondary, pj, sj,
- backref=BackRef(prop.key, _prop=prop),
+ backref=BackRef(prop.key, _prop=prop),
is_backref=True,
**self.kwargs)
-
+
mapper._compile_property(self.key, relation);
prop.reverse_property = mapper._get_property(self.key)
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
-"""Defines the [sqlalchemy.orm.query#Query] class, the central
-construct used by the ORM to construct database queries.
-
-The ``Query`` class should not be confused with the [sqlalchemy.sql.expression#Select]
-class, which defines database SELECT operations at the SQL (non-ORM) level.
-``Query`` differs from ``Select`` in that it returns ORM-mapped objects and interacts
-with an ORM session, whereas the ``Select`` construct interacts directly with the database
-to return iterable result sets.
-"""
+"""The Query class and support.
+
+Defines the [sqlalchemy.orm.query#Query] class, the central construct used by
+the ORM to construct database queries.
+The ``Query`` class should not be confused with the
+[sqlalchemy.sql.expression#Select] class, which defines database SELECT
+operations at the SQL (non-ORM) level. ``Query`` differs from ``Select`` in
+that it returns ORM-mapped objects and interacts with an ORM session, whereas
+the ``Select`` construct interacts directly with the database to return
+iterable result sets.
+"""
+from itertools import chain
from sqlalchemy import sql, util, exceptions, logging
from sqlalchemy.sql import util as sql_util
from sqlalchemy.sql import expression, visitors, operators
from sqlalchemy.orm import mapper, object_mapper
from sqlalchemy.orm.mapper import _state_mapper
from sqlalchemy.orm import util as mapperutil
-from itertools import chain
-import warnings
__all__ = ['Query', 'QueryContext']
class Query(object):
"""Encapsulates the object-fetching operations provided by Mappers."""
-
+
def __init__(self, class_or_mapper, session=None, entity_name=None):
if isinstance(class_or_mapper, type):
self.mapper = mapper.class_mapper(class_or_mapper, entity_name=entity_name)
else:
self.mapper = class_or_mapper.compile()
self.select_mapper = self.mapper.get_select_mapper().compile()
-
+
self._session = session
-
+
self._with_options = []
self._lockmode = None
self._extension = self.mapper.extension.copy()
self._primary_adapter=None
self._only_load_props = None
self._refresh_instance = None
-
+
def _no_criterion(self, meth):
q = self._clone()
-
+
if q._criterion or q._statement or q._from_obj is not self.table:
- warnings.warn(RuntimeWarning("Query.%s() being called on a Query with existing criterion; criterion is being ignored." % meth))
-
+ util.warn(
+ ("Query.%s() being called on a Query with existing criterion; "
+ "criterion is being ignored.") % meth)
+
q._from_obj = self.table
q._alias_ids = {}
q._joinpoint = self.mapper
q._statement = q._aliases = q._criterion = None
q._order_by = q._group_by = q._distinct = False
return q
-
+
def _no_statement(self, meth):
q = self._clone()
if q._statement:
- raise exceptions.InvalidRequestError("Query.%s() being called on a Query with an existing full statement - can't apply criterion." % meth)
+ raise exceptions.InvalidRequestError(
+ ("Query.%s() being called on a Query with an existing full "
+ "statement - can't apply criterion.") % meth)
return q
-
+
def _clone(self):
q = Query.__new__(Query)
q.__dict__ = self.__dict__.copy()
return q
-
+
def _get_session(self):
if self._session is None:
return self.mapper.get_session()
q = self._clone()
q._current_path = path
return q
-
+
def yield_per(self, count):
- """yield only ``count`` rows at a time.
-
- WARNING: use this method with caution; if the same instance
- is present in more than one batch of rows, end-user changes
- to attributes will be overwritten.
- In particular, it's usually impossible to use this setting with
- eagerly loaded collections (i.e. any lazy=False) since those
- collections will be cleared for a new load when encountered
- in a subsequent result batch.
+ """Yield only ``count`` rows at a time.
+
+ WARNING: use this method with caution; if the same instance is present
+ in more than one batch of rows, end-user changes to attributes will be
+ overwritten.
+
+ In particular, it's usually impossible to use this setting with
+ eagerly loaded collections (i.e. any lazy=False) since those
+ collections will be cleared for a new load when encountered in a
+ subsequent result batch.
"""
+
q = self._clone()
q._yield_per = count
return q
-
+
def get(self, ident, **kwargs):
- """Return an instance of the object based on the given
- identifier, or None if not found.
+ """Return an instance of the object based on the given identifier, or None if not found.
- The `ident` argument is a scalar or tuple of primary key
- column values in the order of the table def's primary key
- columns.
+ The `ident` argument is a scalar or tuple of primary key column values
+ in the order of the table def's primary key columns.
"""
ret = self._extension.get(self, ident, **kwargs)
return ret
# convert composite types to individual args
- # TODO: account for the order of columns in the
+ # TODO: account for the order of columns in the
# ColumnProperty it corresponds to
if hasattr(ident, '__composite_values__'):
ident = ident.__composite_values__()
return self._get(key, ident, **kwargs)
def load(self, ident, raiseerr=True, **kwargs):
- """Return an instance of the object based on the given
- identifier.
-
- If not found, raises an exception. The method will **remove
- all pending changes** to the object already existing in the
- Session. The `ident` argument is a scalar or tuple of primary
- key column values in the order of the table def's primary key
- columns.
+ """Return an instance of the object based on the given identifier.
+
+ If not found, raises an exception. The method will **remove all
+ pending changes** to the object already existing in the Session. The
+ `ident` argument is a scalar or tuple of primary key column values in
+ the order of the table def's primary key columns.
"""
+
ret = self._extension.load(self, ident, **kwargs)
if ret is not mapper.EXT_CONTINUE:
return ret
if instance is None and raiseerr:
raise exceptions.InvalidRequestError("No instance found for identity %s" % repr(ident))
return instance
-
+
def query_from_parent(cls, instance, property, **kwargs):
- """return a newly constructed Query object, with criterion corresponding to
- a relationship to the given parent instance.
+ """Return a new Query with criterion corresponding to a parent instance.
- instance
- a persistent or detached instance which is related to class represented
- by this query.
+ Return a newly constructed Query object, with criterion corresponding
+ to a relationship to the given parent instance.
- property
- string name of the property which relates this query's class to the
- instance.
-
- \**kwargs
- all extra keyword arguments are propigated to the constructor of
- Query.
-
+ instance
+ a persistent or detached instance which is related to class
+ represented by this query.
+
+ property
+ string name of the property which relates this query's class to the
+ instance.
+
+ \**kwargs
+ all extra keyword arguments are propigated to the constructor of
+ Query.
"""
-
+
mapper = object_mapper(instance)
prop = mapper.get_property(property, resolve_synonyms=True)
target = prop.mapper
criterion = prop.compare(operators.eq, instance, value_is_parent=True)
return Query(target, **kwargs).filter(criterion)
query_from_parent = classmethod(query_from_parent)
-
+
def autoflush(self, setting):
q = self._clone()
q._autoflush = setting
return q
-
+
def populate_existing(self):
- """return a Query that will refresh all instances loaded.
-
- this includes all entities accessed from the database, including
+ """Return a Query that will refresh all instances loaded.
+
+ This includes all entities accessed from the database, including
secondary entities, eagerly-loaded collection items.
-
- All changes present on entities which are already present in the session will
- be reset and the entities will all be marked "clean".
-
+
+ All changes present on entities which are already present in the
+ session will be reset and the entities will all be marked "clean".
+
This is essentially the en-masse version of load().
"""
-
+
q = self._clone()
q._populate_existing = True
return q
-
+
def with_parent(self, instance, property=None):
"""add a join criterion corresponding to a relationship to the given parent instance.
by this query.
property
- string name of the property which relates this query's class to the
+ string name of the property which relates this query's class to the
instance. if None, the method will attempt to find a suitable property.
currently, this method only works with immediate parent relationships, but in the
- future may be enhanced to work across a chain of parent mappers.
+ future may be enhanced to work across a chain of parent mappers.
"""
from sqlalchemy.orm import properties
def add_entity(self, entity, alias=None, id=None):
"""add a mapped entity to the list of result columns to be returned.
-
+
This will have the effect of all result-returning methods returning a tuple
- of results, the first element being an instance of the primary class for this
+ of results, the first element being an instance of the primary class for this
Query, and subsequent elements matching columns or entities which were
specified via add_column or add_entity.
-
+
When adding entities to the result, its generally desireable to add
limiting criterion to the query which can associate the primary entity
of this Query along with the additional entities. The Query selects
from all tables with no joining criterion by default.
-
+
entity
a class or mapper which will be added to the results.
-
+
alias
a sqlalchemy.sql.Alias object which will be used to select rows. this
will match the usage of the given Alias in filter(), order_by(), etc. expressions
-
+
id
- a string ID matching that given to query.join() or query.outerjoin(); rows will be
+ a string ID matching that given to query.join() or query.outerjoin(); rows will be
selected from the aliased join created via those methods.
"""
q = self._clone()
q._entities = q._entities + [(entity, alias, id)]
return q
-
+
def add_column(self, column, id=None):
- """add a SQL ColumnElement to the list of result columns to be returned.
-
- This will have the effect of all result-returning methods returning a tuple
- of results, the first element being an instance of the primary class for this
- Query, and subsequent elements matching columns or entities which were
- specified via add_column or add_entity.
+ """Add a SQL ColumnElement to the list of result columns to be returned.
+
+ This will have the effect of all result-returning methods returning a
+ tuple of results, the first element being an instance of the primary
+ class for this Query, and subsequent elements matching columns or
+ entities which were specified via add_column or add_entity.
When adding columns to the result, its generally desireable to add
limiting criterion to the query which can associate the primary entity
- of this Query along with the additional columns, if the column is based on a
- table or selectable that is not the primary mapped selectable. The Query selects
- from all tables with no joining criterion by default.
-
- column
- a string column name or sql.ColumnElement to be added to the results.
-
+ of this Query along with the additional columns, if the column is
+ based on a table or selectable that is not the primary mapped
+ selectable. The Query selects from all tables with no joining
+ criterion by default.
+
+ column
+ a string column name or sql.ColumnElement to be added to the results.
"""
-
+
q = self._clone()
# duck type to get a ClauseElement
if hasattr(column, 'clause_element'):
column = column.clause_element()
-
- # alias non-labeled column elements.
+
+ # alias non-labeled column elements.
if isinstance(column, sql.ColumnElement) and not hasattr(column, '_label'):
column = column.label(None)
-
+
q._entities = q._entities + [(column, None, id)]
return q
-
+
def options(self, *args):
"""Return a new Query object, applying the given list of
MapperOptions.
"""
-
+
return self._options(False, *args)
def _conditional_options(self, *args):
return self._options(True, *args)
-
+
def _options(self, conditional, *args):
q = self._clone()
# most MapperOptions write to the '_attributes' dictionary,
for opt in opts:
opt.process_query(q)
return q
-
+
def with_lockmode(self, mode):
"""Return a new Query object with the specified locking mode."""
q = self._clone()
def params(self, *args, **kwargs):
"""add values for bind parameters which may have been specified in filter().
-
+
parameters may be specified using \**kwargs, or optionally a single dictionary
- as the first positional argument. The reason for both is that \**kwargs is
+ as the first positional argument. The reason for both is that \**kwargs is
convenient, however some parameter dictionaries contain unicode keys in which case
\**kwargs cannot be used.
"""
-
+
q = self._clone()
if len(args) == 1:
d = args[0]
q._params = q._params.copy()
q._params.update(kwargs)
return q
-
+
def filter(self, criterion):
"""apply the given filtering criterion to the query and return the newly resulting ``Query``
-
+
the criterion is any sql.ClauseElement applicable to the WHERE clause of a select.
"""
-
+
if isinstance(criterion, basestring):
criterion = sql.text(criterion)
-
+
if criterion is not None and not isinstance(criterion, sql.ClauseElement):
raise exceptions.ArgumentError("filter() argument must be of type sqlalchemy.sql.ClauseElement or string")
-
-
+
+
if self._aliases is not None:
criterion = self._aliases.adapt_clause(criterion)
elif self.table not in self._get_joinable_tables():
criterion = sql_util.ClauseAdapter(self._from_obj).traverse(criterion)
-
+
q = self._no_statement("filter")
if q._criterion is not None:
q._criterion = q._criterion & criterion
clauses = [self._joinpoint.get_property(key, resolve_synonyms=True).compare(operators.eq, value)
for key, value in kwargs.iteritems()]
-
+
return self.filter(sql.and_(*clauses))
def _get_joinable_tables(self):
visitors.traverse(self._from_obj, visit_join=visit_join, traverse_options={'column_collections':False, 'aliased_selectables':False})
self.__joinable_tables = {self._from_obj : currenttables}
return self.__joinable_tables[self._from_obj]
-
+
def _join_to(self, keys, outerjoin=False, start=None, create_aliases=True):
if start is None:
start = self._joinpoint
-
+
clause = self._from_obj
currenttables = self._get_joinable_tables()
adapt_criterion = self.table not in currenttables
-
+
mapper = start
alias = self._aliases
for key in util.to_list(keys):
prop = mapper.get_property(key, resolve_synonyms=True)
if prop._is_self_referential() and not create_aliases:
raise exceptions.InvalidRequestError("Self-referential query on '%s' property requires create_aliases=True argument." % str(prop))
-
+
if prop.select_table not in currenttables or create_aliases:
if prop.secondary:
if create_aliases:
- alias = mapperutil.PropertyAliasedClauses(prop,
+ alias = mapperutil.PropertyAliasedClauses(prop,
prop.get_join(mapper, primary=True, secondary=False),
prop.get_join(mapper, primary=False, secondary=True),
alias
clause = clause.join(prop.select_table, prop.get_join(mapper, primary=False), isouter=outerjoin)
else:
if create_aliases:
- alias = mapperutil.PropertyAliasedClauses(prop,
+ alias = mapperutil.PropertyAliasedClauses(prop,
prop.get_join(mapper, primary=True, secondary=False),
None,
alias
# TODO: this check is not strong enough for different paths to the same endpoint which
# does not use secondary tables
raise exceptions.InvalidRequestError("Can't join to property '%s'; a path to this table along a different secondary table already exists. Use the `alias=True` argument to `join()`." % prop.key)
-
+
mapper = prop.mapper
-
+
if create_aliases:
return (clause, mapper, alias)
else:
"""Execute the SQL ``avg()`` function against the given column."""
return self._col_aggregate(col, sql.func.avg)
-
+
def order_by(self, criterion):
"""apply one or more ORDER BY criterion to the query and return the newly resulting ``Query``"""
q = self._no_statement("order_by")
-
+
if self._aliases is not None:
criterion = [expression._literal_as_text(o) for o in util.to_list(criterion) or []]
criterion = self._aliases.adapt_list(criterion)
-
- if q._order_by is False:
+
+ if q._order_by is False:
q._order_by = util.to_list(criterion)
else:
q._order_by = q._order_by + util.to_list(criterion)
"""apply one or more GROUP BY criterion to the query and return the newly resulting ``Query``"""
q = self._no_statement("group_by")
- if q._group_by is False:
+ if q._group_by is False:
q._group_by = util.to_list(criterion)
else:
q._group_by = q._group_by + util.to_list(criterion)
return q
-
+
def having(self, criterion):
"""apply a HAVING criterion to the quer and return the newly resulting ``Query``."""
-
+
if isinstance(criterion, basestring):
criterion = sql.text(criterion)
-
+
if criterion is not None and not isinstance(criterion, sql.ClauseElement):
raise exceptions.ArgumentError("having() argument must be of type sqlalchemy.sql.ClauseElement or string")
-
-
+
+
if self._aliases is not None:
criterion = self._aliases.adapt_clause(criterion)
-
+
q = self._no_statement("having")
if q._having is not None:
q._having = q._having & criterion
else:
q._having = criterion
return q
-
+
def join(self, prop, id=None, aliased=False, from_joinpoint=False):
"""create a join of this ``Query`` object's criterion
to a relationship and return the newly resulting ``Query``.
"""
return self._join(prop, id=id, outerjoin=False, aliased=aliased, from_joinpoint=from_joinpoint)
-
+
def outerjoin(self, prop, id=None, aliased=False, from_joinpoint=False):
"""create a left outer join of this ``Query`` object's criterion
to a relationship and return the newly resulting ``Query``.
-
+
'prop' may be a string property name or a list of string
property names.
"""
q._from_obj = clause
q._joinpoint = mapper
q._aliases = aliases
-
+
a = aliases
while a is not None:
q._alias_ids.setdefault(a.mapper, []).append(a)
q._alias_ids.setdefault(a.table, []).append(a)
q._alias_ids.setdefault(a.alias, []).append(a)
a = a.parentclauses
-
+
if id:
q._alias_ids[id] = aliases
return q
def reset_joinpoint(self):
- """return a new Query reset the 'joinpoint' of this Query reset
+ """return a new Query reset the 'joinpoint' of this Query reset
back to the starting mapper. Subsequent generative calls will
be constructed from the new joinpoint.
def select_from(self, from_obj):
- """Set the `from_obj` parameter of the query and return the newly
+ """Set the `from_obj` parameter of the query and return the newly
resulting ``Query``. This replaces the table which this Query selects
from with the given table.
-
-
- `from_obj` is a single table or selectable.
+
+
+ `from_obj` is a single table or selectable.
"""
new = self._no_criterion('select_from')
if isinstance(from_obj, (tuple, list)):
util.warn_deprecated("select_from() now accepts a single Selectable as its argument, which replaces any existing FROM criterion.")
from_obj = from_obj[-1]
-
+
if isinstance(from_obj, expression._SelectBaseMixin):
# alias SELECTs and unions
from_obj = from_obj.alias()
-
+
new._from_obj = from_obj
return new
-
+
def __getitem__(self, item):
if isinstance(item, slice):
start = item.start
This results in an execution of the underlying query.
"""
return list(self)
-
-
+
+
def from_statement(self, statement):
"""Execute the given SELECT statement and return results.
-
- This method bypasses all internal statement compilation, and the
- statement is executed without modification.
-
+
+ This method bypasses all internal statement compilation, and the
+ statement is executed without modification.
+
The statement argument is either a string, a ``select()`` construct,
or a ``text()`` construct, and should return the set of columns
appropriate to the entity class represented by this ``Query``.
-
+
Also see the ``instances()`` method.
-
+
"""
-
+
if isinstance(statement, basestring):
statement = sql.text(statement)
q = self._no_criterion('from_statement')
q._statement = statement
return q
-
+
def first(self):
"""Return the first result of this ``Query`` or None if the result doesn't contain any row.
This results in an execution of the underlying query.
"""
- if self._column_aggregate is not None:
+ if self._column_aggregate is not None:
return self._col_aggregate(*self._column_aggregate)
-
+
ret = list(self[0:1])
if len(ret) > 0:
return ret[0]
This results in an execution of the underlying query.
"""
- if self._column_aggregate is not None:
+ if self._column_aggregate is not None:
return self._col_aggregate(*self._column_aggregate)
ret = list(self[0:2])
-
+
if len(ret) == 1:
return ret[0]
elif len(ret) == 0:
raise exceptions.InvalidRequestError('No rows returned for one()')
else:
raise exceptions.InvalidRequestError('Multiple rows returned for one()')
-
+
def __iter__(self):
context = self._compile_context()
context.statement.use_labels = True
if self._autoflush and not self._populate_existing:
self.session._autoflush()
return self._execute_and_instances(context)
-
+
def _execute_and_instances(self, querycontext):
result = self.session.execute(querycontext.statement, params=self._params, mapper=self.mapper, instance=self._refresh_instance)
return self.iterate_instances(result, querycontext=querycontext)
def instances(self, cursor, *mappers_or_columns, **kwargs):
return list(self.iterate_instances(cursor, *mappers_or_columns, **kwargs))
-
+
def iterate_instances(self, cursor, *mappers_or_columns, **kwargs):
session = self.session
context = kwargs.pop('querycontext', None)
if context is None:
context = QueryContext(self)
-
+
context.runid = _new_runid()
-
+
mappers_or_columns = tuple(self._entities) + mappers_or_columns
tuples = bool(mappers_or_columns)
if self._primary_adapter:
def main(context, row):
- return self.select_mapper._instance(context, self._primary_adapter(row), None,
+ return self.select_mapper._instance(context, self._primary_adapter(row), None,
extension=context.extension, only_load_props=context.only_load_props, refresh_instance=context.refresh_instance
)
else:
def main(context, row):
- return self.select_mapper._instance(context, row, None,
+ return self.select_mapper._instance(context, row, None,
extension=context.extension, only_load_props=context.only_load_props, refresh_instance=context.refresh_instance
)
-
+
if tuples:
process = []
process.append(main)
while True:
context.progress = util.Set()
-
+
if self._yield_per:
fetch = cursor.fetchmany(self._yield_per)
if not fetch:
return
else:
fetch = cursor.fetchall()
-
+
if tuples:
rows = util.OrderedSet()
for row in fetch:
for ii in context.progress:
context.attributes.get(('populating_mapper', ii), _state_mapper(ii))._post_instance(context, ii)
ii.commit_all()
-
+
for row in rows:
yield row
-
+
if not self._yield_per:
break
-
+
def _get(self, key=None, ident=None, refresh_instance=None, lockmode=None, only_load_props=None):
lockmode = lockmode or self._lockmode
if not self._populate_existing and not refresh_instance and not self.mapper.always_refresh and lockmode is None:
return self.session.identity_map[key]
except KeyError:
pass
-
+
if ident is None:
if key is not None:
ident = key[1]
except IndexError:
raise exceptions.InvalidRequestError("Could not find enough values to formulate primary key for query.get(); primary key columns are %s" % ', '.join(["'%s'" % str(c) for c in self.primary_key_columns]))
q = q.params(params)
-
+
if lockmode is not None:
q = q.with_lockmode(lockmode)
q = q._select_context_options(populate_existing=bool(refresh_instance), version_check=(lockmode is not None), only_load_props=only_load_props, refresh_instance=refresh_instance)
def _count(self):
"""Apply this query's criterion to a SELECT COUNT statement.
-
- this is the purely generative version which will become
+
+ this is the purely generative version which will become
the public method in version 0.5.
"""
if self._autoflush and not self._populate_existing:
self.session._autoflush()
return self.session.scalar(s, params=self._params, mapper=self.mapper)
-
+
def compile(self):
"""compiles and returns a SQL statement based on the criterion and conditions within this Query."""
return self._compile_context().statement
-
+
def _compile_context(self):
context = QueryContext(self)
self._statement.use_labels = True
context.statement = self._statement
return context
-
+
whereclause = self._criterion
from_obj = self._from_obj
-
- # indicates if the "from" clause of the query does not include
+
+ # indicates if the "from" clause of the query does not include
# the normally mapped table, i.e. the user issued select_from(somestatement)
# or similar. all clauses which derive from the mapped table will need to
# be adapted to be relative to the user-supplied selectable.
if not adapt_criterion and whereclause is not None and (self.mapper is not self.select_mapper):
whereclause = sql_util.ClauseAdapter(from_obj, equivalents=self.select_mapper._get_equivalent_columns()).traverse(whereclause)
- # TODO: mappers added via add_entity(), adapt their queries also,
+ # TODO: mappers added via add_entity(), adapt their queries also,
# if those mappers are polymorphic
order_by = self._order_by
whereclause = sql.and_(whereclause, self.select_mapper.polymorphic_on.in_([m.polymorphic_identity for m in self.select_mapper.polymorphic_iterator()]))
context.from_clause = from_obj
-
+
# give all the attached properties a chance to modify the query
# TODO: doing this off the select_mapper. if its the polymorphic mapper, then
# it has no relations() on it. should we compile those too into the query ? (i.e. eagerloads)
if self._eager_loaders and self._nestable(**self._select_args()):
# eager loaders are present, and the SELECT has limiting criterion
# produce a "wrapped" selectable.
-
+
# ensure all 'order by' elements are ClauseElement instances
# (since they will potentially be aliased)
# locate all embedded Column clauses so they can be added to the
# "inner" select statement where they'll be available to the enclosing
# statement's "order by"
-
+
cf = util.Set()
if order_by:
order_by = [expression._literal_as_text(o) for o in util.to_list(order_by) or []]
cf = [from_obj.corresponding_column(c) or c for c in cf]
s2 = sql.select(context.primary_columns + list(cf), whereclause, from_obj=context.from_clause, use_labels=True, correlate=False, order_by=util.to_list(order_by), **self._select_args())
-
+
s3 = s2.alias()
-
+
self._primary_adapter = mapperutil.create_row_adapter(s3, self.table)
statement = sql.select([s3] + context.secondary_columns, for_update=for_update, use_labels=True)
if context.eager_joins:
eager_joins = sql_util.ClauseAdapter(s3).traverse(context.eager_joins)
statement.append_from(eager_joins, _copy_collection=False)
-
+
if order_by:
statement.append_order_by(*sql_util.ClauseAdapter(s3).copy_and_process(order_by))
-
+
statement.append_order_by(*context.eager_order_by)
else:
if adapt_criterion:
if adapt_criterion:
order_by = sql_util.ClauseAdapter(from_obj).copy_and_process(order_by)
-
+
if self._distinct and order_by:
cf = util.Set()
for o in order_by:
cf.update(sql_util.find_columns(o))
for c in cf:
context.primary_columns.append(c)
-
+
statement = sql.select(context.primary_columns + context.secondary_columns, whereclause, from_obj=from_obj, use_labels=True, for_update=for_update, order_by=util.to_list(order_by), **self._select_args())
-
+
if context.eager_joins:
if adapt_criterion:
context.eager_joins = sql_util.ClauseAdapter(from_obj).traverse(context.eager_joins)
if adapt_criterion:
context.eager_order_by = sql_util.ClauseAdapter(from_obj).copy_and_process(context.eager_order_by)
statement.append_order_by(*context.eager_order_by)
-
+
context.statement = statement
-
+
return context
def _select_args(self):
"""for tuples added via add_entity() or add_column(), attempt to locate
an AliasedClauses object which should be used to formulate the query as well
as to process result rows."""
-
+
(m, alias, alias_id) = m
if alias is not None:
return alias
return aliases[0]
else:
return None
-
+
def __log_debug(self, msg):
self.logger.debug(msg)
if params is not None:
q = q.params(params)
return list(q)
-
+
def _legacy_select_from(self, from_obj):
q = self._clone()
if len(from_obj) > 1:
raise exceptions.ArgumentError("Multiple-entry from_obj parameter no longer supported")
q._from_obj = from_obj[0]
return q
-
+
def _legacy_select_kwargs(self, **kwargs): #pragma: no cover
q = self
if "order_by" in kwargs and kwargs['order_by']:
def select_statement(self, statement, **params): #pragma: no cover
"""DEPRECATED. Use query.from_statement(statement)"""
-
+
return self._select_statement(statement, params)
def select_text(self, text, **params): #pragma: no cover
if only_load_props:
self._only_load_props = util.Set(only_load_props)
return self
-
+
def join_to(self, key): #pragma: no cover
"""DEPRECATED. use join() to create joins based on property names."""
if mapper_ in seen:
return None
seen.add(mapper_)
-
+
prop = mapper_.get_property(key, resolve_synonyms=True, raiseerr=False)
if prop is not None:
if isinstance(prop, properties.PropertyLoader):
self.eager_joins = None
self.options = query._with_options
self.attributes = query._attributes.copy()
-
+
def exec_with_path(self, mapper, propkey, func, *args, **kwargs):
oldpath = self.path
self.path += (mapper.base_mapper, propkey)
return _runid
finally:
_id_lock.release()
-
to stay the same in future releases.
"""
-import re
-import datetime
-import warnings
+import datetime, re
from itertools import chain
from sqlalchemy import util, exceptions
from sqlalchemy.sql import operators, visitors
existing = self[key]
if not existing.shares_lineage(value):
table = getattr(existing, 'table', None) and existing.table.description
- warnings.warn(RuntimeWarning("Column %r on table %r being replaced by another column with the same key. Consider use_labels for select() statements." % (key, table)))
+ util.warn(("Column %r on table %r being replaced by another "
+ "column with the same key. Consider use_labels "
+ "for select() statements.") % (key, table))
util.OrderedProperties.__setitem__(self, key, value)
def remove(self, column):
import inspect
import datetime as dt
-import warnings
from sqlalchemy import exceptions
from sqlalchemy.util import pickle, Decimal as _python_Decimal
return value.encode(dialect.encoding)
elif assert_unicode and not isinstance(value, (unicode, NoneType)):
if assert_unicode == 'warn':
- warnings.warn(RuntimeWarning("Unicode type received non-unicode bind param value %r" % value))
+ util.warn("Unicode type received non-unicode bind "
+ "param value %r" % value)
return value
else:
raise exceptions.InvalidRequestError("Unicode type received non-unicode bind param value %r" % value)
import itertools, sys, warnings, sets, weakref
import __builtin__
-from sqlalchemy import exceptions, logging
+from sqlalchemy import exceptions
try:
import thread, threading
except ImportError:
def Decimal(arg):
if Decimal.warn:
- warnings.warn(RuntimeWarning(
- "True Decimal types not available on this Python, "
- "falling back to floats."))
+ warn("True Decimal types not available on this Python, "
+ "falling back to floats.")
Decimal.warn = False
return float(arg)
Decimal.warn = True
try:
return func(*args, **kwargs)
except:
- warnings.warn(RuntimeWarning("%s('%s') ignored" % sys.exc_info()[0:2]))
+ warn("%s('%s') ignored" % sys.exc_info()[0:2])
class SimpleProperty(object):
"""A *default* property accessor."""
def _get_key(self):
return self.scopefunc()
+def warn(msg):
+ if isinstance(msg, basestring):
+ warnings.warn(msg, exceptions.SAWarning, stacklevel=3)
+ else:
+ warnings.warn(msg, stacklevel=3)
def warn_deprecated(msg):
- warnings.warn(logging.SADeprecationWarning(msg), stacklevel=3)
+ warnings.warn(msg, exceptions.SADeprecationWarning, stacklevel=3)
def deprecated(func, message=None, add_deprecation_to_docstring=True):
+ """Decorates a function and issues a deprecation warning on use.
+
+ message
+ If provided, issue message in the warning. A sensible default
+ is used if not provided.
+
+ add_deprecation_to_docstring
+ Default True. If False, the wrapped function's __doc__ is left
+ as-is. If True, the 'message' is prepended to the docs if
+ provided, or sensible default if message is omitted.
+ """
+
if message is not None:
warning = message % dict(func=func.__name__)
else:
warning = "Call to deprecated function %s" % func.__name__
def func_with_warning(*args, **kwargs):
- warnings.warn(logging.SADeprecationWarning(warning),
+ warnings.warn(exceptions.SADeprecationWarning(warning),
stacklevel=2)
return func(*args, **kwargs)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
