aioconsul 0.3

安装

pip install aioconsul

使用量

大多数函数都是协程，因此必须将其嵌入到异步任务中：

from aioconsul import Consul
client = Consul()

@asyncio.coroutine
def main():
    node_name = yield from client.agent.config().node_name
    print('I am %s!' % node_name)

loop = asyncio.get_event_loop()
loop.run_until_complete(main)
loop.run_until_complete(future)

代理检查

目前，这个库可以处理简单的检查：

from aioconsul import Consul
client = Consul()

# list all checks
for check in (yield from client.agent.checks()):
    print(check.id)

# look for a check
check = yield from client.agent.checks.get('my-check')

# register a script check
check = yield from client.agent.checks.register_script('my-script-check',
                                                       script='~/script.sh',
                                                       interval='5m')

# register a http check
check = yield from client.agent.checks.register_ttl('my-http-check',
                                                    http='http://example.com',
                                                    interval='10h')

# register a ttl check
check = yield from client.agent.checks.register_ttl('my-ttl-check',
                                                    ttl='10s')

# mark ttl check passing
yield from client.agent.checks.passing(check, note='Make it so')

# deregister any check
yield from client.agent.checks.deregister(check)

代理服务

目前，这个库可以处理简单的检查：

from aioconsul import Consul
client = Consul()

# list all services
for srv in (yield from client.agent.services()):
    print(srv.id)

# search a service by its name
srv = yield from client.agent.services.get('my-service')

# disable a service
yield from client.agent.services.maintenance(srv,
                                             enable=False,
                                             reason='Migrating stuff')

# and reenable it
yield from client.agent.services.maintenance(srv,
                                             enable=True,
                                             reason='Done')

目录

此库可以查阅目录：

from aioconsul import Consul
client = Consul()

# listing all nodes from catalog
for node in (yield from client.catalog.nodes()):
    print(node.name)
    print(node.address)

# getting a node with all of its service
node = yield from client.catalog.get('my-node')
print(node.services)

# getting all nodes that belong to a service
nodes = yield from client.catalog.nodes(service='my-service')
print(nodes)

以及注册支票、服务和节点：

from aioconsul import Consul
client = Consul()

node = {'name': 'my-local-node',
        'address': '127.0.0.1'}
check = {'name': 'baz',
         'state': 'passing',
         'service_id': 'bar'}
service={'name': 'bar'}

resp = yield from client.catalog.register(node, check=check, service=service)
assert resp

resp = yield from client.catalog.deregister(node, check=check, service=service)
assert resp

事件

from aioconsul import Consul
client = Consul()

# send an event
event = yield from client.event.fire('my-event', node_filter='.*')

# list all events
for event in (yield from client.event.items()):
    print(event.name)

健康

from aioconsul import Consul
client = Consul()

# checks for a node
for check in (yield from client.health.node('my-local-node')):
    assert check.status == 'passing'

# health of a node
for check in (yield from client.health.node('my-local-node')):
    assert check.status == 'passing'

# health of a check id
for check in (yield from client.health.checks('serfHealth')):
    assert check.status == 'passing'

# health of a check id
for check in (yield from client.health.checks('serfHealth')):
    assert check.status == 'passing'

# health of a service
for node in (yield from client.health.service('foo', state='any')):
    for check in node.checks:
        if check.id == 'service:foo':
            assert check.status == 'passing'

# passing checks
for check in (yield from client.health.state('passing')):
    assert check.status == 'passing'

Kv和会话

简单示例：

from aioconsul import Consul
client = Consul()

# set a k/v
yield from client.kv.set('my.key', 'my.value')

# fetch a k/v
obj = yield from client.kv.get('my.key')

# fetched values have a special attribute `consul`
assert obj.key.name == 'my.key'

# delete a k/v
yield from client.kv.delete('my.key')

许多k/v:

# list many k/v
for key, value in (yield from client.kv.items('')):
    print(key, value)

短暂的k/v:

session = yield from client.sessions.create(behavior='delete')
yield from client.kv.create('my.key', 'my.key')
yield from client.sessions.delete(session)

try:
    # try to fetch previous k/v
    obj = yield from client.kv.get('my.key')
except client.kv.NotFound:
    # but it was destroyed with the session
    pass

前交叉韧带

from aioconsul import Consul, PermissionDenied
client = Consul(token=master_token)

# create a token
token = (yield from client.acl.create('my-acl', rules=[
    ('key', '', 'read'),
    ('key', 'foo/', 'deny'),
]))

# access to kv with the fresh token
node = Consul(token=token)
yield from node.kv.get('foo')
with pytest.raises(PermissionDenied):
    yield from node.kv.set('foo', 'baz')
with pytest.raises(node.kv.NotFound):
    yield from node.kv.get('foo/bar')

测试

测试依赖于Consul二进制和py.test。

1. 安装consun binary，它必须可以在$PATH中访问。

2. 安装测试要求：

   pip install -r requirements-tests.txt
   

3. 然后运行测试：

   py.test --cov-report html --cov aioconsul tests