self.drop(bind=bind, checkfirst=checkfirst)
-class _ColonCast(elements.Cast):
+class _ColonCast(elements.CompilerColumnElement):
__visit_name__ = "colon_cast"
+ __slots__ = ("type", "clause", "typeclause")
def __init__(self, expression, type_):
self.type = type_
class SupportsAnnotations:
+ __slots__ = ()
+
_annotations = EMPTY_ANNOTATIONS
@util.memoized_property
class SupportsCloneAnnotations(SupportsAnnotations):
+ __slots__ = ()
_clone_annotations_traverse_internals = [
("_annotations", InternalTraversal.dp_annotations_key)
class SupportsWrappingAnnotations(SupportsAnnotations):
+ __slots__ = ()
+
def _annotate(self, values):
"""return a copy of this ClauseElement with annotations
updated by the given dictionary.
# this was a Visitable, but to allow accurate detection of
# column elements this is actually a column element
-class _CompileLabel(elements.ColumnElement):
+class _CompileLabel(elements.CompilerColumnElement):
"""lightweight label object which acts as an expression.Label."""
__visit_name__ = "label"
- __slots__ = "element", "name"
+ __slots__ = "element", "name", "_alt_names"
def __init__(self, col, name, alt_names=()):
self.element = col
return operators.inv(coercions.expect(roles.ExpressionElementRole, clause))
+class CompilerElement(Traversible):
+ """base class for SQL elements that can be compiled to produce a
+ SQL string.
+
+ .. versionadded:: 2.0
+
+ """
+
+ __slots__ = ()
+ __visit_name__ = "compiler_element"
+
+ supports_execution = False
+
+ stringify_dialect = "default"
+
+ @util.preload_module("sqlalchemy.engine.default")
+ @util.preload_module("sqlalchemy.engine.url")
+ def compile(self, bind=None, dialect=None, **kw):
+ """Compile this SQL expression.
+
+ The return value is a :class:`~.Compiled` object.
+ Calling ``str()`` or ``unicode()`` on the returned value will yield a
+ string representation of the result. The
+ :class:`~.Compiled` object also can return a
+ dictionary of bind parameter names and values
+ using the ``params`` accessor.
+
+ :param bind: An ``Engine`` or ``Connection`` from which a
+ ``Compiled`` will be acquired. This argument takes precedence over
+ this :class:`_expression.ClauseElement`'s bound engine, if any.
+
+ :param column_keys: Used for INSERT and UPDATE statements, a list of
+ column names which should be present in the VALUES clause of the
+ compiled statement. If ``None``, all columns from the target table
+ object are rendered.
+
+ :param dialect: A ``Dialect`` instance from which a ``Compiled``
+ will be acquired. This argument takes precedence over the `bind`
+ argument as well as this :class:`_expression.ClauseElement`
+ 's bound engine,
+ if any.
+
+ :param compile_kwargs: optional dictionary of additional parameters
+ that will be passed through to the compiler within all "visit"
+ methods. This allows any custom flag to be passed through to
+ a custom compilation construct, for example. It is also used
+ for the case of passing the ``literal_binds`` flag through::
+
+ from sqlalchemy.sql import table, column, select
+
+ t = table('t', column('x'))
+
+ s = select(t).where(t.c.x == 5)
+
+ print(s.compile(compile_kwargs={"literal_binds": True}))
+
+ .. versionadded:: 0.9.0
+
+ .. seealso::
+
+ :ref:`faq_sql_expression_string`
+
+ """
+
+ if not dialect:
+ if bind:
+ dialect = bind.dialect
+ elif self.bind:
+ dialect = self.bind.dialect
+ else:
+ if self.stringify_dialect == "default":
+ default = util.preloaded.engine_default
+ dialect = default.StrCompileDialect()
+ else:
+ url = util.preloaded.engine_url
+ dialect = url.URL.create(
+ self.stringify_dialect
+ ).get_dialect()()
+
+ return self._compiler(dialect, **kw)
+
+ def _compiler(self, dialect, **kw):
+ """Return a compiler appropriate for this ClauseElement, given a
+ Dialect."""
+
+ return dialect.statement_compiler(dialect, self, **kw)
+
+ def __str__(self):
+ return str(self.compile())
+
+
@inspection._self_inspects
class ClauseElement(
- roles.SQLRole,
SupportsWrappingAnnotations,
MemoizedHasCacheKey,
HasCopyInternals,
- Traversible,
+ CompilerElement,
):
"""Base class for elements of a programmatically constructed SQL
expression.
"""
- supports_execution = False
-
- stringify_dialect = "default"
-
_from_objects = []
bind = None
description = None
return self
- @util.preload_module("sqlalchemy.engine.default")
- @util.preload_module("sqlalchemy.engine.url")
- def compile(self, bind=None, dialect=None, **kw):
- """Compile this SQL expression.
-
- The return value is a :class:`~.Compiled` object.
- Calling ``str()`` or ``unicode()`` on the returned value will yield a
- string representation of the result. The
- :class:`~.Compiled` object also can return a
- dictionary of bind parameter names and values
- using the ``params`` accessor.
-
- :param bind: An ``Engine`` or ``Connection`` from which a
- ``Compiled`` will be acquired. This argument takes precedence over
- this :class:`_expression.ClauseElement`'s bound engine, if any.
-
- :param column_keys: Used for INSERT and UPDATE statements, a list of
- column names which should be present in the VALUES clause of the
- compiled statement. If ``None``, all columns from the target table
- object are rendered.
-
- :param dialect: A ``Dialect`` instance from which a ``Compiled``
- will be acquired. This argument takes precedence over the `bind`
- argument as well as this :class:`_expression.ClauseElement`
- 's bound engine,
- if any.
-
- :param compile_kwargs: optional dictionary of additional parameters
- that will be passed through to the compiler within all "visit"
- methods. This allows any custom flag to be passed through to
- a custom compilation construct, for example. It is also used
- for the case of passing the ``literal_binds`` flag through::
-
- from sqlalchemy.sql import table, column, select
-
- t = table('t', column('x'))
-
- s = select(t).where(t.c.x == 5)
-
- print(s.compile(compile_kwargs={"literal_binds": True}))
-
- .. versionadded:: 0.9.0
-
- .. seealso::
-
- :ref:`faq_sql_expression_string`
-
- """
-
- if not dialect:
- if bind:
- dialect = bind.dialect
- elif self.bind:
- dialect = self.bind.dialect
- else:
- if self.stringify_dialect == "default":
- default = util.preloaded.engine_default
- dialect = default.StrCompileDialect()
- else:
- url = util.preloaded.engine_url
- dialect = url.URL.create(
- self.stringify_dialect
- ).get_dialect()()
-
- return self._compiler(dialect, **kw)
-
def _compile_w_cache(
self,
dialect,
return compiled_sql, extracted_params, cache_hit
- def _compiler(self, dialect, **kw):
- """Return a compiler appropriate for this ClauseElement, given a
- Dialect."""
-
- return dialect.statement_compiler(dialect, self, **kw)
-
- def __str__(self):
- if util.py3k:
- return str(self.compile())
- else:
- return unicode(self.compile()).encode( # noqa
- "ascii", "backslashreplace"
- ) # noqa
-
def __invert__(self):
# undocumented element currently used by the ORM for
# relationship.contains()
)
+class CompilerColumnElement(
+ roles.DMLColumnRole,
+ roles.DDLConstraintColumnRole,
+ roles.ColumnsClauseRole,
+ CompilerElement,
+):
+ """A compiler-only column element used for ad-hoc string compilations.
+
+ .. versionadded:: 2.0
+
+ """
+
+ __slots__ = ()
+
+
class ColumnElement(
roles.ColumnArgumentOrKeyRole,
roles.StatementOptionRole,
"""
__visit_name__ = "column_element"
+
primary_key = False
foreign_keys = []
_proxies = ()
"""
+ __slots__ = ()
allows_lambda = False
uses_inspection = False
class UsesInspection:
+ __slots__ = ()
_post_inspect = None
uses_inspection = True
class AllowsLambdaRole:
+ __slots__ = ()
allows_lambda = True
class HasCacheKeyRole(SQLRole):
+ __slots__ = ()
_role_name = "Cacheable Core or ORM object"
class LiteralValueRole(SQLRole):
+ __slots__ = ()
_role_name = "Literal Python value"
class ColumnArgumentRole(SQLRole):
+ __slots__ = ()
_role_name = "Column expression"
class ColumnArgumentOrKeyRole(ColumnArgumentRole):
+ __slots__ = ()
_role_name = "Column expression or string key"
class StrAsPlainColumnRole(ColumnArgumentRole):
+ __slots__ = ()
_role_name = "Column expression or string key"
class ColumnListRole(SQLRole):
"""Elements suitable for forming comma separated lists of expressions."""
+ __slots__ = ()
+
class TruncatedLabelRole(SQLRole):
+ __slots__ = ()
_role_name = "String SQL identifier"
class ColumnsClauseRole(AllowsLambdaRole, UsesInspection, ColumnListRole):
+ __slots__ = ()
_role_name = "Column expression or FROM clause"
@property
class LimitOffsetRole(SQLRole):
+ __slots__ = ()
_role_name = "LIMIT / OFFSET expression"
class ByOfRole(ColumnListRole):
+ __slots__ = ()
_role_name = "GROUP BY / OF / etc. expression"
class GroupByRole(AllowsLambdaRole, UsesInspection, ByOfRole):
+ __slots__ = ()
# note there's a special case right now where you can pass a whole
# ORM entity to group_by() and it splits out. we may not want to keep
# this around
class OrderByRole(AllowsLambdaRole, ByOfRole):
+ __slots__ = ()
_role_name = "ORDER BY expression"
class StructuralRole(SQLRole):
- pass
+ __slots__ = ()
class StatementOptionRole(StructuralRole):
+ __slots__ = ()
_role_name = "statement sub-expression element"
class OnClauseRole(AllowsLambdaRole, StructuralRole):
+ __slots__ = ()
_role_name = "SQL expression for ON clause"
class WhereHavingRole(OnClauseRole):
+ __slots__ = ()
_role_name = "SQL expression for WHERE/HAVING role"
class ExpressionElementRole(SQLRole):
+ __slots__ = ()
_role_name = "SQL expression element"
class ConstExprRole(ExpressionElementRole):
+ __slots__ = ()
_role_name = "Constant True/False/None expression"
class LabeledColumnExprRole(ExpressionElementRole):
- pass
+ __slots__ = ()
class BinaryElementRole(ExpressionElementRole):
+ __slots__ = ()
_role_name = "SQL expression element or literal value"
class InElementRole(SQLRole):
+ __slots__ = ()
_role_name = (
"IN expression list, SELECT construct, or bound parameter object"
)
class JoinTargetRole(AllowsLambdaRole, UsesInspection, StructuralRole):
+ __slots__ = ()
_role_name = (
"Join target, typically a FROM expression, or ORM "
"relationship attribute"
class FromClauseRole(ColumnsClauseRole, JoinTargetRole):
+ __slots__ = ()
_role_name = "FROM expression, such as a Table or alias() object"
_is_subquery = False
class StrictFromClauseRole(FromClauseRole):
+ __slots__ = ()
# does not allow text() or select() objects
@property
class AnonymizedFromClauseRole(StrictFromClauseRole):
+ __slots__ = ()
# calls .alias() as a post processor
def _anonymous_fromclause(self, name=None, flat=False):
class ReturnsRowsRole(SQLRole):
+ __slots__ = ()
_role_name = (
"Row returning expression such as a SELECT, a FROM clause, or an "
"INSERT/UPDATE/DELETE with RETURNING"
class StatementRole(SQLRole):
+ __slots__ = ()
_role_name = "Executable SQL or text() construct"
_propagate_attrs = util.immutabledict()
class SelectStatementRole(StatementRole, ReturnsRowsRole):
+ __slots__ = ()
_role_name = "SELECT construct or equivalent text() construct"
def subquery(self):
class HasCTERole(ReturnsRowsRole):
- pass
+ __slots__ = ()
class IsCTERole(SQLRole):
+ __slots__ = ()
_role_name = "CTE object"
class CompoundElementRole(AllowsLambdaRole, SQLRole):
"""SELECT statements inside a CompoundSelect, e.g. UNION, EXTRACT, etc."""
+ __slots__ = ()
_role_name = (
"SELECT construct for inclusion in a UNION or other set construct"
)
# TODO: are we using this?
class DMLRole(StatementRole):
- pass
+ __slots__ = ()
class DMLTableRole(FromClauseRole):
+ __slots__ = ()
_role_name = "subject table for an INSERT, UPDATE or DELETE"
class DMLColumnRole(SQLRole):
+ __slots__ = ()
_role_name = "SET/VALUES column expression or string key"
class DMLSelectRole(SQLRole):
"""A SELECT statement embedded in DML, typically INSERT from SELECT"""
+ __slots__ = ()
_role_name = "SELECT statement or equivalent textual object"
class DDLRole(StatementRole):
- pass
+ __slots__ = ()
class DDLExpressionRole(StructuralRole):
+ __slots__ = ()
_role_name = "SQL expression element for DDL constraint"
class DDLConstraintColumnRole(SQLRole):
+ __slots__ = ()
_role_name = "String column name or column expression for DDL constraint"
class DDLReferredColumnRole(DDLConstraintColumnRole):
+ __slots__ = ()
_role_name = (
"String column name or Column object for DDL foreign key constraint"
)
class HasCopyInternals:
+ __slots__ = ()
+
def _clone(self, **kw):
raise NotImplementedError()
"""
+ __slots__ = ()
+
def __class_getitem__(cls, key):
# allow generic classes in py3.9+
return cls
from sqlalchemy.dialects.postgresql import ExcludeConstraint
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.dialects.postgresql import TSRANGE
+from sqlalchemy.dialects.postgresql.base import _ColonCast
from sqlalchemy.dialects.postgresql.base import PGDialect
from sqlalchemy.dialects.postgresql.psycopg2 import PGDialect_psycopg2
from sqlalchemy.orm import aliased
__dialect__ = postgresql.dialect()
+ def test_colon_cast_is_slots(self):
+
+ c1 = _ColonCast(column("q"), String(50))
+
+ assert not hasattr(c1, "__dict__")
+
+ self.assert_compile(c1, "q::VARCHAR(50)")
+
def test_update_returning(self):
dialect = postgresql.dialect()
table1 = table(
from sqlalchemy.sql import util as sql_util
from sqlalchemy.sql.elements import BooleanClauseList
from sqlalchemy.sql.elements import ColumnElement
+from sqlalchemy.sql.elements import CompilerColumnElement
from sqlalchemy.sql.expression import ClauseElement
from sqlalchemy.sql.expression import ClauseList
from sqlalchemy.sql.expression import HasPrefixes
class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = "default"
+ def test_compiler_column_element_is_slots(self):
+ class SomeColThing(CompilerColumnElement):
+ __slots__ = ("name",)
+ __visit_name__ = "some_col_thing"
+
+ def __init__(self, name):
+ self.name = name
+
+ c1 = SomeColThing("some name")
+ eq_(c1.name, "some name")
+ assert not hasattr(c1, "__dict__")
+
+ def test_compile_label_is_slots(self):
+
+ c1 = compiler._CompileLabel(column("q"), "somename")
+
+ eq_(c1.name, "somename")
+ assert not hasattr(c1, "__dict__")
+
def test_attribute_sanity(self):
assert hasattr(table1, "c")
assert hasattr(table1.select().subquery(), "c")
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
