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

Python sql.bindparam函数代码示例

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

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



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

示例1: gdft_xhr

def gdft_xhr(request):
    if request.method == 'POST':
        form = GenDev_FailType(request.POST)
        if form.is_valid():
            conn = engine.connect()


            d = select([device.brand_name, func.count(device.MDR_report_key),
                problem_code.problem_description, master_record.manufacturer_name],
                and_(
                    device.MDR_report_key==master_record.MDR_report_key,
                    master_record.event_type==bindparam("event_type"),
                    device_problem.MDR_report_key==master_record.MDR_report_key,
                    device_problem.problem_code_key==problem_code.device_problem_code,
                    device.generic_name.contains(bindparam("generic_name")),
                    master_record.date_recieved > bindparam("date_after"),
                    )).group_by(device_problem.problem_code_key)

            res = conn.execute(d, generic_name=form.cleaned_data["gendev"], event_type=form.cleaned_data["failtype"],
                    date_after=form.cleaned_data["date_after"])
            li = res.fetchall()
            li = sorted(li, key=lambda x: x[1], reverse=True)
            s= ""
            for i in li:
                s = s + str(i) + "<br>"
            return HttpResponse(s)

    return HttpResponse("Error in form")
开发者ID:NSkelsey,项目名称:research,代码行数:28,代码来源:views.py


示例2: has_table

 def has_table(self, connection, table_name, schema=None):
     # seems like case gets folded in pg_class...
     if schema is None:
         cursor = connection.execute(
             sql.text(
                 "select name "
                 "from sys.tables "
                 "where system = false "
                 "and type = 0 "
                 "and name=:name",
                 bindparams=[
                     sql.bindparam('name', util.text_type(table_name),
                                   type_=sqltypes.Unicode)]
             )
         )
     else:
         cursor = connection.execute(
             sql.text(
                 "SELECT tables.name "
                 "FROM sys.tables, sys.schemas "
                 "WHERE tables.system = FALSE "
                 "AND tables.schema_id = schemas.id "
                 "AND type = 0 "
                 "AND tables.name = :name "
                 "AND schemas.name = :schema",
                 bindparams=[
                     sql.bindparam('name',
                                   util.text_type(table_name),
                                   type_=sqltypes.Unicode),
                     sql.bindparam('schema',
                                   util.text_type(schema),
                                   type_=sqltypes.Unicode)]
             )
         )
     return bool(cursor.first())
开发者ID:gijzelaerr,项目名称:sqlalchemy-monetdb,代码行数:35,代码来源:dialect.py


示例3: init

    def init(self, session, api):
        self.session = session
        self.api = api

        # compile queries
        self.command_queries = {}
        self.command_queries['character'] = \
            session.query(Command).\
            filter(Command.actor_id == bindparam('actor_id'))

        self.command_queries['children'] = \
            session.query(Actor, Command).\
            filter(Actor.id == bindparam('actor_id')).\
            join(Actor.parent, aliased=True).\
            filter(Command.actor_id == Actor.id)

        location = aliased(Actor)
        self.command_queries['location'] = \
            session.query(Command).\
            join(location).\
            join(Actor, location.id == Actor.parent_id).\
            filter(Actor.id == bindparam('actor_id'))

        location = aliased(Actor)
        children = aliased(Actor)
        self.command_queries['location_children'] = \
            session.query(Command).\
            join(children).\
            join(location, location.id == children.parent_id).\
            join(Actor, location.id == Actor.parent_id).\
            filter(Actor.id == bindparam('actor_id')).\
            filter(Command.actor_id == children.id)

        self.command_precedence = ['character', 'children', 'location', 'location_children']
开发者ID:nijotz,项目名称:goat-tower,代码行数:34,代码来源:engine.py


