我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlalchemy.sql.table()。
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 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 drop_table(self, name, **kw): """Issue a "drop table" instruction using the current migration context. e.g.:: drop_table("accounts") :param name: Name of the table :param schema: Optional schema name to operate within. .. versionadded:: 0.4.0 :param \**kw: Other keyword arguments are passed to the underlying :class:`sqlalchemy.schema.Table` object created for the command. """ self.impl.drop_table( self._table(name, **kw) )
def upgrade(): ip_policy_cidrs = table('quark_ip_policy_cidrs', column('id', sa.String(length=36)), column('first_ip', INET()), column('last_ip', INET()), column('cidr', sa.String(length=64))) connection = op.get_bind() # 1. Retrieve all ip_policy_cidr rows. results = connection.execute( select([ip_policy_cidrs.c.id, ip_policy_cidrs.c.cidr]) ).fetchall() # 2. Populate first_ip, last_ip for each IP Policy CIDR. for ippc in results: net = netaddr.IPNetwork(ippc["cidr"]).ipv6() connection.execute(ip_policy_cidrs.update().values( first_ip=net.first, last_ip=net.last).where( ip_policy_cidrs.c.id == ippc["id"]))
def setUp(self): super(Test552b213c2b8c, self).setUp() alembic_command.upgrade(self.config, '45a07fac3d38') self.ip_policy = table( 'quark_ip_policy', column('id', sa.String(length=36)), column('tenant_id', sa.String(length=255)), column('created_at', sa.DateTime())) self.ip_policy_cidrs = table( 'quark_ip_policy_cidrs', column('id', sa.String(length=36)), column('created_at', sa.DateTime()), column('ip_policy_id', sa.String(length=36)), column('cidr', sa.String(length=64))) self.subnets = table( 'quark_subnets', column('id', sa.String(length=36)), column('tenant_id', sa.String(length=255)), column('_cidr', sa.String(length=64)), column('ip_policy_id', sa.String(length=36)))
def rename_table(self, old_table_name, new_table_name, schema=None): """Emit an ALTER TABLE to rename a table. :param old_table_name: old name. :param new_table_name: new name. :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. .. versionadded:: 0.7.0 'schema' can now accept a :class:`~sqlalchemy.sql.elements.quoted_name` construct. """ self.impl.rename_table( old_table_name, new_table_name, schema=schema )
def drop_table(self, name, **kw): """Issue a "drop table" instruction using the current migration context. e.g.:: drop_table("accounts") :param name: Name of the table :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. .. versionadded:: 0.7.0 'schema' can now accept a :class:`~sqlalchemy.sql.elements.quoted_name` construct. :param \**kw: Other keyword arguments are passed to the underlying :class:`sqlalchemy.schema.Table` object created for the command. """ self.impl.drop_table( self._table(name, **kw) )
def drop_index(self, name, table_name=None, schema=None): """Issue a "drop index" instruction using the current migration context. e.g.:: drop_index("accounts") :param name: name of the index. :param table_name: name of the owning table. Some backends such as Microsoft SQL Server require this. :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. .. versionadded:: 0.7.0 'schema' can now accept a :class:`~sqlalchemy.sql.elements.quoted_name` construct. """ # need a dummy column name here since SQLAlchemy # 0.7.6 and further raises on Index with no columns self.impl.drop_index( self._index(name, table_name, ['x'], schema=schema) )
def upgrade(): """Insert fixtures for models, tests purpose only""" users = [generate_users(i) for i in range(1, 52)] articles = [generate_articles(i, 50) for i in range(1, 200)] article = table( 'article', sa.Column('id', sa.Integer(), nullable=False, primary_key=True, autoincrement=True), sa.Column('title', sa.String(length=100), nullable=False), sa.Column('content', sa.Text(), nullable=False), sa.Column('user_id', sa.Integer(), nullable=True), ) user = table( 'user', sa.Column('id', sa.Integer(), nullable=False, primary_key=True, autoincrement=True), sa.Column('username', sa.String(length=80), nullable=False), sa.Column('email', sa.String(length=120), nullable=False), ) op.bulk_insert(user, users) op.bulk_insert(article, articles)
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 __init__(self, *clauses, **kw): """Return a :class:`.Tuple`. Main usage is to produce a composite IN construct:: from sqlalchemy import tuple_ tuple_(table.c.col1, table.c.col2).in_( [(1, 2), (5, 12), (10, 19)] ) .. warning:: The composite IN construct is not supported by all backends, and is currently known to work on Postgresql and MySQL, but not SQLite. Unsupported backends will raise a subclass of :class:`~sqlalchemy.exc.DBAPIError` when such an expression is invoked. """ clauses = [_literal_as_binds(c) for c in clauses] self._type_tuple = [arg.type for arg in clauses] self.type = kw.pop('type_', self._type_tuple[0] if self._type_tuple else type_api.NULLTYPE) super(Tuple, self).__init__(*clauses, **kw)
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 _get_table(self): return self.__dict__['table']
def _set_table(self, table): self._memoized_property.expire_instance(self) self.__dict__['table'] = table
def _from_objects(self): t = self.table if t is not None: return [t] else: return []
def _render_label_in_columns_clause(self): return self.table is not None
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 __init__(self, element, values): Annotated.__init__(self, element, values) ColumnElement.comparator._reset(self) for attr in ('name', 'key', 'table'): if self.__dict__.get(attr, False) is None: self.__dict__.pop(attr)
def table(self): """pull 'table' from parent, if not present""" return self._Annotated__element.table
