db.py 文件源码

def execute(self, query, params=[]):
        if len(params) > 0 and len(query.split(';')) > 1:
            raise Exception("Multiple queries with parameters is unsupported")

        # Expand lists in paramters
        prev = -1
        new_params = []
        for p in params:
            prev = query.find('?', prev+1)
            if type(p) in [np.uint16, np.uint32, np.uint64]:
                new_params.append(np.int64(p))  # sqlite is really fussy about this
            elif type(p) in [list, tuple]:
                rep = "(" + ",".join("?"*len(p)) + ")"
                query = query[:prev] + rep + query[prev+1:]
                prev += len(rep)
                new_params.extend(p)
            else:
                new_params.append(p)

        for q in query.split(';'):
            self.cur.execute(q, tuple(new_params))
        return self.cur