#!/usr/bin/env python
#coding=utf-8
import shelve
import threadingclass A(object):  """一个用于测试的类 其中只打印出hello和bye  """def __init__(self, name):self.name = nameself.__actions = ['hello', 'bye']def _hello(self):print ('hello, %s' % self.name)def _bye(self):print ('bye, %s' % self.name)def act(self):for a in self.__actions:if callable(getattr(self, '_'+a)):getattr(self, '_'+a)()FILENAME = 'shelve_test'
KEY = 'myobj'
DATA = A('qx')def test1():  """基本用法  """with shelve.open(FILENAME) as d:d[KEY] = DATAdata = d[KEY]print ('type(data)=%s' % type(data))del d[KEY]try:data = d[KEY]except KeyError:print ('key=%s not found' % KEY)d[KEY] = DATAdef test2():  """当没有使用write_back=True时 shelve无法检测到可变对象的更新  """with shelve.open(FILENAME) as d:d['xx'] = [0, 1, 2]print ("d['xx']=", d['xx'])  # [0, 1, 2]d['xx'].append(3)print ("d['xx']=", d['xx'])  # [0, 1, 2]def test3():  """继续test2 只能通过重新赋值来更新d['xx']  """with shelve.open(FILENAME) as d:d['xx'] = [0, 1, 2]print ("d['xx']=", d['xx'])  # [0, 1, 2]data = d['xx']data.append(3)d['xx'] = dataprint ("d['xx']=", d['xx'])  # [0, 1, 2, 3]def test4(flag=None):  """不支持并发读写 这是由Unix文件系统的文件锁保证的 但是支持多线程并发读  """if flag is None: flag = 'c'def _test():d = shelve.open(FILENAME, flag=flag)try:data = d[KEY]except Exception as exc:print ('exception:%s' % exc)else:data.act()print (d['xx'])finally:d.close()t1 = threading.Thread(target=_test)t2 = threading.Thread(target=_test)t1.start();t2.start()t1.join();t2.join()def split(seq):  """打印分隔线  """print ('====test %d====' % seq)if __name__ == '__main__':split(1);test1()  # key=myobj not foundsplit(2);test2()split(3);test3()split(4);test4('r')  # 如果调用test4() 会报异常 Resource temporarily unavailable