expression.py

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

    """

    return _TextClause(text, bind=bind, *args, **kwargs)

def null():
    """Return a ``_Null`` object, which compiles to ``NULL`` in a sql statement."""

    return _Null()

class _FunctionGenerator(object):
    """Generate ``_Function`` objects based on getattr calls."""

    def __init__(self, **opts):
        self.__names = []
        self.opts = opts

    def __getattr__(self, name):
        # passthru __ attributes; fixes pydoc
        if name.startswith('__'):
            try:
                return self.__dict__[name]
            except KeyError:
                raise AttributeError(name)

        elif name.startswith('_'):
            name = name[0:-1]
        f = _FunctionGenerator(**self.opts)
        f.__names = list(self.__names) + [name]
        return f

    def __call__(self, *c, **kwargs):
        o = self.opts.copy()
        o.update(kwargs)
        if len(self.__names) == 1:
            global functions
            if functions is None:
                from sqlalchemy.sql import functions
            func = getattr(functions, self.__names[-1].lower(), None)
            if func is not None:
                return func(*c, **o)

        return _Function(self.__names[-1], packagenames=self.__names[0:-1], *c, **o)

func = _FunctionGenerator()

# TODO: use UnaryExpression for this instead ?
modifier = _FunctionGenerator(group=False)

def _clone(element):
    return element._clone()

def _compound_select(keyword, *selects, **kwargs):
    return CompoundSelect(keyword, *selects, **kwargs)

def _is_literal(element):
    return not isinstance(element, ClauseElement)

def _literal_as_text(element):
    if isinstance(element, Operators):
        return element.expression_element()
    elif _is_literal(element):
        return _TextClause(unicode(element))
    else:
        return element

def _literal_as_column(element):
    if isinstance(element, Operators):
        return element.clause_element()
    elif _is_literal(element):
        return literal_column(str(element))
    else:
        return element

def _literal_as_binds(element, name=None, type_=None):
    if isinstance(element, Operators):
        return element.expression_element()
    elif _is_literal(element):
        if element is None:
            return null()
        else:
            return _BindParamClause(name, element, type_=type_, unique=True)
    else:
        return element

def _corresponding_column_or_error(fromclause, column, require_embedded=False):
    c = fromclause.corresponding_column(column, require_embedded=require_embedded)
    if not c:
        raise exceptions.InvalidRequestError("Given column '%s', attached to table '%s', failed to locate a corresponding column from table '%s'" % (str(column), str(getattr(column, 'table', None)), fromclause.description))
    return c

def _selectable(element):
    if hasattr(element, '__selectable__'):
        return element.__selectable__()
    elif isinstance(element, Selectable):
        return element
    else:
        raise exceptions.ArgumentError("Object '%s' is not a Selectable and does not implement `__selectable__()`" % repr(element))

def is_column(col):
    """True if ``col`` is an instance of ``ColumnElement``."""
    return isinstance(col, ColumnElement)


class _FigureVisitName(type):
    def __init__(cls, clsname, bases, dict):
        if not '__visit_name__' in cls.__dict__:
            m = re.match(r'_?(\w+?)(?:Expression|Clause|Element|$)', clsname)
            x = m.group(1)
            x = re.sub(r'(?!^)[A-Z]', lambda m:'_'+m.group(0).lower(), x)
            cls.__visit_name__ = x.lower()
        super(_FigureVisitName, cls).__init__(clsname, bases, dict)

class ClauseElement(object):
    """Base class for elements of a programmatically constructed SQL expression."""
    __metaclass__ = _FigureVisitName

    def _clone(self):
        """Create a shallow copy of this ClauseElement.

        This method may be used by a generative API.  Its also used as
        part of the "deep" copy afforded by a traversal that combines
        the _copy_internals() method.
        """
        c = self.__class__.__new__(self.__class__)
        c.__dict__ = self.__dict__.copy()

        # this is a marker that helps to "equate" clauses to each other
        # when a Select returns its list of FROM clauses.  the cloning
        # process leaves around a lot of remnants of the previous clause
        # typically in the form of column expressions still attached to the
        # old table.
        c._is_clone_of = self

        return c

    def _cloned_set(self):
        f = self
        while f is not None:
            yield f
            f = getattr(f, '_is_clone_of', None)
    _cloned_set = property(_cloned_set)

    def _get_from_objects(self, **modifiers):
        """Return objects represented in this ``ClauseElement`` that
        should be added to the ``FROM`` list of a query, when this
        ``ClauseElement`` is placed in the column clause of a
        ``Select`` statement.
        """

        raise NotImplementedError(repr(self))

    def unique_params(self, *optionaldict, **kwargs):
        """Return a copy with ``bindparam()`` elments replaced.

        Same functionality as ``params()``, except adds `unique=True`
        to affected bind parameters so that multiple statements can be
        used.
        """

        return self._params(True, optionaldict, kwargs)

    def params(self, *optionaldict, **kwargs):
        """Return a copy with ``bindparam()`` elments replaced.

        Returns a copy of this ClauseElement with ``bindparam()``
        elements replaced with values taken from the given dictionary::

          >>> clause = column('x') + bindparam('foo')
          >>> print clause.compile().params
          {'foo':None}
          >>> print clause.params({'foo':7}).compile().params
          {'foo':7}
        """

        return self._params(False, optionaldict, kwargs)

    def _params(self, unique, optionaldict, kwargs):
        if len(optionaldict) == 1:
            kwargs.update(optionaldict[0])
        elif len(optionaldict) > 1:
            raise exceptions.ArgumentError("params() takes zero or one positional dictionary argument")

        def visit_bindparam(bind):
            if bind.key in kwargs:
                bind.value = kwargs[bind.key]
            if unique:
                bind._convert_to_unique()
        return visitors.traverse(self, visit_bindparam=visit_bindparam, clone=True)

    def compare(self, other):
        """Compare this ClauseElement to the given ClauseElement.

        Subclasses should override the default behavior, which is a
        straight identity comparison.
        """

        return self is other

    def _copy_internals(self, clone=_clone):
        """Reassign internal elements to be clones of themselves.

        Called during a copy-and-traverse operation on newly
        shallow-copied elements to create a deep copy.
        """

        pass

    def get_children(self, **kwargs):
        """Return immediate child elements of this ``ClauseElement``.

        This is used for visit traversal.

        \**kwargs may contain flags that change the collection that is
        returned, for example to return a subset of items in order to
        cut down on larger traversals, or to return child items from a
        different context (such as schema-level collections instead of
        clause-level).
        """
        return []

    def self_group(self, against=None):
        return self

    def supports_execution(self):
        """Return True if this clause element represents a complete executable statement."""

        return False

    def bind(self):
        """Returns the Engine or Connection to which this ClauseElement is bound, or None if none found."""

        try:
            if self._bind is not None:
                return self._bind
        except AttributeError:
            pass
        for f in self._get_from_objects():
            if f is self:
                continue
            engine = f.bind
            if engine is not None:
                return engine
        else:
            return None
    bind = property(bind)

    def execute(self, *multiparams, **params):
        """Compile and execute this ``ClauseElement``."""

        e = self.bind
        if e is None:
            label = getattr(self, 'description', self.__class__.__name__)
            msg = ('This %s is not bound to an Engine or Connection.  '
                   'Execution can not proceed without a database to execute '
                   'against.  Either execute with an explicit connection or '
                   'bind the MetaData of the underlying tables to enable '
                   'implicit execution.') % label
            raise exceptions.UnboundExecutionError(msg)
        return e.execute_clauseelement(self, multiparams, params)

    def scalar(self, *multiparams, **params):
        """Compile and execute this ``ClauseElement``, returning the result's scalar representation."""

        return self.execute(*multiparams, **params).scalar()

    def compile(self, bind=None, column_keys=None, compiler=None, dialect=None, inline=False):
        """Compile this SQL expression.

        Uses the given ``Compiler``, or the given ``AbstractDialect``
        or ``Engine`` to create a ``Compiler``.  If no `compiler`
        arguments are given, tries to use the underlying ``Engine``
        this ``ClauseElement`` is bound to to create a ``Compiler``,
        if any.

        Finally, if there is no bound ``Engine``, uses an
        ``DefaultDialect`` to create a default ``Compiler``.

        `parameters` is a dictionary representing the default bind
        parameters to be used with the statement.  If `parameters` is
        a list, it is assumed to be a list of dictionaries and the
        first dictionary in the list is used with which to compile
        against.

        The bind parameters can in some cases determine the output of
        the compilation, such as for ``UPDATE`` and ``INSERT``
        statements the bind parameters that are present determine the
        ``SET`` and ``VALUES`` clause of those statements.
        """

        if compiler is None:
            if dialect is not None:
                compiler = dialect.statement_compiler(dialect, self, column_keys=column_keys, inline=inline)
            elif bind is not None:
                compiler = bind.statement_compiler(self, column_keys=column_keys, inline=inline)
            elif self.bind is not None:
                compiler = self.bind.statement_compiler(self, column_keys=column_keys, inline=inline)

        if compiler is None:
            global DefaultDialect
            if DefaultDialect is None:
                from sqlalchemy.engine.default import DefaultDialect
            dialect = DefaultDialect()
            compiler = dialect.statement_compiler(dialect, self, column_keys=column_keys, inline=inline)
        compiler.compile()
        return compiler

    def __str__(self):
        return unicode(self.compile()).encode('ascii', 'backslashreplace')

    def __and__(self, other):
        return and_(self, other)

    def __or__(self, other):
        return or_(self, other)

    def __invert__(self):
        return self._negate()

    def _negate(self):
        if hasattr(self, 'negation_clause'):
            return self.negation_clause
        else:
            return _UnaryExpression(self.self_group(against=operators.inv), operator=operators.inv, negate=None)

    def __repr__(self):
        friendly = getattr(self, 'description', None)
        if friendly is None:
            return object.__repr__(self)
        else:
            return '<%s.%s at 0x%x; %s>' % (
                self.__module__, self.__class__.__name__, id(self), friendly)


class Operators(object):
    def __and__(self, other):
        return self.operate(operators.and_, other)

    def __or__(self, other):
        return self.operate(operators.or_, other)

    def __invert__(self):
        return self.operate(operators.inv)

    def op(self, opstring):
        def op(b):
            return self.operate(operators.op, opstring, b)
        return op

    def clause_element(self):
        raise NotImplementedError()

    def operate(self, op, *other, **kwargs):
        raise NotImplementedError()

    def reverse_operate(self, op, other, **kwargs):
        raise NotImplementedError()

class ColumnOperators(Operators):
    """Defines comparison and math operations."""

    timetuple = None
    """Hack, allows datetime objects to be compared on the LHS."""

    def __lt__(self, other):
        return self.operate(operators.lt, other)

    def __le__(self, other):
        return self.operate(operators.le, other)

    def __eq__(self, other):
        return self.operate(operators.eq, other)

    def __ne__(self, other):
        return self.operate(operators.ne, other)