Examples¶
Retry decorator¶
TODO: Make a more configurable retry decorator and add it in aspectlib.contrib
.
class Client(object):
def __init__(self, address):
self.address = address
self.connect()
def connect(self):
# establish connection
def action(self, data):
# do some stuff
def retry(retries=(1, 5, 15, 30, 60), retry_on=(IOError, OSError), prepare=None):
assert len(retries)
@aspectlib.Aspect
def retry_aspect(*args, **kwargs):
durations = retries
while True:
try:
yield aspectlib.Proceed
break
except retry_on as exc:
if durations:
logging.warn(exc)
time.sleep(durations[0])
durations = durations[1:]
if prepare:
prepare(*args, **kwargs)
else:
raise
return retry_aspect
Now patch the Client
class to have the retry functionality on all its methods:
aspectlib.weave(Client, retry())
or with different retry options (reconnect before retry):
aspectlib.weave(Client, retry(prepare=lambda self, *_: self.connect())
or just for one method:
aspectlib.weave(Client.action, retry())
You can see here the advantage of having reusable retry functionality. Also, the retry handling is
decoupled from the Client
class.
Debugging¶
... those damn sockets:
>>> import aspectlib, socket, sys
>>> with aspectlib.weave(
... socket.socket,
... aspectlib.debug.log(
... print_to=sys.stdout,
... stacktrace=None,
... ),
... lazy=True,
... ):
... s = socket.socket()
... s.connect(('example.com', 80))
... s.send(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')
... s.recv(8)
... s.close()
...
{socket...}.connect(('example.com', 80))
{socket...}.connect => None
{socket...}.send(...'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')
{socket...}.send => 37
37
{socket...}.recv(8)
{socket...}.recv => ...HTTP/1.1...
...'HTTP/1.1'
...
The output looks a bit funky because it is written to be run by doctest - so you don’t use broken examples :)
Testing¶
Mock behavior for tests:
class MyTestCase(unittest.TestCase):
def test_stuff(self):
@aspectlib.Aspect
def mock_stuff(self, value):
if value == 'special':
yield aspectlib.Return('mocked-result')
else:
yield aspectlib.Proceed
with aspectlib.weave(foo.Bar.stuff, mock_stuff):
obj = foo.Bar()
self.assertEqual(obj.stuff('special'), 'mocked-result')