我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用sqlalchemy.sql.column()。
def columns(self): """The set of columns exported by this :class:`.FunctionElement`. Function objects currently have no result column names built in; this method returns a single-element column collection with an anonymously named column. An interim approach to providing named columns for a function as a FROM clause is to build a :func:`.select` with the desired columns:: from sqlalchemy.sql import column stmt = select([column('x'), column('y')]).\ select_from(func.myfunction()) """ return ColumnCollection(self.label(None))
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() ClauseElement._cloned_set._reset(c) ColumnElement.comparator._reset(c) # 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 _find_columns(clause): """locate Column objects within the given expression.""" cols = util.column_set() traverse(clause, {}, {'column': cols.add}) return cols # there is some inconsistency here between the usage of # inspect() vs. checking for Visitable and __clause_element__. # Ideally all functions here would derive from inspect(), # however the inspect() versions add significant callcount # overhead for critical functions like _interpret_as_column_or_from(). # Generally, the column-based functions are more performance critical # and are fine just checking for __clause_element__(). It is only # _interpret_as_from() where we'd like to be able to receive ORM entities # that have no defined namespace, hence inspect() is needed there.
def upgrade(): task = table('task', column('id'), column('info') ) conn = op.get_bind() query = select([task.c.id, task.c.info]) tasks = conn.execute(query) update_values = [] for row in tasks: info_data = row.info info_dict = json.loads(info_data) if info_dict.get('n_answers'): del info_dict['n_answers'] update_values.append({'task_id': row.id, 'new_info': json.dumps(info_dict)}) task_update = task.update().\ where(task.c.id == bindparam('task_id')).\ values(info=bindparam('new_info')) if len(update_values) > 0: conn.execute(task_update, update_values)
def downgrade(): task = table('task', column('id'), column('info'), column('n_answers') ) conn = op.get_bind() query = select([task.c.id, task.c.info, task.c.n_answers]) tasks = conn.execute(query) update_values = [] for row in tasks: info_data = row.info info_dict = json.loads(info_data) info_dict['n_answers'] = row.n_answers update_values.append({'task_id': row.id, 'new_info': json.dumps(info_dict)}) task_update = task.update().\ where(task.c.id == bindparam('task_id')).\ values(info=bindparam('new_info')) if len(update_values) > 0: conn.execute(task_update, update_values)
def test_row_w_scalar_select(self): """test that a scalar select as a column is returned as such and that type conversion works OK. (this is half a SQLAlchemy Core test and half to catch database backends that may have unusual behavior with scalar selects.) """ datetable = self.tables.has_dates s = select([datetable.alias('x').c.today]).as_scalar() s2 = select([datetable.c.id, s.label('somelabel')]) row = config.db.execute(s2).first() eq_(row['somelabel'], datetime.datetime(2006, 5, 12, 12, 0, 0))
def define_tables(cls, metadata): cls.tables.percent_table = Table('percent%table', metadata, Column("percent%", Integer), Column( "spaces % more spaces", Integer), ) cls.tables.lightweight_percent_table = sql.table( 'percent%table', sql.column("percent%"), sql.column("spaces % more spaces") )
def alias(self, name=None, flat=False): """Produce a :class:`.Alias` construct against this :class:`.FunctionElement`. This construct wraps the function in a named alias which is suitable for the FROM clause. e.g.:: from sqlalchemy.sql import column stmt = select([column('data')]).select_from( func.unnest(Table.data).alias('data_view') ) Would produce: .. sourcecode:: sql SELECT data FROM unnest(sometable.data) AS data_view .. versionadded:: 0.9.8 The :meth:`.FunctionElement.alias` method is now supported. Previously, this method's behavior was undefined and did not behave consistently across versions. """ return Alias(self, name)
def params(self, *optionaldict, **kwargs): """Return a copy with :func:`bindparam()` elements replaced. Returns a copy of this ClauseElement with :func:`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 expression(self): """Return a column expression. Part of the inspection interface; returns self. """ return self
def _compare_name_for_result(self, other): """Return True if the given column element compares to this one when targeting within a result row.""" return hasattr(other, 'name') and hasattr(self, 'name') and \ other.name == self.name
def compare(self, other, use_proxies=False, equivalents=None, **kw): """Compare this ColumnElement to another. Special arguments understood: :param use_proxies: when True, consider two columns that share a common base column as equivalent (i.e. shares_lineage()) :param equivalents: a dictionary of columns as keys mapped to sets of columns. If the given "other" column is present in this dictionary, if any of the columns in the corresponding set() pass the comparison test, the result is True. This is used to expand the comparison to other columns that may be known to be equivalent to this one via foreign key or other criterion. """ to_compare = (other, ) if equivalents and other in equivalents: to_compare = equivalents[other].union(to_compare) for oth in to_compare: if use_proxies and self.shares_lineage(oth): return True elif hash(oth) == hash(self): return True else: return False
def literal_column(text, type_=None): """Produce a :class:`.ColumnClause` object that has the :paramref:`.column.is_literal` flag set to True. :func:`.literal_column` is similar to :func:`.column`, except that it is more often used as a "standalone" column expression that renders exactly as stated; while :func:`.column` stores a string name that will be assumed to be part of a table and may be quoted as such, :func:`.literal_column` can be that, or any other arbitrary column-oriented expression. :param text: the text of the expression; can be any SQL expression. Quoting rules will not be applied. To specify a column-name expression which should be subject to quoting rules, use the :func:`column` function. :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` object which will provide result-set translation and additional expression semantics for this column. If left as None the type will be NullType. .. seealso:: :func:`.column` :func:`.text` :ref:`sqlexpression_literal_column` """ return ColumnClause(text, type_=type_, is_literal=True)
def _create_nullsfirst(cls, column): """Produce the ``NULLS FIRST`` modifier for an ``ORDER BY`` expression. :func:`.nullsfirst` is intended to modify the expression produced by :func:`.asc` or :func:`.desc`, and indicates how NULL values should be handled when they are encountered during ordering:: from sqlalchemy import desc, nullsfirst stmt = select([users_table]).\\ order_by(nullsfirst(desc(users_table.c.name))) The SQL expression from the above would resemble:: SELECT id, name FROM user ORDER BY name DESC NULLS FIRST Like :func:`.asc` and :func:`.desc`, :func:`.nullsfirst` is typically invoked from the column expression itself using :meth:`.ColumnElement.nullsfirst`, rather than as its standalone function version, as in:: stmt = (select([users_table]). order_by(users_table.c.name.desc().nullsfirst()) ) .. seealso:: :func:`.asc` :func:`.desc` :func:`.nullslast` :meth:`.Select.order_by` """ return UnaryExpression( _literal_as_label_reference(column), modifier=operators.nullsfirst_op, wraps_column_expression=False)
def _create_nullslast(cls, column): """Produce the ``NULLS LAST`` modifier for an ``ORDER BY`` expression. :func:`.nullslast` is intended to modify the expression produced by :func:`.asc` or :func:`.desc`, and indicates how NULL values should be handled when they are encountered during ordering:: from sqlalchemy import desc, nullslast stmt = select([users_table]).\\ order_by(nullslast(desc(users_table.c.name))) The SQL expression from the above would resemble:: SELECT id, name FROM user ORDER BY name DESC NULLS LAST Like :func:`.asc` and :func:`.desc`, :func:`.nullslast` is typically invoked from the column expression itself using :meth:`.ColumnElement.nullslast`, rather than as its standalone function version, as in:: stmt = select([users_table]).\\ order_by(users_table.c.name.desc().nullslast()) .. seealso:: :func:`.asc` :func:`.desc` :func:`.nullsfirst` :meth:`.Select.order_by` """ return UnaryExpression( _literal_as_label_reference(column), modifier=operators.nullslast_op, wraps_column_expression=False)
def _create_desc(cls, column): """Produce a descending ``ORDER BY`` clause element. e.g.:: from sqlalchemy import desc stmt = select([users_table]).order_by(desc(users_table.c.name)) will produce SQL as:: SELECT id, name FROM user ORDER BY name DESC The :func:`.desc` function is a standalone version of the :meth:`.ColumnElement.desc` method available on all SQL expressions, e.g.:: stmt = select([users_table]).order_by(users_table.c.name.desc()) :param column: A :class:`.ColumnElement` (e.g. scalar SQL expression) with which to apply the :func:`.desc` operation. .. seealso:: :func:`.asc` :func:`.nullsfirst` :func:`.nullslast` :meth:`.Select.order_by` """ return UnaryExpression( _literal_as_label_reference(column), modifier=operators.desc_op, wraps_column_expression=False)
def _create_distinct(cls, expr): """Produce an column-expression-level unary ``DISTINCT`` clause. This applies the ``DISTINCT`` keyword to an individual column expression, and is typically contained within an aggregate function, as in:: from sqlalchemy import distinct, func stmt = select([func.count(distinct(users_table.c.name))]) The above would produce an expression resembling:: SELECT COUNT(DISTINCT name) FROM user The :func:`.distinct` function is also available as a column-level method, e.g. :meth:`.ColumnElement.distinct`, as in:: stmt = select([func.count(users_table.c.name.distinct())]) The :func:`.distinct` operator is different from the :meth:`.Select.distinct` method of :class:`.Select`, which produces a ``SELECT`` statement with ``DISTINCT`` applied to the result set as a whole, e.g. a ``SELECT DISTINCT`` expression. See that method for further information. .. seealso:: :meth:`.ColumnElement.distinct` :meth:`.Select.distinct` :data:`.func` """ expr = _literal_as_binds(expr) return UnaryExpression( expr, operator=operators.distinct_op, type_=expr.type, wraps_column_expression=False)
def __init__(self, func, partition_by=None, order_by=None): """Produce an :class:`.Over` object against a function. Used against aggregate or so-called "window" functions, for database backends that support window functions. E.g.:: from sqlalchemy import over over(func.row_number(), order_by='x') Would produce "ROW_NUMBER() OVER(ORDER BY x)". :param func: a :class:`.FunctionElement` construct, typically generated by :data:`~.expression.func`. :param partition_by: a column element or string, or a list of such, that will be used as the PARTITION BY clause of the OVER construct. :param order_by: a column element or string, or a list of such, that will be used as the ORDER BY clause of the OVER construct. This function is also available from the :data:`~.expression.func` construct itself via the :meth:`.FunctionElement.over` method. .. versionadded:: 0.7 """ self.func = func if order_by is not None: self.order_by = ClauseList( *util.to_list(order_by), _literal_as_text=_literal_as_label_reference) if partition_by is not None: self.partition_by = ClauseList( *util.to_list(partition_by), _literal_as_text=_literal_as_label_reference)
def _gen_label(self, name): t = self.table if self.is_literal: return None elif t is not None and t.named_with_column: if getattr(t, 'schema', None): label = t.schema.replace('.', '_') + "_" + \ t.name + "_" + name else: label = t.name + "_" + name # propagate name quoting rules for labels. if getattr(name, "quote", None) is not None: if isinstance(label, quoted_name): label.quote = name.quote else: label = quoted_name(label, name.quote) elif getattr(t.name, "quote", None) is not None: # can't get this situation to occur, so let's # assert false on it for now assert not isinstance(label, quoted_name) label = quoted_name(label, t.name.quote) # ensure the label name doesn't conflict with that # of an existing column if label in t.c: _label = label counter = 1 while _label in t.c: _label = label + "_" + str(counter) counter += 1 label = _label return _as_truncated(label) else: return name
def _select_iterables(elements): """expand tables into individual columns in the given list of column expressions. """ return itertools.chain(*[c._select_iterable for c in elements])
def _corresponding_column_or_error(fromclause, column, require_embedded=False): c = fromclause.corresponding_column(column, require_embedded=require_embedded) if c is None: raise exc.InvalidRequestError( "Given column '%s', attached to table '%s', " "failed to locate a corresponding column from table '%s'" % (column, getattr(column, 'table', None), fromclause.description) ) return c
def to_index(self, migration_context=None): if self._orig_index is not None: return self._orig_index schema_obj = schemaobj.SchemaObjects(migration_context) # need a dummy column name here since SQLAlchemy # 0.7.6 and further raises on Index with no columns return schema_obj.index( self.index_name, self.table_name, ['x'], schema=self.schema)
def __init__(self, table_name, column, schema=None): super(AddColumnOp, self).__init__(table_name, schema=schema) self.column = column
def reverse(self): return DropColumnOp.from_column_and_tablename( self.schema, self.table_name, self.column)
def to_diff_tuple(self): return ("add_column", self.schema, self.table_name, self.column)
def batch_add_column(cls, operations, column): """Issue an "add column" instruction using the current batch migration context. .. seealso:: :meth:`.Operations.add_column` """ op = cls( operations.impl.table_name, column, schema=operations.impl.schema ) return operations.invoke(op)
def to_column(self, migration_context=None): if self._orig_column is not None: return self._orig_column schema_obj = schemaobj.SchemaObjects(migration_context) return schema_obj.column(self.column_name, NULLTYPE)
def batch_drop_column(cls, operations, column_name): """Issue a "drop column" instruction using the current batch migration context. .. seealso:: :meth:`.Operations.drop_column` """ op = cls( operations.impl.table_name, column_name, schema=operations.impl.schema) return operations.invoke(op)