# flashdb 模块 API 文档 ## API ### class KVDB(_flashdb.KVDB): ``` python def get_blob(self,key,size):... ``` ``` python def set_by_fmt(self,key,v,fmt):... ``` ``` python def get_by_fmt(self,key,size,fmt):... ``` ### class TSDB(_flashdb.TSDB): ``` python def __init__(self,name:str,path:str,max_len:int=1024,user_data=None):... ``` ``` python def tsl_iter_by_time(self,from_time,to_time,callback:any,user_data:any)->int:... ``` ### class TSL(_flashdb.TSL): ## Examples ### flashdb_kvdb1.py ```python import flashdb import struct DB_PATH = "test/out/fdb_kvdb" print('test boot_count increment 1') boot_count = 0 boot_count_blob = struct.pack('i', boot_count) boot_times = [0, 1, 2, 3, 0, 0, 0, 0, 0, 0] boot_time_tuple = tuple(boot_times) boot_time_blob = struct.pack('@10Q', *boot_time_tuple) default_kv = { 'username': 'armink', # string KV 'password': "123456", # string KV 'boot_count': boot_count_blob, # int type kv 'boot_time': boot_time_blob, # array type kv } # print(default_kv) fdb = flashdb.KVDB("env", DB_PATH, default_kv, None) res = fdb.get_blob("boot_count", len(boot_count_blob)) assert res is not None boot_count = struct.unpack("i", res)[0] boot_count = boot_count+1 boot_count_blob = struct.pack('i', boot_count) fdb.set_blob("boot_count", boot_count_blob) res = fdb.get_blob("boot_count", len(boot_count_blob)) assert res is not None new_boot_count = struct.unpack("i", res)[0] assert new_boot_count == boot_count print('PASS') ``` ### flashdb_kvdb2.py ```python import flashdb import struct DB_PATH = "test/out/fdb_kvdb" print('test boot_count increment 2') boot_count=0 boot_count_fmt='i' boot_count_blob = struct.pack(boot_count_fmt, boot_count) boot_count_size = len(boot_count_blob) boot_time = [0,1,2,3,0,0,0,0,0,0] boot_time_fmt='@10Q' boot_time_tuple = tuple(boot_time) boot_time_blob = struct.pack(boot_time_fmt, *boot_time_tuple) default_kv={ 'username': 'armink', # string KV 'password': "123456", # string KV 'boot_count': boot_count_blob, # int type kv 'boot_time': boot_time_blob, # array type kv } fdb = flashdb.KVDB("env", DB_PATH, default_kv, None) #print(default_kv) kvdb = fdb boot_count = fdb.get_by_fmt("boot_count", boot_count_size, boot_count_fmt) assert boot_count is not None print("==================== kvdb_basic_sample ====================") print( "get the 'boot_count' value is %d" % boot_count) boot_count = boot_count +1 res =fdb.set_by_fmt("boot_count", boot_count, boot_count_fmt) new_boot_count = fdb.get_by_fmt("boot_count", boot_count_size, boot_count_fmt) assert new_boot_count is not None print( "get the 'boot_count' value is %d" % new_boot_count) print("===========================================================") assert new_boot_count == boot_count print('PASS') ``` ### flashdb_tsdb1.py ```python import flashdb import struct import time import os DB_PATH = "test/out/fdb_tsdb" tsdb = flashdb.TSDB("env", DB_PATH, max_len=512) tic = time.time() * 1000 for i in range(10): blob_i = struct.pack('i', i) time.sleep(0.001) ret = tsdb.tsl_append(blob_i) toc = time.time() * 1000 assert ret == 0 sum_by_time = 0 def callback(tsl, user_data) -> int: global sum_by_time # print(tsl.get_time(), tsl.to_blob()) t = tsl.get_time() blob_i = tsl.to_blob() i = struct.unpack('i', blob_i)[0] print(t, i, user_data) if user_data == 'user_data_by_time': sum_by_time += i return False # False: continue, True: stop assert tsdb.tsl_iter(callback, 'user_data') == 0 assert tsdb.tsl_iter_reverse(callback, 'user_data_reverse') == 0 print('toc - tic', toc - tic) assert tsdb.tsl_iter_by_time(tic, toc, callback, 'user_data_by_time') == 0 assert sum_by_time == 45 print('PASS') ```