示例4: __init__

    def __init__(self, config):
        self._orm_engine = engine_from_config(config["database"]["orm"], prefix="")

        metadata = MetaData()

        aa = models.create_alarm_action_model(metadata).alias("aa")
        nm = models.create_notification_method_model(metadata).alias("nm")
        nmt = models.create_notification_method_type_model(metadata).alias("nmt")
        a = models.create_alarm_model(metadata).alias("a")

        self._orm_query = (
            select([nm.c.id, nm.c.type, nm.c.name, nm.c.address, nm.c.period])
            .select_from(aa.join(nm, aa.c.action_id == nm.c.id))
            .where(
                and_(
                    aa.c.alarm_definition_id == bindparam("alarm_definition_id"),
                    aa.c.alarm_state == bindparam("alarm_state"),
                )
            )
        )

        self._orm_get_alarm_state = select([a.c.state]).where(a.c.id == bindparam("alarm_id"))

        self._orm_nmt_query = select([nmt.c.name])

        self._orm_get_notification = select([nm.c.name, nm.c.type, nm.c.address, nm.c.period]).where(
            nm.c.id == bindparam("notification_id")
        )

        self._orm = None
开发者ID:sapcc,项目名称:monasca-notification,代码行数:30,代码来源:orm_repo.py


示例5: has_table

 def has_table(self, connection, table_name, schema=None):
     # seems like case gets folded in pg_class...
     if schema is None:
         cursor = connection.execute(
             sql.text(
             "select relname from pg_class c join pg_namespace n on "
             "n.oid=c.relnamespace where n.nspname=current_schema() and "
             "lower(relname)=:name",
             bindparams=[
                     sql.bindparam('name', unicode(table_name.lower()),
                     type_=sqltypes.Unicode)]
             )
         )
     else:
         cursor = connection.execute(
             sql.text(
             "select relname from pg_class c join pg_namespace n on "
             "n.oid=c.relnamespace where n.nspname=:schema and "
             "lower(relname)=:name",
                 bindparams=[
                     sql.bindparam('name', 
                     unicode(table_name.lower()), type_=sqltypes.Unicode),
                     sql.bindparam('schema', 
                     unicode(schema), type_=sqltypes.Unicode)] 
             )
         )
     return bool(cursor.first())
开发者ID:advatar,项目名称:thevault,代码行数:27,代码来源:base.py


示例6: _save_concepts

    def _save_concepts(cls, concepts):
        r'''Prepare to save concepts in a temporary table.

        >>> concepts = dict(
        ...     names=['apples', 'bananas', 'cherries'],
        ...     keys=[r'\\tk\a', r'\\tk\b', r'\\tk\c'])
        >>> tmp, ins, bind = DataExtract._save_concepts(concepts)

        >>> print tmp.delete()
        DELETE FROM global_temp_fact_param_table

        >>> print ins
        ... # doctest: +NORMALIZE_WHITESPACE
        INSERT INTO global_temp_fact_param_table (char_param1, char_param2)
        VALUES (:path, :name)

        >>> sorted(bind[0].keys())
        ['name', 'path']

        >>> sorted(bind[0].values())
        ['\\a', 'apples']

        '''
        names = concepts['names']
        paths = I2B2MetaData.keys_to_paths(concepts['keys'])
        bind = [dict(name=name,
                     path=path)
                for (path, name) in zip(paths, names)]
        tmp = i2b2_star.t_global_temp_fact_param_table
        ins = tmp.insert().values(char_param1=bindparam('path'),
                                  char_param2=bindparam('name'))
        return tmp, ins, bind
开发者ID:UTHSCSA-CIRD,项目名称:databuilder,代码行数:32,代码来源:dfbuilder.py


示例7: has_sequence

    def has_sequence(self, connection, sequence_name, schema=None):
        if schema is None:
            cursor = connection.execute(
                sql.text(
                    "SELECT relname FROM pg_class c join pg_namespace n on "
                    "n.oid=c.relnamespace where relkind='S' and "
                    "n.nspname=current_schema() "
                    "and lower(relname)=:name",
                    bindparams=[
                        sql.bindparam('name', unicode(sequence_name.lower()),
                        type_=sqltypes.Unicode)
                    ] 
                )
            )
        else:
            cursor = connection.execute(
                sql.text(
                "SELECT relname FROM pg_class c join pg_namespace n on "
                "n.oid=c.relnamespace where relkind='S' and "
                "n.nspname=:schema and lower(relname)=:name",
                bindparams=[
                    sql.bindparam('name', unicode(sequence_name.lower()),
                     type_=sqltypes.Unicode),
                    sql.bindparam('schema', 
                                unicode(schema), type_=sqltypes.Unicode)
                ]
            )
            )

        return bool(cursor.first())
开发者ID:advatar,项目名称:thevault,代码行数:30,代码来源:base.py


