• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python sqlalchemy.column函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中sqlalchemy.column函数的典型用法代码示例。如果您正苦于以下问题:Python column函数的具体用法?Python column怎么用?Python column使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了column函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: execute

    def execute(self, connection, filter_values):
        max_date_query = sqlalchemy.select([
            sqlalchemy.func.max(sqlalchemy.column('completed_on')).label('completed_on'),
            sqlalchemy.column('case_id').label('case_id')
        ]).select_from(sqlalchemy.table(self.table_name))

        if self.filters:
            for filter in self.filters:
                max_date_query.append_whereclause(filter.build_expression())

        max_date_query.append_group_by(
            sqlalchemy.column('case_id')
        )

        max_date_subquery = sqlalchemy.alias(max_date_query, 'max_date')

        asha_table = self.get_asha_table_name()
        checklist_query = sqlalchemy.select()
        for column in self.columns:
            checklist_query.append_column(column.build_column())

        checklist_query = checklist_query.where(
            sqlalchemy.literal_column('"{}".case_id'.format(asha_table)) == max_date_subquery.c.case_id
        ).where(
            sqlalchemy.literal_column('"{}".completed_on'.format(asha_table)) == max_date_subquery.c.completed_on
        ).select_from(sqlalchemy.table(asha_table))

        return connection.execute(checklist_query, **filter_values).fetchall()
开发者ID:kkrampa,项目名称:commcare-hq,代码行数:28,代码来源:sql_data.py


示例2: downgrade

def downgrade():
    # FIXME: this adds extraneous commas
    return
    log = sa.table('log', sa.column('type', sa.String), sa.column('msg', sa.String))
    rows = op.get_bind().execute(log.select().where(log.c.type == 'kick')).fetchall()
    values = [{'old_msg': x.msg, 'msg': x.msg.replace(' ', ',', 1)} for x in rows]
    op.get_bind().execute(log.update().where(log.c.msg == sa.bindparam('old_msg')).values(msg=sa.bindparam('msg')), values)
开发者ID:Polarcraft,项目名称:KbveBot,代码行数:7,代码来源:3614b38ddf9_kick_logging_change.py


示例3: test_no_table_needs_pl

    def test_no_table_needs_pl(self):
        Subset = self.classes.Subset

        selectable = select([column("x"), column("y"), column("z")]).alias()
        assert_raises_message(
            sa.exc.ArgumentError, "could not assemble any primary key columns", mapper, Subset, selectable
        )
开发者ID:FluxIX,项目名称:sqlalchemy,代码行数:7,代码来源:test_selectable.py


示例4: datatype_records_to_subset_and_migrate

def datatype_records_to_subset_and_migrate(likechars):
    stmt_for_pkeys = conn_popler_2.execute(
        select(
            from_obj=Maintable,
            columns=[
                column('lter_proj_site'),
                column('samplingprotocol')
            ]).
        where(
            column('samplingprotocol').like(
                '%{}%'.format(likechars))
        )
    )
    data = DataFrame(stmt_for_pkeys.fetchall())
    data.columns = stmt_for_pkeys.keys()

    records_to_get = data['lter_proj_site'].values.tolist()

    stmt_for_records = conn_popler_2.execute(
        select(
            from_table=Rawtable,
        ).
        where(column('lter_proj_site').in_(records_to_get)).
        order_by('sampleid')
    )
    data2 = DataFrame(stmt_for_records.fetchall())
    data2.columns = stmt_for_records.keys()
    data2.drop('individ', axis=1, inplace=True)
开发者ID:bibsian,项目名称:database-development,代码行数:28,代码来源:database_migration_refactor_unfinished.py


示例5: upgrade

def upgrade():
    ### commands auto generated by Alembic - please adjust! ###
    subsc = op.create_table(
        'subscription',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('user_id', sa.Integer(), sa.ForeignKey('user.id'), nullable=False),
        sa.Column('feed_id', sa.Integer(), sa.ForeignKey('feed.id'), nullable=False),
        sa.Column('name', sa.String(length=256), nullable=True),
        sa.Column('tags', sa.String(length=256), nullable=True),
        sa.PrimaryKeyConstraint('id')
    )

    feed = sa.table(
        'feed',
        sa.column('id', sa.Integer()),
        sa.column('name', sa.String()))

    u2f = sa.table(
        'users_to_feeds',
        sa.column('user_id', sa.Integer()),
        sa.column('feed_id', sa.Integer()))

    values = sa.select(
        [u2f.c.user_id, u2f.c.feed_id, feed.c.name]
    ).select_from(
        u2f.join(feed, feed.c.id == u2f.c.feed_id)
    )

    op.execute(subsc.insert().from_select(
        ['user_id', 'feed_id', 'name'], values))

    op.drop_table('users_to_feeds')
