python-database

发表于 更新于 分类于 阅读次数： 本文字数： 2.6k 阅读时长 ≈ 2 分钟

sqlite api

sqlite

self.conn = sqlite3.connect('conf.db')
self.cursor = self.conn.cursor()
sql = """
create table if not exists t_measurements(f_id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
f_solutionId INTEGER NOT NULL,
f_name CHAR(300) NOT NULL , f_measureType CHAR(100) NOT NULL , f_points CHAR(500) NOT NULL );
"""
self.cursor.execute(sql)
sql = """
create table if not exists t_solution(f_id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
f_name CHAR(300) NOT NULL , f_createDT CHAR(100) NOT NULL );
"""
self.cursor.execute(sql)
self.conn.commit()

delete

def delete_task(conn, id):
    """
    Delete a task by task id
    :param conn:  Connection to the SQLite database
    :param id: id of the task
    :return:
    """
    sql = 'DELETE FROM tasks WHERE id=?'
    cur = conn.cursor()
    cur.execute(sql, (id,))
    conn.commit()

def delete_all_tasks(conn):
    """
    Delete all rows in the tasks table
    :param conn: Connection to the SQLite database
    :return:
    """
    sql = 'DELETE FROM tasks'
    cur = conn.cursor()
    cur.execute(sql)
    conn.commit()

插入

sql = 'insert into t_solution(f_id , f_name , f_createDT) values (?,?,?)'
self.cursor.execute(sql, (None, new_name, dt))
self.conn.commit()
self.cursor.lastrowid

records

用了一下 不好用，例子很少，常用的功能找起来费劲

据说是 Requests 作者的作品，值得试用一下，通用而优雅的数据库访问库

Records 的特点：

  • 基于 SQLAlchemy 和 Tablib 开发。
  • 方法非常简单且统一。
  • 支持缓存查询的数据.
  • 自动实现上下文管理器，不需要关注数据连接状态。
  • 支持数据库事务，保持数据的原子性和一致性。
  • 支持安全的参数化查询，以安全的形式进行动态参数传递，防止 sql 语句不规范导致的安全问题；
  • 支持主流数据库，如 RedShift, Postgres, MySQL, SQLite, Oracle, 和 MS-SQL
pip install records

Records 是基于 SQLAlchemy 实现的，所以数据库链接方式参考 SQLAlchemy

20200916_154524.png

Records 操作数据的 步骤非常的简单，如下：

  • 连接数据库，返回 DB 数据库对象。
  • 使用 query 方法执行 SQL 语句。

sqlite

connect

self.db = records.Database('sqlite:///conf.db')

create table

sql = """create table if not exists t_measurements(f_id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
f_name CHAR(300) NOT NULL , f_measureType CHAR(100) NOT NULL , f_points CHAR(500) NOT NULL );
"""
self.db.query(sql)

select

sql = 'select * from t_solution;'
        for row in self.cursor.execute(sql):
            self.solution_dict[row[1]] = {'id': row[0], 'name': row[1], 'createDT': row[2]}
            self.mlbSolution.insert('end', *map(unicode, (row[1], row[2])))

统计行数

sql = 'select COUNT(*) from t_config;'
self.cursor.execute(sql)
cur_result = self.cursor.fetchone()
if not cur_result or 0 == cur_result[0]:
    sql = 'insert into t_config(f_id,f_devIp,f_devPort) values (?,?,?)'
    self.cursor.execute(sql, (None, '', ''))
    self.conn.commit()

insert

sql = 'insert into t_solution(f_id , f_name , f_createDT) values (?,?,?)'
self.cursor.execute(sql, (None, new_name, dt))
self.conn.commit()

导出

print(rows.export('json'))
print(rows.export('csv'))