interfaces.py

SQLAlchemy. 经典的Python ORM框架。学习必看。

        ``PropertyLoaders``.

        This is called by a ``Query``'s ``join_by`` method to formulate a set
        of key/value pairs into a ``WHERE`` criterion that spans multiple
        tables if needed.
        """

        return None

    def set_parent(self, parent):
        self.parent = parent

    def init(self, key, parent):
        """Called after all mappers are compiled to assemble
        relationships between mappers, establish instrumented class
        attributes.
        """

        self.key = key
        self.do_init()

    def do_init(self):
        """Perform subclass-specific initialization steps.

        This is a *template* method called by the
        ``MapperProperty`` object's init() method."""

        pass

    def register_dependencies(self, *args, **kwargs):
        """Called by the ``Mapper`` in response to the UnitOfWork
        calling the ``Mapper``'s register_dependencies operation.
        Should register with the UnitOfWork all inter-mapper
        dependencies as well as dependency processors (see UOW docs
        for more details).
        """

        pass

    def is_primary(self):
        """Return True if this ``MapperProperty``'s mapper is the
        primary mapper for its class.

        This flag is used to indicate that the ``MapperProperty`` can
        define attribute instrumentation for the class at the class
        level (as opposed to the individual instance level).
        """

        return not self.parent.non_primary

    def merge(self, session, source, dest):
        """Merge the attribute represented by this ``MapperProperty``
        from source to destination object"""

        raise NotImplementedError()

    def compare(self, operator, value):
        """Return a compare operation for the columns represented by
        this ``MapperProperty`` to the given value, which may be a
        column value or an instance.  'operator' is an operator from
        the operators module, or from sql.Comparator.

        By default uses the PropComparator attached to this MapperProperty
        under the attribute name "comparator".
        """

        return operator(self.comparator, value)

class PropComparator(expression.ColumnOperators):
    """defines comparison operations for MapperProperty objects"""

    def expression_element(self):
        return self.clause_element()

    def contains_op(a, b):
        return a.contains(b)
    contains_op = staticmethod(contains_op)

    def any_op(a, b, **kwargs):
        return a.any(b, **kwargs)
    any_op = staticmethod(any_op)

    def has_op(a, b, **kwargs):
        return a.has(b, **kwargs)
    has_op = staticmethod(has_op)

    def __init__(self, prop):
        self.prop = prop

    def contains(self, other):
        """Return true if this collection contains other"""
        return self.operate(PropComparator.contains_op, other)

    def any(self, criterion=None, **kwargs):
        """Return true if this collection contains any member that meets the given criterion.

        criterion
          an optional ClauseElement formulated against the member class' table
          or attributes.

        \**kwargs
          key/value pairs corresponding to member class attribute names which
          will be compared via equality to the corresponding values.
        """

        return self.operate(PropComparator.any_op, criterion, **kwargs)

    def has(self, criterion=None, **kwargs):
        """Return true if this element references a member which meets the given criterion.

        criterion
          an optional ClauseElement formulated against the member class' table
          or attributes.

        \**kwargs
          key/value pairs corresponding to member class attribute names which
          will be compared via equality to the corresponding values.
        """

        return self.operate(PropComparator.has_op, criterion, **kwargs)

class StrategizedProperty(MapperProperty):
    """A MapperProperty which uses selectable strategies to affect
    loading behavior.

    There is a single default strategy selected by default.  Alternate
    strategies can be selected at Query time through the usage of
    ``StrategizedOption`` objects via the Query.options() method.
    """

    def _get_context_strategy(self, context):
        path = context.path
        return self._get_strategy(context.attributes.get(("loaderstrategy", path), self.strategy.__class__))

    def _get_strategy(self, cls):
        try:
            return self._all_strategies[cls]
        except KeyError:
            # cache the located strategy per class for faster re-lookup
            strategy = cls(self)
            strategy.init()
            self._all_strategies[cls] = strategy
            return strategy

    def setup(self, querycontext, **kwargs):
        self._get_context_strategy(querycontext).setup_query(querycontext, **kwargs)

    def create_row_processor(self, selectcontext, mapper, row):
        return self._get_context_strategy(selectcontext).create_row_processor(selectcontext, mapper, row)

    def do_init(self):
        self._all_strategies = {}
        self.strategy = self.create_strategy()
        self._all_strategies[self.strategy.__class__] = self.strategy
        self.strategy.init()
        if self.is_primary():
            self.strategy.init_class_attribute()

def build_path(mapper, key, prev=None):
    if prev:
        return prev + (mapper.base_mapper, key)
    else:
        return (mapper.base_mapper, key)

def serialize_path(path):
    if path is None:
        return None

    return [
        (mapper.class_, mapper.entity_name, key)
        for mapper, key in [(path[i], path[i+1]) for i in range(0, len(path)-1, 2)]
    ]

