sqlite 在 Python 下的再封装

原来一直使用 MongoDB 作为主力数据库，但是单位的电脑是 32 位的机器，有单一数据库的限制，故一直在寻找替代品。最近发现原来 sqlite 用起来也不错，其 api 简洁明了。另外，作为一个文件型数据库， Python 自带，不需要安装服务器软件，用起来非常方便，并且也不存在安全隐患。

由于本人经常写命令行的软件，故需要对系统提供的 api 再进一步封装，更加方便自己的使用。

程序的封装 sqlite.py

文件头

import atexit
import sqlite3
from contextlib import contextmanager, closing
from pathlib import Path

_conn = None
_config = {}

db_config

def db_config(database: str, **kw):
    global _config
    kw['database'] = str(database)
    _config = kw

connect

def connect():
    global _conn
    if not _conn:
        _conn = sqlite3.connect(**_config)
        atexit.register(_conn.close)
    return _conn

trans

@contextmanager
def trans():
    try:
        conn = connect()
        yield
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise e

execute

def execute(sql: str, params: list = []):
    return connect().execute(sql, params)

executemany

def executemany(sql: str, params: list = []):
    return connect().executemany(sql, params)

executescript

def executescript(sql: str):
    return connect().executescript(sql)

executefile

def executefile(pkg: str, filename: str):
    '''
    执行程序中附带的资源文件
    pkg         : 所在包的名称
    filename    : 相关于包的文件名，包括路径
    '''
    from pkgutil import get_data
    data = get_data(pkg, filename)
    sql = data.decode('utf8')
    return executescript(sql)

find

def find(sql: str, params: list = [], multi=True):
    '''执行sql 语句，并返行多行或一行记录'''
    cur = execute(sql, params)
    with closing(cur):
        return cur.fetchall()if multi else cur.fetchone()

findone

def findone(sql: str, params: list = []):
    '''执行 sql 语句，并返回一行记录'''
    return find(sql, params, multi=False)

findvalue

def findvalue(sql: str, params: list = []):
    '''执行 sql 语句，并返回一个值 '''
    row = findone(sql, params)
    return row and row[0]

具体使用方法

经过封装后，执行数据库指令或查询时不用调用 connect 来连接数据库，先用 db_config来设置好 connect的参数就行了。后面的程序在执行数据库命令或查询语句的时候会自动连接，并在程序退出的时候自带关闭（使用 atexit来实现）。

配置连接参数

调用 db_config就行了，其参数与 sqlite3 库中的 connect的参数完全一致。使用方法如下：

db_config(':memory:')

执行 sql 语句

共有四条命令可用，其用途如下：

execute：没有参数或只有一行参数时使用
executemay:同一条语句多行参数时使用
executescript:执行多条语句，并且无参数时使用，一般在定义表或删除表的时候使用
executefile:执行多条语句，并且语句以包数据文件的时候使用

注意： execute 和 executemany 需要在 trans 环境下执行，同一个trans下全部执行成功则提交服务器，执行中存在异常则全部回滚。如果不在 trans环境下执行，则连接关闭后修改的数据完全丢失。executescript 和 executefile不需要在 trans环境下执行，会自动提交。故只应使用 DDL 语句，而不应该插入或修改数据。

具体示例：

建表文件：test.sql

create table if not exists test(
    id  int  primary key,
    name text,
    age  int
);

create table if not exists school(
    userid int primary key,
    class   text
)

执行语句示例

db_config(':memory:')

executescript('drop table if exists test;'
             'drop table if exists school;')

executefile('pkg','test.sql')

with trans():
    execute('insert into test values(?,?,?)',[1,'Tom',23])
    executemany('insert into test values(?,?,?)',[
        [2,'Alice',22],
        [3,'John',21]])

查询语句

执行查询可以使用 find、findone 和findvalue三个命令，其差异如下：

find: 执行查询后提取多行数据
findone: 执行查询后只提取第一行数据
findvalue:执行查询后只提取第一行数据的第一个值

使用示例：

usercount=findvalue('select count(id) from test')
id,name,age=findone('select * from test where name=?',['tom'])
for id,name,age in find('select * from test'):
    print(id,name,age)