Usage¶
Comprehensive example¶
server.py
#!/usr/bin/env python3.6
from isc.server import Node, expose, on, local_timer
from time import sleep
import logging
class ExampleService(object):
name = 'example'
def __init__(self):
self.tracker = None
@expose
def add(self, a, b, wait=0):
sleep(wait)
return str(a + b) * 8000
@expose
def raise_error(self):
raise Exception('testing')
def private_method(self):
return 'Cannot call me!'
@on('boom')
def do_stuff(self, arg):
print('Got stuff:', arg)
@expose
def slow_method(self):
sleep(3)
@expose
def start_tracking(self):
self.tracker = tracker.SummaryTracker()
@expose
def get_summary(self):
return list(self.tracker.format_diff())
# @local_timer(timeout=3)
# def print_stats(self):
# print('Stats: foobar')
service = ExampleService()
node = Node(exchange='isctest')
node.set_logging_level(logging.DEBUG)
node.register_service(service)
if __name__ == '__main__':
try:
node.run()
except KeyboardInterrupt:
node.stop()
client.py
#!/usr/bin/env python3.6
from isc.client import Client, RemoteException, TimeoutException
client = Client(exchange='isctest')
client.start()
import time
time.sleep(2)
client.notify('boom', dict(foo='bar'))
a
assert client.example.add(2, 3) == '5' * 8000
assert client.invoke('example', 'add', 2, 3) == '5' * 8000
try:
client.example.add(2, '3')
except RemoteException:
pass
else:
assert False
try:
client.example.raise_error()
except RemoteException:
pass
else:
assert False
try:
client.example.private_method()
except RemoteException:
pass
else:
assert False
try:
client.example.unexisting_method()
except RemoteException:
pass
else:
assert False
try:
client.set_invoke_timeout(1)
client.example.slow_method()
except TimeoutException:
pass
else:
assert False
client.stop()
Caveats¶
If you import anything from isc.server
, keep in mind that
this library uses gevent
to patch built-in libraries. This
doesn’t apply to isc.client
though.