def deserialize_path(path):
    if path is None:
        return None

    global class_mapper
    if class_mapper is None:
        from sqlalchemy.orm import class_mapper

    return tuple(
        chain(*[(class_mapper(cls, entity), key) for cls, entity, key in path])
    )

class MapperOption(object):
    """Describe a modification to a Query."""

    def process_query(self, query):
        pass

    def process_query_conditionally(self, query):
        """same as process_query(), except that this option may not apply
        to the given query.

        Used when secondary loaders resend existing options to a new
        Query."""
        self.process_query(query)

class ExtensionOption(MapperOption):
    """a MapperOption that applies a MapperExtension to a query operation."""

    def __init__(self, ext):
        self.ext = ext

    def process_query(self, query):
        query._extension = query._extension.copy()
        query._extension.insert(self.ext)


class PropertyOption(MapperOption):
    """A MapperOption that is applied to a property off the mapper or
    one of its child mappers, identified by a dot-separated key.
    """

    def __init__(self, key, mapper=None):
        self.key = key
        self.mapper = mapper

    def process_query(self, query):
        self._process(query, True)

    def process_query_conditionally(self, query):
        self._process(query, False)

    def _process(self, query, raiseerr):
        if self._should_log_debug:
            self.logger.debug("applying option to Query, property key '%s'" % self.key)
        paths = self._get_paths(query, raiseerr)
        if paths:
            self.process_query_property(query, paths)

    def process_query_property(self, query, paths):
        pass

    def _get_paths(self, query, raiseerr):
        path = None
        l = []
        current_path = list(query._current_path)

        if self.mapper:
            global class_mapper
            if class_mapper is None:
                from sqlalchemy.orm import class_mapper
            mapper = self.mapper
            if isinstance(self.mapper, type):
                mapper = class_mapper(mapper)
            if mapper is not query.mapper and mapper not in [q[0] for q in query._entities]:
                raise exceptions.ArgumentError("Can't find entity %s in Query.  Current list: %r" % (str(mapper), [str(m) for m in [query.mapper] + query._entities]))
        else:
            mapper = query.mapper
        for token in self.key.split('.'):
            if current_path and token == current_path[1]:
                current_path = current_path[2:]
                continue
            prop = mapper.get_property(token, resolve_synonyms=True, raiseerr=raiseerr)
            if prop is None:
                return []
            path = build_path(mapper, prop.key, path)
            l.append(path)
            mapper = getattr(prop, 'mapper', None)
        return l

PropertyOption.logger = logging.class_logger(PropertyOption)
PropertyOption._should_log_debug = logging.is_debug_enabled(PropertyOption.logger)

class AttributeExtension(object):
    """An abstract class which specifies `append`, `delete`, and `set`
    event handlers to be attached to an object property.
    """

    def append(self, obj, child, initiator):
        pass

    def remove(self, obj, child, initiator):
        pass

    def set(self, obj, child, oldchild, initiator):
        pass


class StrategizedOption(PropertyOption):
    """A MapperOption that affects which LoaderStrategy will be used
    for an operation by a StrategizedProperty.
    """

    def is_chained(self):
        return False

    def process_query_property(self, query, paths):
        if self.is_chained():
            for path in paths:
                query._attributes[("loaderstrategy", path)] = self.get_strategy_class()
        else:
            query._attributes[("loaderstrategy", paths[-1])] = self.get_strategy_class()

    def get_strategy_class(self):
        raise NotImplementedError()


class LoaderStrategy(object):
    """Describe the loading behavior of a StrategizedProperty object.

    The ``LoaderStrategy`` interacts with the querying process in three
    ways:

    * it controls the configuration of the ``InstrumentedAttribute``
      placed on a class to handle the behavior of the attribute.  this
      may involve setting up class-level callable functions to fire
      off a select operation when the attribute is first accessed
      (i.e. a lazy load)

    * it processes the ``QueryContext`` at statement construction time,
      where it can modify the SQL statement that is being produced.
      simple column attributes may add their represented column to the
      list of selected columns, *eager loading* properties may add
      ``LEFT OUTER JOIN`` clauses to the statement.

    * it processes the ``SelectionContext`` at row-processing time.  This
      includes straight population of attributes corresponding to rows,
      setting instance-level lazyloader callables on newly
      constructed instances, and appending child items to scalar/collection
      attributes in response to eagerly-loaded relations.
    """

    def __init__(self, parent):
        self.parent_property = parent
        self.is_class_level = False

    def init(self):
        self.parent = self.parent_property.parent
        self.key = self.parent_property.key

    def init_class_attribute(self):
        pass

    def setup_query(self, context, **kwargs):
        pass

    def create_row_processor(self, selectcontext, mapper, row):
        """Return row processing functions which fulfill the contract specified
        by MapperProperty.create_row_processor.

        StrategizedProperty delegates its create_row_processor method directly
        to this method.
        """

        raise NotImplementedError()