示例8: __init__

    def __init__(self):
        self._orm_engine = engine_from_config({
            'url': CONF.orm.url
        }, prefix='')

        metadata = MetaData()

        aa = models.create_alarm_action_model(metadata).alias('aa')
        nm = models.create_notification_method_model(metadata).alias('nm')
        nmt_insert = models.create_notification_method_type_model(metadata)
        nmt = nmt_insert.alias('nmt')
        a = models.create_alarm_model(metadata).alias('a')

        self._orm_query = select([nm.c.id, nm.c.type, nm.c.name, nm.c.address, nm.c.period])\
            .select_from(aa.join(nm, aa.c.action_id == nm.c.id))\
            .where(
                and_(aa.c.alarm_definition_id == bindparam('alarm_definition_id'),
                     aa.c.alarm_state == bindparam('alarm_state')))

        self._orm_get_alarm_state = select([a.c.state]).where(a.c.id == bindparam('alarm_id'))

        self._orm_nmt_query = select([nmt.c.name])

        self._orm_get_notification = select([nm.c.name, nm.c.type, nm.c.address, nm.c.period])\
            .where(nm.c.id == bindparam('notification_id'))

        self._orm_add_notification_type = insert(nmt_insert).values(name=bindparam('b_name'))

        self._orm = None
开发者ID:openstack,项目名称:monasca-notification,代码行数:29,代码来源:orm_repo.py


示例9: export_terms

        def export_terms(dest_star, job):
            v = self.variable_table(dest_star)
            keys = job.concepts['keys']
            names = job.concepts['names']
            paths = I2B2MetaData.keys_to_paths(keys)
            v.create(bind=dest_db)
            dest_db.execute(v.insert(),
                            [dict(id=id,
                                  item_key=key,
                                  concept_path=path,
                                  name_char=name,
                                  name=strip_counts(name))
                             for (id, (path, key, name)) in
                             enumerate(zip(paths, keys, names))])

            [(q_cd, result_cd), (q_md, result_md)] = job.term_info()
            cd = dest_star.tables['concept_dimension']
            md = dest_star.tables['modifier_dimension']

            values = lambda _: dict(concept_path=bindparam('concept_path'),
                                    concept_cd=bindparam('concept_cd'))
            tc.copy_in_chunks(dest_db, result_cd, cd,
                              'concept_dimension', [],
                              values=values)
            values = lambda _: dict(concept_path=bindparam('modifier_path'),
                                    concept_cd=bindparam('modifer_cd'))
            tc.copy_in_chunks(dest_db, result_cd, md,
                              'modifier_dimension', [])
开发者ID:UTHSCSA-CIRD,项目名称:databuilder,代码行数:28,代码来源:dfbuilder.py


示例10: get_query

def get_query(name, user_id=None):
    """Get the named pre-built query, sharding on user_id if given.

    This is a helper function to return an appropriate pre-built SQL query
    while taking sharding of the WBO table into account.  Call it with the
    name of the query and optionally the user_id on which to shard.
    """
    if user_id is None:
        table = wbo
    else:
        table = get_wbo_table(user_id)

    queries['ITEM_ID_COL_USER'] = and_(
        table.c.collection == bindparam('collection_id'),
        table.c.username == bindparam('user_id'),
        table.c.id == bindparam('item_id'),
        table.c.ttl > bindparam('ttl'))

    query = queries.get(name)
    if query is None:
        raise ValueError(name)

    if isinstance(query, str):
        if '%(wbo)s' in query:
            query = query % {'wbo': table.name}
        query = text(query)

    return query
开发者ID:rfk,项目名称:moz-server-storage,代码行数:28,代码来源:queries.py


示例11: has_type

 def has_type(self, connection, type_name, schema=None):
     bindparams = [
         sql.bindparam('typname',
             unicode(type_name), type_=sqltypes.Unicode),
         sql.bindparam('nspname',
             unicode(schema), type_=sqltypes.Unicode),
         ]
     if schema is not None:
         query = """
         SELECT EXISTS (
             SELECT * FROM pg_catalog.pg_type t, pg_catalog.pg_namespace n
             WHERE t.typnamespace = n.oid
             AND t.typname = :typname
             AND n.nspname = :nspname
             )
             """
     else:
         query = """
         SELECT EXISTS (
             SELECT * FROM pg_catalog.pg_type t
             WHERE t.typname = :typname
             AND pg_type_is_visible(t.oid)
             )
             """
     cursor = connection.execute(sql.text(query, bindparams=bindparams))
     return bool(cursor.scalar())