开发者ID:Lancey6,项目名称:woodwind,代码行数:32,代码来源:564f5a5061f_.py


示例6: test_unconsumed_names_kwargs

 def test_unconsumed_names_kwargs(self):
     t = table("t", column("x"), column("y"))
     assert_raises_message(
         exc.CompileError,
         "Unconsumed column names: z",
         t.insert().values(x=5, z=5).compile,
     )
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:7,代码来源:test_insert.py


示例7: test_legacy_typemap

    def test_legacy_typemap(self):
        table1 = table(
            "mytable",
            column("myid", Integer),
            column("name", String),
            column("description", String),
        )
        with testing.expect_deprecated(
            "The text.typemap parameter is deprecated"
        ):
            t = text(
                "select id, name from user",
                typemap=dict(id=Integer, name=String),
            )

        stmt = select([table1.c.myid]).select_from(
            table1.join(t, table1.c.myid == t.c.id)
        )
        compiled = stmt.compile()
        eq_(
            compiled._create_result_map(),
            {
                "myid": (
                    "myid",
                    (table1.c.myid, "myid", "myid"),
                    table1.c.myid.type,
                )
            },
        )
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:29,代码来源:test_deprecations.py


示例8: filtra_por_partido

def filtra_por_partido(_partido):
    """retorna dados do partido passado por parâmetro"""
    _select = select([column('nome_doador'), column('nome_candidato'), column('partido'), column('valor')]).\
        select_from(dados_agrup_doadores).where(column('partido') == _partido)
    # retorna resultado filtrado e count do resuldato

    return [dict(doador=x[0], candidato=x[1], partido=x[2], valor=x[3]) for x in _select.execute()],\
           list(_select.count().execute())[0][0]
开发者ID:flaviomeira,项目名称:CandidaturasDB,代码行数:8,代码来源:db_engine.py


示例9: upgrade

def upgrade():
    log = sa.table('log', sa.column('type', sa.String), sa.column('msg', sa.String))
    rows = op.get_bind().execute(log.select().where(log.c.type == 'kick').where(log.c.msg.like('%,%'))).fetchall()
    rows = [x for x in rows if ',' in x.msg and x.msg.find(',') < x.msg.find(' ')]
    if not rows:
        return
    values = [{'old_msg': x.msg, 'msg': x.msg.replace(',', ' ', 1)} for x in rows]
    op.get_bind().execute(log.update().where(log.c.msg == sa.bindparam('old_msg')).values(msg=sa.bindparam('msg')), values)
开发者ID:N6UDP,项目名称:cslbot,代码行数:8,代码来源:3614b38ddf9_kick_logging_change.py


示例10: test_unconsumed_names_kwargs_w_keys

    def test_unconsumed_names_kwargs_w_keys(self):
        t = table("t", column("x"), column("y"))

        assert_raises_message(
            exc.CompileError,
            "Unconsumed column names: j",
            t.update().values(x=5, j=7).compile,
            column_keys=['j']
        )
开发者ID:eoghanmurray,项目名称:sqlalchemy,代码行数:9,代码来源:test_update.py


示例11: test_compile_with_one_unnamed_table

def test_compile_with_one_unnamed_table():
    t = ibis.table([('a', 'string')])
    s = ibis.table([('b', 'string')], name='s')
    join = t.join(s, t.a == s.b)
    result = ibis.sqlite.compile(join)
    sqla_t = sa.table('t0', sa.column('a', sa.String)).alias('t0')
    sqla_s = sa.table('s', sa.column('b', sa.String)).alias('t1')
    sqla_join = sqla_t.join(sqla_s, sqla_t.c.a == sqla_s.c.b)
    expected = sa.select([sqla_t.c.a, sqla_s.c.b]).select_from(sqla_join)
    assert str(result) == str(expected)
开发者ID:deepfield,项目名称:ibis,代码行数:10,代码来源:test_functions.py


示例12: upgrade

def upgrade():
    source = sa.table('source', sa.column('id'), sa.column('updated', sa.DateTime))
    dt = sa.bindparam('dt', UPDATED)
    touch = sa.update(source, bind=op.get_bind())\
        .where(source.c.id == sa.bindparam('id_'))\
        .where(source.c.updated < dt)\
        .values(updated=dt)

    for id_ in IDS:
        touch.execute(id_=id_)
