轻量级TinyDB数据库文件写入和增删改查操作

生信与基因组学

1. TinyDB 数据库简介

TinyDB 是一个轻量级的 NoSQL 文档型数据库，由 Python 实现，无需额外的配置，以 JSON 文件作为存储，默认使用文件系统来存储数据。

2. 安装基本库

pip install tinydb
pip install Faker

3. 数据库操作

import pandas as pd
from addict import Dict
from tinydb import TinyDB, Query

# Faker生成假数据
def generate_fake_data(data_number: int, output_path: str):
    import faker 
    # 创建一个Faker对象，指定语言
    Faker.seed(123)
    fake = Faker('en_US')

    # 生成特定数量的假数据列表
    list_name = [fake.name() for _ in range(data_number)]
    list_address = [fake.address()  for _ in range(data_number)]
    list_phone_number = [fake.phone_number()  for _ in range(data_number)]
    list_city = [fake.city()  for _ in range(data_number)]

    # 生成dataframe
    dataframe = pd.DataFrame()
    dataframe['name'] = list_name
    dataframe['address'] = list_address
    dataframe['phone_number'] = list_phone_number
    dataframe['city'] = list_city

    # 保存数据
    dataframe.to_csv(output_path, sep='\t', index=False)

# 读取文件为dataframe
def read_dataframe(file_path):
    if file_path.endswith(('.tsv', '.txt')):
        return pd.read_table(file_path, sep='\t')
    elif file_path.endswith(('.xlsx', 'xls')):
        return pd.read_excel(file_path)
    elif file_path.enswith('.csv'):
        return pd.read_csv(file_path)
    else:
        raise Exception(f"Unknown format of file {file_path}")

# 将dataframe转换为字典列表
def dataframe2dict(dataframe: pd.DataFrame):
    list_columns = list(dataframe.columns)
    list_data = []
    # 遍历dataframe
    for idx, row in dataframe.iterrows():
        dict_tmp = Dict()
        for column in list_columns:
            dict_tmp.update({f'{column}': row[f'{column}']})
        list_data.append(dict_tmp)

    return list_data

# 获取Tinydb对象, 无json存储文件则创建
def get_database(db_file: str):
    if not db_file.endswith('.json'):
        raise Exception("The database file must be .json file!")
    return TinyDB(db_file)

# 关闭数据库
def close_database(db_object):
    db_object.close()

# 查询数据库记录
def query_record(db_file: str, query_type='all', query_name=''):
    db = get_database(db_file)
    if query_type == 'all':
        query_result = db.all()
    elif query_type == 'name':
        query = Query()
        query_result = db.search(query.name == query_name)

    close_database(db)
    return query_result

# 写入数据库记录
def write_record(db_file: str, write_data):
    db = get_database(db_file)
    if isinstance(write_data, dict):
        # 判断写入数据类型是否为字典，是则按单条数据写入
        db.insert(write_data)
    elif isinstance(write_data, list):
        # 判断写入数据类型是否为列表，是则按多条数据写入
        db.insert_multiple(write_data)
    elif isinstance(write_data, str):
        # 写入字符串
        db.insert_multiple(write_data)
    else:
        # 写入其他类型数据
        db.insert_multiple(str(write_data))

    close_database(db)

# 根据姓名name更新记录, 更新数据为字典格式
def update_record(db_file: str, update_name: str, update_data: dict):
    db = get_database(db_file)
    query = Query()
    db.update(update_data, query.name == update_name)

    close_database(db)

# 根据姓名name删除记录
def remove_record(db_file: str, remove_name: str):
    db = get_database(db_file)
    query = Query()
    db.remove(query.name == remove_name)

    close_database(db)

4. 将文本文件内容写入数据库和查询全部数据库记录

file_path = "test.txt"

# 生成10条假数据，保存为test.txt
generate_fake_data(data_number=10, output_path=file_path)

# 读取文件
dataframe = read_dataframe(file_path)

# 转换为字典
list_data = dataframe2dict(dataframe)

# 数据库存储文件
database_file = 'data.json'

# 数据库基本操作
print("*" * 10 + "写入字典记录" +"*" * 10)
# write_database(db_file=database_file, write_data={'name': 'test', 'address': 'GZ', 'phone_number': 123456, 'city': 'GZ'})

print("*" * 10 + "写入列表记录" +"*" * 10)
write_record(db_file=database_file, write_data=list_data)

print("*" * 10 + "查询全部记录" +"*" * 10)
query_result = query_record(db_file=database_file)

print(query_result[:20])

5. 更新、指定查询和删除数据库记录

print("*" * 10 + "更新记录" +"*" * 10)
update_record(db_file=database_file, update_name='Cathy Miller', update_data={'address': 'NA'})

print("*" * 10 + "查询更新后记录" +"*" * 10)
query_result = query_record(db_file=database_file, query_type='name', query_name='Ryan Ross')

print(query_result)

print("*" * 10 + "删除记录" +"*" * 10)
remove_record(db_file=database_file, remove_name='Ryan Ross')

print("*" * 10 + "查询删除后记录" +"*" * 10)
query_result = query_record(db_file=database_file, query_type='name', query_name='Ryan Ross')

print(query_result)

微信扫一扫分享文章