python-etcd3-async

中文文档 | English

Python async client for the etcd API v3, supporting both synchronous and asynchronous operations.

• Free software: Apache Software License 2.0
• Python 3.12+

Features

• Full etcd v3 API support: Key-Value operations, Leases, Locks, Watches, Transactions
• Async/Sync dual support: Same API for both async and sync usage with feature parity
• Automatic reconnection: Auto-rebuild stale connections (e.g., after etcd server restart)
• Keepalive error propagation: Lease keepalive errors are properly propagated for connection loss detection
• Endpoint failover: Built-in support for multiple endpoints with automatic failover
• Context manager: Clean resource management for both sync and async
• Observability: Structured logging, OpenTelemetry tracing, and Prometheus metrics

Quick Start

Async Client

import asyncio
from etcd3 import async_client

async def main():
    # Create client
    client = await async_client(host='localhost', port=2379)

    # Key-Value operations
    await client.put('/mykey', 'Hello etcd!')
    value, meta = await client.get('/mykey')
    print(value)  # b'Hello etcd!'

    await client.delete('/mykey')
    await client.close()

asyncio.run(main())

Sync Client

from etcd3 import client

# Create sync client
etcd = client(host='localhost', port=2379)

# Key-Value operations
etcd.put('/mykey', 'Hello etcd!')
value, meta = etcd.get('/mykey')
print(value)  # b'Hello etcd!'

etcd.delete('/mykey')

Key Features

Key-Value Operations

# Put with lease
lease = await client.lease(30)
await client.put('/app/instance', 'value', lease=lease)

# Get with prefix
async for value, meta in client.get_prefix('/app/'):
    print(f'{meta.key}: {value}')

# Range query
async for value, meta in client.get_range('/start', '/end'):
    print(value)

Leases (TTL)

# Create lease with TTL
lease = await client.lease(60)
await client.put('/ephemeral/key', 'value', lease=lease)

# Keep lease alive
async for response in lease.refresh():
    print(f'Lease TTL: {response.TTL}')

# Revoke lease (deletes all attached keys)
await lease.revoke()

Distributed Locks

# Acquire lock (full path like Go SDK)
lock = await client.lock('/myapp/mylock', ttl=30)
acquired = await lock.acquire(timeout=10)

if acquired:
    try:
        # Critical section
        print('Lock acquired')
    finally:
        await lock.release()

# Or use context manager
lock = client.lock('/myapp/mylock', ttl=30)
async with lock:
    # Critical section
    pass

Watch Changes

# Watch single key
events_iter, cancel = await client.watch('/mykey')
async for event in events_iter:
    print(f'Event: {event.event_type}, value: {event.value}')

# Watch prefix
events_iter, cancel = await client.watch_prefix('/app/')
async for event in events_iter:
    print(f'Changed: {event.key}')

# Cancel watch
cancel()

# Watch with timeout (sync/async)
events_iter, cancel = client.watch('/key', timeout=10)
for event in events_iter:
    print(event)

Transactions

# Conditional transaction
success, response = await client.transaction(
    compare=[
        client.transactions.value('/key') == 'expected',
    ],
    success=[
        client.transactions.put('/key', 'new_value'),
    ],
    failure=[
        client.transactions.get('/key'),
    ]
)

Cluster Management

# List cluster members
async for member in client.members():
    print(f'{member.name}: {member.clientURLs}')

# Add/remove/update members
await client.add_member(['http://localhost:2380'])
await client.remove_member(member_id)
await client.update_member(member_id, ['http://localhost:2380'])

Maintenance Operations

# Get cluster status
status = await client.status()
print(f'Leader: {status.leader}')

# Compact history
await client.compact(revision, physical=True)

# Defragment database
await client.defragment()

# Alarm management
await client.create_alarm()  # Activate NOSPACE alarm
alarms = [alarm async for alarm in client.list_alarms()]
await client.disarm_alarm()

Examples

Leader Election

# Async: examples/leader_election.py
# Sync: examples/leader_election_sync.py

from etcd3 import async_client

async def main():
    client = await async_client(host='localhost', port=2379)
    election = LeaderElection(client, 'my-service-leader', ttl=30)

    if await election.try_become_leader():
        print('Became leader')
        # Do leader work...
    else:
        print('Not leader')

    await election.close()
    await client.close()

Service Registration

# Async: examples/service_registration.py
# Sync: examples/service_registration_sync.py

from etcd3 import async_client

async def main():
    client = await async_client(host='localhost', port=2379)
    registry = ServiceRegistration(client)

    lease = await registry.register(
        service_name='my-service',
        instance_id='instance-001',
        host='192.168.1.100',
        port=8080,
        ttl=30,
    )
    registry.start_keepalive(lease)
    # Service stays registered...

Service Discovery

# Async: examples/service_discovery.py
# Sync: examples/service_discovery_sync.py

from etcd3 import async_client

async def main():
    client = await async_client(host='localhost', port=2379)
    discovery = ServiceDiscovery(client)

    # Discover instances
    instances = await discovery.discover('my-service')
    print(f'Found {len(instances)} instances')

    # Watch for changes
    def on_change(new_instances):
        print(f'Changed: {len(new_instances)} instances')

    await discovery.watch('my-service', on_change)

Configuration

# Basic configuration
client = await async_client(
    host='localhost',
    port=2379,
    timeout=10,  # Request timeout in seconds
)

# Multiple endpoints with failover
client = await async_client(
    endpoints=['localhost:2379', 'localhost:22379'],
    failover=True,  # Automatically switch to healthy endpoint
)

# TLS authentication
client = await async_client(
    host='localhost',
    port=2379,
    ca_cert='/path/to/ca.crt',
    cert_key='/path/to/client.key',
    cert_cert='/path/to/client.crt',
)

# User authentication
client = await async_client(
    host='localhost',
    port=2379,
    user='username',
    password='password',
)

Observability

Configure via Environment

export ETCD_OBS_LOG_LEVEL=DEBUG
export ETCD_OBS_LOG_FORMAT=json
export ETCD_OBS_TRACING_ENABLED=true
export ETCD_OBS_PROMETHEUS_ENABLED=true

Configure Programmatically

from etcd3 import ObservabilityConfig

config = ObservabilityConfig(
    tracing_enabled=True,
    log_format="json",
    log_level="INFO",
    prometheus_enabled=True,
)
config.apply()

Access Metrics

from etcd3.exporters.prometheus import get_exporter

exporter = get_exporter()
if exporter:
    print(exporter.metrics().decode())

API Reference

Variable	Default	Description
ETCD_OBS_LOG_LEVEL	INFO	Logging level (DEBUG, INFO, WARNING, ERROR)
ETCD_OBS_LOG_FORMAT	text	Log format (text, json, colored)
ETCD_OBS_TRACING_ENABLED	true	Enable OpenTelemetry tracing
ETCD_OBS_TRACING_SERVICE_NAME	etcd3-client	Service name for tracing
ETCD_OBS_PROMETHEUS_ENABLED	false	Enable Prometheus metrics export

API Reference

Client Factory

Function	Description
async_client()	Create async client
client()	Create sync client

Key Methods

Category	Methods
KV	get, put, delete, get_prefix, get_range, get_all
Leases	lease, revoke_lease, refresh_lease, get_lease_info
Locks	lock
Watches	watch, watch_prefix, add_watch_callback, watch_once, cancel_watch
Transactions	transaction
Cluster	members, add_member, remove_member, update_member, status
Maintenance	compact, defragment, hash, create_alarm, list_alarms, disarm_alarm, snapshot

References

• Initial code 参考:
  1. python-etcd3
  2. Go etcd client v3