示例

完整的示例请参考sample。以下是简单的示例：

  1. #!/usr/bin/env python
  2. # coding:utf-8
  4. from pypegasus.pgclient import Pegasus
  6. from twisted.internet import reactor
  7. from twisted.internet.defer import inlineCallbacks
  10. @inlineCallbacks
  11. def basic_test():
  12.     # init
  13.     c = Pegasus(['127.0.0.1:34601', '127.0.0.1:34602'], 'temp')
  15.     suc = yield c.init()
  16.     if not suc:
  17.         reactor.stop()
  18.         print('ERROR: connect pegasus server failed')
  19.         return
  21.     # set
  22.     try:
  23.         ret = yield c.set('hkey1', 'skey1', 'value', 0, 500)
  24.         print('set ret: ', ret)
  25.     except Exception as e:
  26.         print(e)
  28.     # get
  29.     ret = yield c.get('hkey1', 'skey1')
  30.     print('get ret: ', ret)
  32.     reactor.stop()
  35. if __name__ == "__main__":
  36.     reactor.callWhenRunning(basic_test)
  37.     reactor.run()