开发者ID:advatar,项目名称:thevault,代码行数:26,代码来源:base.py


示例12: get

 def get(self, database, query, qual):
     stmt = qualstat_getstatdata()
     c = inner_cc(stmt)
     stmt = stmt.alias()
     stmt = (stmt.select()
         .where((c.qualid == bindparam("qualid")))
         .where(stmt.c.occurences > 0)
         .column((c.queryid == bindparam("query")).label("is_my_query")))
     quals = list(self.execute(
         stmt,
         params={"query": query,
                 "from": self.get_argument("from"),
                 "to": self.get_argument("to"),
                 "qualid": qual}))
     my_qual = None
     other_queries = {}
     for qual in quals:
         if qual['is_my_query']:
             my_qual = resolve_quals(self.connect(database=database),
                                     [qual])[0]
         else:
             other_queries[qual['queryid']] = qual['query']
     if my_qual is None:
         self.render("xhr.html", content="nodata")
         return
     self.render("database/query/qualdetail.html",
                 qual=my_qual,
                 database=database,
                 other_queries=other_queries)
开发者ID:champeric,项目名称:powa-web,代码行数:29,代码来源:qual.py


示例13: get_table_oid

    def get_table_oid(self, connection, table_name, schema=None, **kw):
        """Fetch the oid for schema.table_name.

        Several reflection methods require the table oid.  The idea for using
        this method is that it can be fetched one time and cached for
        subsequent calls.

        """
        table_oid = None
        if schema is not None:
            schema_where_clause = "n.nspname = :schema"
        else:
            schema_where_clause = "pg_catalog.pg_table_is_visible(c.oid)"
        query = """
            SELECT c.oid
            FROM pg_catalog.pg_class c
            LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
            WHERE (%s)
            AND c.relname = :table_name AND c.relkind in ('r', 'v', 'f')
        """ % schema_where_clause
        # Since we're binding to unicode, table_name and schema_name must be
        # unicode.
        table_name = str(table_name)
        bindparams = [sql.bindparam('table_name', type_=sqltypes.Unicode)]
        if schema is not None:
            schema = str(schema)
            bindparams.append(sql.bindparam('schema', type_=sqltypes.Unicode))
        s = sql.text(
            query, bindparams=bindparams, typemap={'oid': sqltypes.Integer})
        c = connection.execute(s, table_name=table_name, schema=schema)
        table_oid = c.scalar()
        if table_oid is None:
            raise exc.NoSuchTableError(table_name)
        return table_oid
开发者ID:Kozea,项目名称:sqlalchemy_fdw,代码行数:34,代码来源:dialect.py


示例14: get

    def get(self, database, query):
        bs = block_size.c.block_size
        stmt = powa_getstatdata_detailed_db()
        stmt = stmt.where(
            (column("datname") == bindparam("database")) &
            (column("queryid") == bindparam("query")))
        stmt = stmt.alias()
        from_clause = outerjoin(powa_statements, stmt,
                           and_(powa_statements.c.queryid == stmt.c.queryid, powa_statements.c.dbid == stmt.c.dbid))
        c = stmt.c
        rblk = mulblock(sum(c.shared_blks_read).label("shared_blks_read"))
        wblk = mulblock(sum(c.shared_blks_hit).label("shared_blks_hit"))
        stmt = (select([
            column("query"),
            sum(c.calls).label("calls"),
            sum(c.runtime).label("runtime"),
            rblk,
            wblk,
            (rblk + wblk).label("total_blks")])
            .select_from(from_clause)
            .where(powa_statements.c.queryid == bindparam("query"))
            .group_by(column("query"), bs))

        value = self.execute(stmt, params={
            "query": query,
            "database": database,
            "from": self.get_argument("from"),
            "to": self.get_argument("to")
        })
        if value.rowcount < 1:
            self.render("xhr.html", content="No data")
            return
        self.render("database/query/detail.html", stats=value.first())
开发者ID:champeric,项目名称:powa-web,代码行数:33,代码来源:query.py


