python-consul

import consul import os # Configure Consul connection details consul_host = os.environ.get('CONSUL_HOST', '127.0.0.1') consul_port = int(os.environ.get('CONSUL_PORT', '8500')) consul_scheme = os.environ.get('CONSUL_SCHEME', 'http') consul_token = os.environ.get('CONSUL_TOKEN', None) # For ACLs try: # Initialize Consul client c = consul.Consul( host=consul_host, port=consul_port, scheme=consul_scheme, token=consul_token ) # Example: Key-Value store operations key = "my_app/config/setting1" value = "hello_consul_world" # Put a value if c.kv.put(key, value): print(f"Successfully set key '{key}' to '{value}'") else: print(f"Failed to set key '{key}'") # Get a value index, data = c.kv.get(key) if data: retrieved_value = data['Value'].decode('utf-8') print(f"Retrieved key '{key}': '{retrieved_value}' (index: {index})") else: print(f"Key '{key}' not found.") # Example: Register a service (minimal) service_name = "my-test-service" service_id = "my-test-service-1" if c.agent.service.register(name=service_name, service_id=service_id, port=8000, tags=['python']): print(f"Service '{service_name}' registered successfully.") else: print(f"Failed to register service '{service_name}'.") # You can deregister later with: # c.agent.service.deregister(service_id) # print(f"Service '{service_name}' deregistered.") except Exception as e: print(f"An error occurred: {e}") print("Please ensure a Consul agent is running and accessible at " f"{consul_scheme}://{consul_host}:{consul_port}")