最全总结 | 聊聊 Python 数据处理全家桶（Sqlite篇）

第一时间关注 Python 技术干货！

1. 前言

上篇文章 聊到 Python 处理 Mysql 数据库最常见的两种方式，本篇文章继续说另外一种比较常用的数据库：Sqlite

Sqlite 是一种 嵌入式数据库，数据库就是一个文件，体积很小，底层由 C 语言编写，经常被集成到移动应用程序中

事实上，Python 内置了 sqlite3 模块，不需要安装任何依赖，就可以直接操作 Sqlite 数据库

2. 准备

和 Python 操作 Mysql 类似，操作 Sqlite 主要包含下面 2 种方式：

• sqlite3 + 原生 SQL

• SQLAlchemy + ORM

# 导入内置模块sqlite3
import sqlite3

# 创建数据库连接对象，如果数据库不存在，就自动新建一个数据库文件
# 还可以指定其他参数，包含：超时时间
 self.conn = sqlite3.connect(self.path_db)

# 获取操作数据库的游标对象
self.cursor = self.conn.cursor()

# 创建表
SQL_CREATE_TABLE = '''CREATE TABLE IF NOT EXISTS PEOPLE
       (ID INT PRIMARY KEY     NOT NULL,
       NAME           TEXT    NOT NULL,
       AGE            INT     NOT NULL);'''

def create_db_table(self):
    """
    初始化表
    :return:
    """
    self.conn.execute(SQL_CREATE_TABLE)

# 插入一条数据
SQL_INSERT_ONE_DATA = "INSERT INTO PEOPLE(id,name,age) VALUES(3,'xag',23);"

def insert_one(self):
    """新增一条数据"""
    try:
        self.conn.execute(SQL_INSERT_ONE_DATA)
        # 必须要提交，才能正确执行
        self.conn.commit()
    except Exception as e:
        self.conn.rollback()
        print('插入一条记录失败，回滚~')

# 插入多条数据（3个变量，包含：id、name、value）
SQL_INSERT_MANY_DATA = 'INSERT INTO PEOPLE (id,name,age) VALUES(?,?,?);'

# 待插入的数据
self.data = [(4, '张三', 11), (5, '李四', 12), (6, '王五', 13)]

def insert_many(self, data):
    """新增多条数据"""
    try:
        self.conn.executemany(SQL_INSERT_MANY_DATA, data)
        self.conn.commit()
    except Exception as e:
        self.conn.rollback()
        print('插入多条记录失败，回滚~')

# 查询的SQL语句
SQL_QUERY_ONE_DATA = "SELECT * FROM PEOPLE WHERE id={}"

def query_one(self, id):
    """
    查询一条数据
    :param id:
    :return:
    """
    self.cursor.execute(SQL_QUERY_ONE_DATA.format(id))

    # fetchone():查询第一条数据
    # fetchall()：查询所有数据
    # fetchmany(1):查询固定的数量的数据
    result = self.cursor.fetchall()
    print(type(result))
    print(result)

# 更新数据
SQL_UPDATE_ONE_DATA = "UPDATE PEOPLE SET NAME = '{}',AGE={} where id = {}"

def update_one(self, id, name, age):
    """
    修改一条记录
    :param id:
    :param name:
    :param age:
    :return:
    """
    sql_update = SQL_UPDATE_ONE_DATA.format(name, age, id)
    print(sql_update)
    self.conn.execute(sql_update)
    self.conn.commit()

# 删除数据
SQL_DEL_ONE_DATA = "DELETE FROM PEOPLE where id ={}"

def del_one(self, id):
    """通过id去删除一条数据"""
    sql_del = SQL_DEL_ONE_DATA.format(id)
    self.conn.execute(sql_del)
    self.conn.commit()

def teardown(self):
    # 关闭游标和数据库连接，避免资源浪费
    self.cursor.close()
    self.conn.close()

# 安装依赖包
pip3 install sqlalchemy

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base

# 基础类
Base = declarative_base()


# 自定义的表
class People(Base):
    # 表名
    __tablename__ = 'people'

    # 定义字段
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

    def __repr__(self):
        """
        便于打印结果
        :return:
        """
        return ".format(self.id, self.name, self.age)

# 创建数据库连接
engine = create_engine('sqlite:///./xh.db', echo=True)

# 创建表结构
# checkfirst：判断表是否存在，如果存在，就不重复创建
Base.metadata.create_all(engine, checkfirst=True)

# 实例化会话
self.session = sessionmaker(bind=engine)()

def add_one_data(self):
    """新增一条数据"""
    # 创建一个表的实例对象
    people = People(name='xag1', age=24)
    self.session.add(people)

    # 必须提交，才能更新到数据库中
    self.session.commit()

def add_datas(self, data):
    """
    新增多条数据
    :return:
    """
    self.session.add_all(data)
    self.session.commit()

def query_one_data(self, id):
    """
    通过id去查询一条数据
    :param id:
    :return:
    """
    # 通过id去查询数据，取第一条
    people = self.session.query(People).filter_by(id=id).first()
    print(people)
    print(type(people))

def query_all(self):
    """
    查询所有数据
    :return:
    """
    peoples = self.session.query(People).all()
    print(peoples)
    print(type(peoples))

def update1(self, id, name, age):
    """
    更新记录
    :param id:
    :param name:
    :param age:
    :return:
    """
    # 更新步骤：先查询、修改数据、然后确认修改
    people_temp = self.session.query(People).filter_by(id=id).first()

    # 修改数据
    people_temp.name = name
    people_temp.age = age

    # 确认提交修改
    self.session.commit()

def update2(self, id, name, age):
    """
    更新记录方法2
    :param id:
    :param name:
    :param age:
    :return:
    """
    self.session.query(People).filter_by(id=id).update({People.name: name, People.age: age})
    self.session.commit()

def del_one_data1(self, id):
    """
    删除一条数据方法1
    :param id:
    :return:
    """
    people_temp = self.session.query(People).filter_by(id=id).first()

    # 判断是否为空
    if people_temp is not None:
        self.session.delete(people_temp)
        self.session.commit()
    else:
        print('此条记录不存在，删除失败！')

def del_one_data2(self, id):
    """
    删除一条数据方法2
    :param id:
    :return:
    """
    try:
        self.session.query(People).filter_by(id=id).first().delete()
    except Exception as e:
        print('删除失败')

def teardown(self):
    """
    释放资源
    :return:
    """
    # 结束会话
    self.session.close()

    # 销毁引擎
    self.engine.dispose()

5.最后