示例15: test_cloned_alias

    def test_cloned_alias(self):
        entity = table(
            "entity", column("id"), column("employer_id"), column("name")
        )
        tag = table("tag", column("tag"), column("entity_id"))

        tags = (
            select([tag.c.entity_id, func.array_agg(tag.c.tag).label("tags")])
            .group_by(tag.c.entity_id)
            .cte("unaliased_tags")
        )

        entity_tags = tags.alias(name="entity_tags")
        employer_tags = tags.alias(name="employer_tags")

        q = (
            select([entity.c.name])
            .select_from(
                entity.outerjoin(
                    entity_tags, tags.c.entity_id == entity.c.id
                ).outerjoin(
                    employer_tags, tags.c.entity_id == entity.c.employer_id
                )
            )
            .where(entity_tags.c.tags.op("@>")(bindparam("tags")))
            .where(employer_tags.c.tags.op("@>")(bindparam("tags")))
        )

        self.assert_compile(
            q,
            "WITH unaliased_tags AS "
            "(SELECT tag.entity_id AS entity_id, array_agg(tag.tag) AS tags "
            "FROM tag GROUP BY tag.entity_id)"
            " SELECT entity.name "
            "FROM entity "
            "LEFT OUTER JOIN unaliased_tags AS entity_tags ON "
            "unaliased_tags.entity_id = entity.id "
            "LEFT OUTER JOIN unaliased_tags AS employer_tags ON "
            "unaliased_tags.entity_id = entity.employer_id "
            "WHERE (entity_tags.tags @> :tags) AND "
            "(employer_tags.tags @> :tags)",
        )

        cloned = q.params(tags=["tag1", "tag2"])
        self.assert_compile(
            cloned,
            "WITH unaliased_tags AS "
            "(SELECT tag.entity_id AS entity_id, array_agg(tag.tag) AS tags "
            "FROM tag GROUP BY tag.entity_id)"
            " SELECT entity.name "
            "FROM entity "
            "LEFT OUTER JOIN unaliased_tags AS entity_tags ON "
            "unaliased_tags.entity_id = entity.id "
            "LEFT OUTER JOIN unaliased_tags AS employer_tags ON "
            "unaliased_tags.entity_id = entity.employer_id "
            "WHERE (entity_tags.tags @> :tags) AND "
            "(employer_tags.tags @> :tags)",
        )
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:58,代码来源:test_cte.py


示例16: delete_items

    def delete_items(self, user_id, collection_name, item_ids=None,
                     filters=None, limit=None, offset=None, sort=None,
                     storage_time=None):
        """Deletes items. All items are removed unless item_ids is provided"""
        collection_id = self._get_collection_id(user_id, collection_name,
                                                create=False)
        if collection_id is None:
            return False

        wbo = self._get_wbo_table(user_id)
        query = _delete(wbo)
        where = [wbo.c.username == bindparam('user_id'),
                 wbo.c.collection == bindparam('collection_id')]

        if item_ids is not None:
            where.append(wbo.c.id.in_(item_ids))

        if filters is not None:
            for field, value in filters.items():
                field = getattr(wbo.c, field)

                operator, value = value
                if field.name == 'modified':
                    value = _roundedbigint(value)
                if isinstance(value, (list, tuple)):
                    where.append(field.in_(value))
                else:
                    if operator == '=':
                        where.append(field == value)
                    elif operator == '<':
                        where.append(field < value)
                    elif operator == '>':
                        where.append(field > value)

        where = and_(*where)
        query = query.where(where)

        if self.engine_name != 'sqlite':
            if sort is not None:
                if sort == 'oldest':
                    query = query.order_by(wbo.c.modified.asc())
                elif sort == 'newest':
                    query = query.order_by(wbo.c.modified.desc())
                else:
                    query = query.order_by(wbo.c.sortindex.desc())

            if limit is not None and int(limit) > 0:
                query = query.limit(int(limit))

            if offset is not None and int(offset) > 0:
                query = query.offset(int(offset))

        # XXX see if we want to send back more details
        # e.g. by checking the rowcount
        rowcount = self._do_query(query, user_id=user_id,
                                  collection_id=collection_id)
        return rowcount > 0
开发者ID:jbonacci,项目名称:server-storage,代码行数:57,代码来源:sql.py


示例17: between

def between(column):
    return sql.and_(
        sql.or_(
            column >= sql.bindparam("start_date"),
            sql.bindparam("start_date") == None),
        sql.or_(
            column <= sql.bindparam("end_date"),
            sql.bindparam("end_date") == None),
        )
