Python dbm库利用键值对存储数据

import dbm

# 打开或创建一个数据库文件（以读写模式）
db = dbm.open("mydatabase.db", "c")

# 添加键值对
db["key1"] = "value1"
db["key2"] = "value2"

# 通过键检索数据
value1 = db["key1"]
print(value1)  # 输出: b'value1'（注意：dbm库中存储的值是字节串）

# 删除键值对
del db["key2"]

import dbm.gnu

# 打开或创建一个GNU格式的数据库文件
db_gnu = dbm.gnu.open("mydatabase_gnu.db", "c")

import dbm.ndbm

# 打开或创建一个ndbm格式的数据库文件
db_ndbm = dbm.ndbm.open("mydatabase_ndbm.db", "c")

import dbm.dumb

# 打开或创建一个dumb格式的数据库文件
db_dumb = dbm.dumb.open("mydatabase_dumb.db", "c")

import dbm
import transaction
# 打开或创建一个数据库文件
db = dbm.open("mydatabase.db", "c")
# 定义一个事务函数
def perform_transaction():
    try:
        # 在事务中执行多步骤操作
        with transaction(db) as tr:
            # 添加键值对
            tr["key1"] = "value1"
            tr["key2"] = "value2"
            # 如果需要，可以进行其他操作
        # 提交事务
        tr.commit()
        print("Transaction successful")
    except Exception as e:
        # 事务失败时的处理
        print(f"Transaction failed: {e}")
        # 回滚事务
        tr.rollback()
# 调用事务函数
perform_transaction()
# 关闭数据库
db.close()

import dbm
import shutil
def backup_database(source_path, backup_path):
    try:
        # 打开原始数据库
        source_db = dbm.open(source_path, 'r')
        # 复制数据库文件到备份目录
        shutil.copyfile(source_path, backup_path)
        # 关闭数据库
        source_db.close()
        print(f"Backup successful. Backup file saved at {backup_path}")
    except Exception as e:
        print(f"Backup failed: {e}")
def restore_database(backup_path, target_path):
    try:
        # 复制备份文件到目标路径
        shutil.copyfile(backup_path, target_path)
        print(f"Restore successful. Database file restored at {target_path}")
    except Exception as e:
        print(f"Restore failed: {e}")
# 指定数据库文件路径和备份文件路径
source_database_path = "mydatabase.db"
backup_file_path = "backup/mydatabase_backup.db"
# 备份数据库
backup_database(source_database_path, backup_file_path)
# 修改原始数据库内容，模拟数据损坏或丢失
# 恢复数据库
restore_database(backup_file_path, source_database_path)

import dbm
import threading
# 创建一个锁对象
db_lock = threading.Lock()
def update_database(key, value):
    with db_lock:
        try:
            # 打开数据库
            db = dbm.open("mydatabase.db", 'c')
            # 更新数据库
            db[key] = value
            # 关闭数据库
            db.close()
            print(f"Database updated: {key} - {value}")
        except Exception as e:
            print(f"Error updating database: {e}")
# 创建多个线程进行并发更新
threads = []
for i in range(5):
    thread = threading.Thread(target=update_database, args=(f'key{i}', f'value{i}'))
    threads.append(thread)
    thread.start()
# 等待所有线程完成
for thread in threads:
    thread.join()

import dbm
def save_config(config_dict):
    with dbm.open("config.db", 'c') as db:
        for key, value in config_dict.items():
            db[key] = str(value)
def load_config():
    config_dict = {}
    with dbm.open("config.db", 'r') as db:
        for key, value in db.items():
            config_dict[key] = value.decode('utf-8')
    return config_dict
# 示例
config_data = {'username': 'admin', 'password': 'secretpass', 'debug_mode': 'True'}
save_config(config_data)
loaded_config = load_config()
print(loaded_config)

import dbm
import time
def cache_data(key, data, expiration=60):  # 设置默认过期时间为60秒
    with dbm.open("cache.db", 'c') as db:
        db[key] = f"{time.time() + expiration} {data}"
def get_cached_data(key):
    with dbm.open("cache.db", 'r') as db:
        if key in db:
            expiration, data = db[key].decode('utf-8').split(maxsplit=1)
            if float(expiration) > time.time():
                return data
# 示例
cache_data('user:123', 'Cached user data for user 123')
cached_data = get_cached_data('user:123')
print(cached_data)

with dbm.open("example.db", 'c') as db:
    # 操作数据库的代码

# 在此处数据库已自动关闭

import dbm
import threading
# 创建线程锁
db_lock = threading.Lock()
def concurrent_db_operation():
    with db_lock:
        with dbm.open("example.db", 'c') as db:
            # 并发安全的数据库操作
# 在多线程环境中调用
thread1 = threading.Thread(target=concurrent_db_operation)
thread2 = threading.Thread(target=concurrent_db_operation)
thread1.start()
thread2.start()

import dbm
try:
    with dbm.open("example.db", 'c') as db:
        # 操作数据库的代码
except dbm.error as e:
    print(f"DBM error: {e}")