import dbm

# 打開或創(chuàng)建一個數(shù)據(jù)庫文件（以讀寫模式）
db = dbm.open("mydatabase.db", "c")

# 添加鍵值對
db["key1"] = "value1"
db["key2"] = "value2"

# 通過鍵檢索數(shù)據(jù)
value1 = db["key1"]
print(value1)  # 輸出: b'value1'（注意：dbm庫中存儲的值是字節(jié)串）

# 刪除鍵值對
del db["key2"]

import dbm.gnu

# 打開或創(chuàng)建一個GNU格式的數(shù)據(jù)庫文件
db_gnu = dbm.gnu.open("mydatabase_gnu.db", "c")

import dbm.ndbm

# 打開或創(chuàng)建一個ndbm格式的數(shù)據(jù)庫文件
db_ndbm = dbm.ndbm.open("mydatabase_ndbm.db", "c")

import dbm.dumb

# 打開或創(chuàng)建一個dumb格式的數(shù)據(jù)庫文件
db_dumb = dbm.dumb.open("mydatabase_dumb.db", "c")

import dbm
import transaction
# 打開或創(chuàng)建一個數(shù)據(jù)庫文件
db = dbm.open("mydatabase.db", "c")
# 定義一個事務(wù)函數(shù)
def perform_transaction():
    try:
        # 在事務(wù)中執(zhí)行多步驟操作
        with transaction(db) as tr:
            # 添加鍵值對
            tr["key1"] = "value1"
            tr["key2"] = "value2"
            # 如果需要，可以進行其他操作
        # 提交事務(wù)
        tr.commit()
        print("Transaction successful")
    except Exception as e:
        # 事務(wù)失敗時的處理
        print(f"Transaction failed: {e}")
        # 回滾事務(wù)
        tr.rollback()
# 調(diào)用事務(wù)函數(shù)
perform_transaction()
# 關(guān)閉數(shù)據(jù)庫
db.close()

import dbm
import shutil
def backup_database(source_path, backup_path):
    try:
        # 打開原始數(shù)據(jù)庫
        source_db = dbm.open(source_path, 'r')
        # 復(fù)制數(shù)據(jù)庫文件到備份目錄
        shutil.copyfile(source_path, backup_path)
        # 關(guān)閉數(shù)據(jù)庫
        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:
        # 復(fù)制備份文件到目標(biāo)路徑
        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}")
# 指定數(shù)據(jù)庫文件路徑和備份文件路徑
source_database_path = "mydatabase.db"
backup_file_path = "backup/mydatabase_backup.db"
# 備份數(shù)據(jù)庫
backup_database(source_database_path, backup_file_path)
# 修改原始數(shù)據(jù)庫內(nèi)容，模擬數(shù)據(jù)損壞或丟失
# 恢復(fù)數(shù)據(jù)庫
restore_database(backup_file_path, source_database_path)

import dbm
import threading
# 創(chuàng)建一個鎖對象
db_lock = threading.Lock()
def update_database(key, value):
    with db_lock:
        try:
            # 打開數(shù)據(jù)庫
            db = dbm.open("mydatabase.db", 'c')
            # 更新數(shù)據(jù)庫
            db[key] = value
            # 關(guān)閉數(shù)據(jù)庫
            db.close()
            print(f"Database updated: {key} - {value}")
        except Exception as e:
            print(f"Error updating database: {e}")
# 創(chuàng)建多個線程進行并發(fā)更新
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):  # 設(shè)置默認(rèn)過期時間為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:
    # 操作數(shù)據(jù)庫的代碼

# 在此處數(shù)據(jù)庫已自動關(guān)閉

import dbm
import threading
# 創(chuàng)建線程鎖
db_lock = threading.Lock()
def concurrent_db_operation():
    with db_lock:
        with dbm.open("example.db", 'c') as db:
            # 并發(fā)安全的數(shù)據(jù)庫操作
# 在多線程環(huán)境中調(diào)用
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:
        # 操作數(shù)據(jù)庫的代碼
except dbm.error as e:
    print(f"DBM error: {e}")