开发者ID:kohsah,项目名称:bungeni-portal,代码行数:9,代码来源:daterange.py


示例18: test_compare_binds

    def test_compare_binds(self):
        b1 = bindparam("foo", type_=Integer())
        b2 = bindparam("foo", type_=Integer())
        b3 = bindparam("bar", type_=Integer())
        b4 = bindparam("foo", type_=String())

        def c1():
            return 5

        def c2():
            return 6

        b5 = bindparam("foo", type_=Integer(), callable_=c1)
        b6 = bindparam("foo", type_=Integer(), callable_=c2)
        b7 = bindparam("foo", type_=Integer(), callable_=c1)

        b8 = bindparam("foo", type_=Integer, value=5)
        b9 = bindparam("foo", type_=Integer, value=6)

        is_false(b1.compare(b5))
        is_true(b5.compare(b7))
        is_false(b5.compare(b6))
        is_true(b1.compare(b2))

        # currently not comparing "key", as we often have to compare
        # anonymous names.  however we should really check for that
        # is_true(b1.compare(b3))

        is_false(b1.compare(b4))
        is_false(b1.compare(b8))
        is_false(b8.compare(b9))
        is_true(b8.compare(b8))
开发者ID:monetate,项目名称:sqlalchemy,代码行数:32,代码来源:test_compare.py


示例19: get_indexes

    def get_indexes(self, connection, tablename, schema=None, **kw):
        # using system catalogs, don't support index reflection
        # below MS 2005
        if self.server_version_info < MS_2005_VERSION:
            return []

        current_schema = schema or self.default_schema_name
        full_tname = "%s.%s" % (current_schema, tablename)

        rp = connection.execute(
            sql.text("select ind.index_id, ind.is_unique, ind.name "
                "from sys.indexes as ind join sys.tables as tab on "
                "ind.object_id=tab.object_id "
                "join sys.schemas as sch on sch.schema_id=tab.schema_id "
                "where tab.name = :tabname "
                "and sch.name=:schname "
                "and ind.is_primary_key=0", 
                bindparams=[
                    sql.bindparam('tabname', tablename, 
                                    sqltypes.String(convert_unicode=True)),
                    sql.bindparam('schname', current_schema, 
                                    sqltypes.String(convert_unicode=True))
                ]
            )
        )
        indexes = {}
        for row in rp:
            indexes[row['index_id']] = {
                'name':row['name'],
                'unique':row['is_unique'] == 1,
                'column_names':[]
            }
        rp = connection.execute(
            sql.text(
                "select ind_col.index_id, ind_col.object_id, col.name "
                "from sys.columns as col "
                "join sys.tables as tab on tab.object_id=col.object_id "
                "join sys.index_columns as ind_col on "
                "(ind_col.column_id=col.column_id and "
                "ind_col.object_id=tab.object_id) "
                "join sys.schemas as sch on sch.schema_id=tab.schema_id "
                "where tab.name=:tabname "
                "and sch.name=:schname",
                        bindparams=[
                            sql.bindparam('tabname', tablename, 
                                    sqltypes.String(convert_unicode=True)),
                            sql.bindparam('schname', current_schema, 
                                    sqltypes.String(convert_unicode=True))
                        ]),
            )
        for row in rp:
            if row['index_id'] in indexes:
                indexes[row['index_id']]['column_names'].append(row['name'])

        return indexes.values()
开发者ID:Selissi,项目名称:CouchPotatoServer,代码行数:55,代码来源:base.py


示例20: delete_stmt

    def delete_stmt():
        clause = sql.and_()
        for col in mapper._pks_by_table[table]:
            clause.clauses.append(col == sql.bindparam(col.key, type_=col.type))

        if need_version_id:
            clause.clauses.append(
                mapper.version_id_col == sql.bindparam(mapper.version_id_col.key, type_=mapper.version_id_col.type)
            )

        return table.delete(clause)
开发者ID:wdxtub,项目名称:11601-imdb-crawler,代码行数:11,代码来源:persistence.py



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


鲜花

握手

雷人

路过

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

请发表评论

全部评论

专题导读
上一篇:
Python sql.case函数代码示例发布时间:2022-05-27
下一篇:
Python sql.asc函数代码示例发布时间: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