开发者ID:clld,项目名称:wals3,代码行数:10,代码来源:3be1ca59b931_touch_updated_references.py


示例13: test_compare_labels

    def test_compare_labels(self):
        is_true(column("q").label(None).compare(column("q").label(None)))

        is_false(column("q").label("foo").compare(column("q").label(None)))

        is_false(column("q").label(None).compare(column("q").label("foo")))

        is_false(column("q").label("foo").compare(column("q").label("bar")))

        is_true(column("q").label("foo").compare(column("q").label("foo")))
开发者ID:monetate,项目名称:sqlalchemy,代码行数:10,代码来源:test_compare.py


示例14: test_fewer_cols_than_sql_positional

    def test_fewer_cols_than_sql_positional(self):
        c1, c2 = column('q'), column('p')
        stmt = text("select a, b, c, d from text1").columns(c1, c2)

        # no warning as this can be similar for non-positional
        result = testing.db.execute(stmt)
        row = result.first()

        eq_(row[c1], "a1")
        eq_(row["c"], "c1")
开发者ID:mattastica,项目名称:sqlalchemy,代码行数:10,代码来源:test_resultset.py


示例15: test_dupe_col_obj

    def test_dupe_col_obj(self):
        c1, c2, c3 = column('q'), column('p'), column('r')
        stmt = text("select a, b, c, d from text1").columns(c1, c2, c3, c2)

        assert_raises_message(
            exc.InvalidRequestError,
            "Duplicate column expression requested in "
            "textual SQL: <.*.ColumnClause.*; p>",
            testing.db.execute, stmt
        )
开发者ID:mattastica,项目名称:sqlalchemy,代码行数:10,代码来源:test_resultset.py


示例16: test_unconsumed_names_values_dict

    def test_unconsumed_names_values_dict(self):
        t = table("t", column("x"), column("y"))
        t2 = table("t2", column("q"), column("z"))

        assert_raises_message(
            exc.CompileError,
            "Unconsumed column names: j",
            t.update().values(x=5, j=7).values({t2.c.z: 5}).
            where(t.c.x == t2.c.q).compile,
        )
开发者ID:eoghanmurray,项目名称:sqlalchemy,代码行数:10,代码来源:test_update.py


示例17: test_bindparam_name_no_consume_error

    def test_bindparam_name_no_consume_error(self):
        t = table("t", column("x"), column("y"))
        # bindparam names don't get counted
        i = t.insert().values(x=3 + bindparam("x2"))
        self.assert_compile(i, "INSERT INTO t (x) VALUES ((:param_1 + :x2))")

        # even if in the params list
        i = t.insert().values(x=3 + bindparam("x2"))
        self.assert_compile(
            i, "INSERT INTO t (x) VALUES ((:param_1 + :x2))", params={"x2": 1}
        )
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:11,代码来源:test_insert.py


示例18: test_functions_args_noname

    def test_functions_args_noname(self):
        class myfunc(FunctionElement):
            pass

        @compiles(myfunc)
        def visit_myfunc(element, compiler, **kw):
            return "myfunc%s" % (compiler.process(element.clause_expr, **kw),)

        self.assert_compile(myfunc(), "myfunc()")

        self.assert_compile(myfunc(column("x"), column("y")), "myfunc(x, y)")
开发者ID:monetate,项目名称:sqlalchemy,代码行数:11,代码来源:test_compiler.py


示例19: __table_args__

 def __table_args__(cls):
     return (
         Index(
             'recipient_address_in_group', 'address', 'group',
             unique=True, postgresql_where=column('group') != None
         ),
         Index(
             'recipient_address_without_group', 'address',
             unique=True, postgresql_where=column('group') == None
         ),
     )
开发者ID:OneGov,项目名称:onegov.newsletter,代码行数:11,代码来源:models.py


示例20: test_no_tables

    def test_no_tables(self):
        Subset = self.classes.Subset

        selectable = select([column("x"), column("y"), column("z")]).alias()
        mapper(Subset, selectable, primary_key=[selectable.c.x])

        self.assert_compile(
            Session().query(Subset),
            "SELECT anon_1.x AS anon_1_x, anon_1.y AS anon_1_y, "
            "anon_1.z AS anon_1_z FROM (SELECT x, y, z) AS anon_1",
            use_default_dialect=True
        )
开发者ID:anti-social,项目名称:sqlalchemy,代码行数:12,代码来源:test_selectable.py



注:本文中的sqlalchemy.column函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python sqlalchemy.create_engine函数代码示例发布时间:2022-05-27
下一篇:
Python sqlalchemy.cast函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap