0
# V2 Configuration Management (Asynchronous)
1

2
Modern asynchronous API for Nacos configuration operations with advanced features including GRPC support, encryption, filtering, caching, and comprehensive error handling. This API provides type-safe operations using Pydantic models.
3

4
## Capabilities
5

6
### Service Creation and Initialization
7

8
Create and initialize the asynchronous configuration service.
9

10
```python { .api }
11
class NacosConfigService:
12
    @staticmethod
13
    async def create_config_service(client_config: ClientConfig) -> 'NacosConfigService':
14
        """
15
        Create and initialize a Nacos configuration service.
16
        
17
        Args:
18
            client_config (ClientConfig): Client configuration containing server details,
19
                                        authentication, and operational parameters
20
                                        
21
        Returns:
22
            NacosConfigService: Initialized configuration service instance
23
            
24
        Raises:
25
            NacosException: If client_config is invalid or initialization fails
26
        """
27
```
28

29
Usage example:
30

31
```python
32
import asyncio
33
from v2.nacos import ClientConfig, NacosConfigService
34

35
async def main():
36
    # Create client configuration
37
    client_config = ClientConfig(
38
        server_addresses="127.0.0.1:8848",
39
        namespace_id="production",
40
        username="nacos",
41
        password="nacos"
42
    )
43
    
44
    # Create configuration service
45
    config_service = await NacosConfigService.create_config_service(client_config)
46
    
47
    # Use the service...
48
    
49
    # Always shutdown when done
50
    await config_service.shutdown()
51

52
asyncio.run(main())
53
```
54

55
### Get Configuration
56

57
Retrieve configuration values with automatic caching and fallback support.
58

59
```python { .api }
60
async def get_config(self, param: ConfigParam) -> str:
61
    """
62
    Get configuration from Nacos server with caching and failover support.
63
    
64
    Args:
65
        param (ConfigParam): Configuration parameters including data_id and group
66
        
67
    Returns:
68
        str: Configuration content
69
        
70
    Raises:
71
        NacosException: If data_id is empty or invalid
72
    """
73
```
74

75
Usage example:
76

77
```python
78
from v2.nacos import ConfigParam
79

80
# Basic configuration retrieval
81
param = ConfigParam(dataId="database.config", group="DEFAULT_GROUP")
82
config = await config_service.get_config(param)
83
print(f"Database config: {config}")
84

85
# Get configuration with custom group
86
param = ConfigParam(dataId="redis.config", group="CACHE_GROUP")
87
config = await config_service.get_config(param)
88

89
# Get configuration from specific namespace (set in ClientConfig)
90
param = ConfigParam(dataId="app.properties")  # group defaults to DEFAULT_GROUP
91
config = await config_service.get_config(param)
92
```
93

94
### Publish Configuration
95

96
Publish or update configuration values with optional encryption and filtering.
97

98
```python { .api }
99
async def publish_config(self, param: ConfigParam) -> bool:
100
    """
101
    Publish configuration to Nacos server with encryption and filtering support.
102
    
103
    Args:
104
        param (ConfigParam): Configuration parameters including data_id, group, and content
105
        
106
    Returns:
107
        bool: True if publication successful
108
        
109
    Raises:
110
        NacosException: If data_id is empty or invalid
111
    """
112
```
113

114
Usage example:
115

116
```python
117
from v2.nacos import ConfigParam
118

119
# Basic configuration publishing
120
param = ConfigParam(
121
    data_id="database.config",
122
    group="DEFAULT_GROUP",
123
    content="host=localhost\nport=5432\ndb=myapp"
124
)
125
success = await config_service.publish_config(param)
126
print(f"Publish successful: {success}")
127

128
# Publish JSON configuration with type hint
129
import json
130
config_data = {
131
    "database": {
132
        "host": "localhost",
133
        "port": 5432,
134
        "name": "myapp"
135
    }
136
}
137
param = ConfigParam(
138
    data_id="database.json",
139
    group="DEFAULT_GROUP",
140
    content=json.dumps(config_data),
141
    config_type="json"
142
)
143
await config_service.publish_config(param)
144

145
# Publish with application context
146
param = ConfigParam(
147
    data_id="app.properties",
148
    group="DEFAULT_GROUP",
149
    content="app.name=MyApplication\napp.version=1.0.0",
150
    app_name="MyApplication",
151
    config_type="properties"
152
)
153
await config_service.publish_config(param)
154
```
155

156
### Remove Configuration
157

158
Remove configuration from Nacos server.
159

160
```python { .api }
161
async def remove_config(self, param: ConfigParam):
162
    """
163
    Remove configuration from Nacos server.
164
    
165
    Args:
166
        param (ConfigParam): Configuration parameters with data_id and group
167
        
168
    Raises:
169
        NacosException: If data_id is empty or invalid
170
    """
171
```
172

173
Usage example:
174

175
```python
176
# Remove configuration
177
param = ConfigParam(dataId="old.config", group="DEFAULT_GROUP")
178
await config_service.remove_config(param)
179
print("Configuration removed")
180

181
# Remove from custom group
182
param = ConfigParam(dataId="temp.config", group="TEMP_GROUP")
183
await config_service.remove_config(param)
184
```
185

186
### Configuration Listeners
187

188
Add and manage asynchronous configuration change listeners.
189

190
```python { .api }
191
async def add_listener(self, data_id: str, group: str, listener: Callable) -> None:
192
    """
193
    Add configuration change listener.
194
    
195
    Args:
196
        data_id (str): Configuration data ID
197
        group (str): Configuration group
198
        listener (Callable): Async callback function for configuration changes
199
        
200
    Raises:
201
        NacosException: If data_id is empty or invalid
202
    """
203

204
async def remove_listener(self, data_id: str, group: str, listener: Callable):
205
    """
206
    Remove configuration change listener.
207
    
208
    Args:
209
        data_id (str): Configuration data ID
210
        group (str): Configuration group
211
        listener (Callable): Callback function to remove
212
    """
213
```
214

215
Usage example:
216

217
```python
218
# Define async listener
219
async def config_change_listener(config_content):
220
    print(f"Configuration changed: {config_content}")
221
    # Process configuration change asynchronously
222
    await process_config_change(config_content)
223

224
async def process_config_change(content):
225
    # Your async processing logic here
226
    print(f"Processing: {content}")
227

228
# Add listener
229
await config_service.add_listener(
230
    data_id="database.config",
231
    group="DEFAULT_GROUP", 
232
    listener=config_change_listener
233
)
234

235
# Add multiple listeners for the same config
236
async def backup_listener(config_content):
237
    print(f"Backup handler: {config_content}")
238
    await backup_config(config_content)
239

240
await config_service.add_listener(
241
    data_id="database.config",
242
    group="DEFAULT_GROUP",
243
    listener=backup_listener
244
)
245

246
# Remove specific listener
247
await config_service.remove_listener(
248
    data_id="database.config",
249
    group="DEFAULT_GROUP",
250
    listener=config_change_listener
251
)
252
```
253

254
### Server Health Check
255

256
Check the health status of Nacos server.
257

258
```python { .api }
259
async def server_health(self) -> bool:
260
    """
261
    Check if Nacos server is healthy.
262
    
263
    Returns:
264
        bool: True if server is healthy
265
    """
266
```
267

268
Usage example:
269

270
```python
271
# Check server health
272
is_healthy = await config_service.server_health()
273
if is_healthy:
274
    print("Nacos server is healthy")
275
else:
276
    print("Nacos server is unhealthy")
277
    # Handle unhealthy server scenario
278

279
# Health check in a monitoring loop
280
import asyncio
281

282
async def health_monitor():
283
    while True:
284
        try:
285
            healthy = await config_service.server_health()
286
            print(f"Server health: {'OK' if healthy else 'FAIL'}")
287
            await asyncio.sleep(30)  # Check every 30 seconds
288
        except Exception as e:
289
            print(f"Health check failed: {e}")
290
            await asyncio.sleep(10)  # Retry after 10 seconds
291

292
# Start health monitoring
293
asyncio.create_task(health_monitor())
294
```
295

296
### Service Shutdown
297

298
Properly shutdown the configuration service and cleanup resources.
299

300
```python { .api }
301
async def shutdown(self):
302
    """
303
    Shutdown the configuration service and cleanup resources.
304
    This should always be called when the service is no longer needed.
305
    """
306
```
307

308
Usage example:
309

310
```python
311
async def main():
312
    config_service = None
313
    try:
314
        # Create service
315
        client_config = ClientConfig(server_addresses="127.0.0.1:8848")
316
        config_service = await NacosConfigService.create_config_service(client_config)
317
        
318
        # Use service
319
        param = ConfigParam(dataId="test.config", group="DEFAULT_GROUP")
320
        config = await config_service.get_config(param)
321
        
322
    except Exception as e:
323
        print(f"Error: {e}")
324
    finally:
325
        # Always shutdown
326
        if config_service:
327
            await config_service.shutdown()
328

329
# Using context manager pattern
330
class ConfigServiceManager:
331
    def __init__(self, client_config):
332
        self.client_config = client_config
333
        self.service = None
334
        
335
    async def __aenter__(self):
336
        self.service = await NacosConfigService.create_config_service(self.client_config)
337
        return self.service
338
        
339
    async def __aexit__(self, exc_type, exc_val, exc_tb):
340
        if self.service:
341
            await self.service.shutdown()
342

343
# Usage with context manager
344
async def main():
345
    client_config = ClientConfig(server_addresses="127.0.0.1:8848")
346
    async with ConfigServiceManager(client_config) as config_service:
347
        param = ConfigParam(dataId="test.config", group="DEFAULT_GROUP")
348
        config = await config_service.get_config(param)
349
        print(config)
350
```
351

352
## Configuration Parameter Model
353

354
```python { .api }
355
class ConfigParam(BaseModel):
356
    dataId: str
357
    group: str = 'DEFAULT_GROUP'
358
    content: str = ''
359
    type: str = ''
360
    appName: str = ''
361
    tag: str = ''
362
    md5: str = ''
363
```
364

365
The ConfigParam model supports:
366

367
- **dataId**: Unique identifier for the configuration
368
- **group**: Configuration group (defaults to 'DEFAULT_GROUP')
369
- **content**: Configuration content (for publish operations)
370
- **type**: Configuration format hint ('json', 'yaml', 'properties', etc.)
371
- **appName**: Application name for context
372
- **tag**: Configuration tags for organization
373
- **md5**: MD5 hash for compare-and-swap operations
374

375
Usage examples:
376

377
```python
378
# Minimal configuration
379
param = ConfigParam(dataId="app.config")
380

381
# Full configuration for publishing
382
param = ConfigParam(
383
    dataId="database.json",
384
    group="DATABASE_GROUP",
385
    content='{"host": "localhost", "port": 5432}',
386
    type="json",
387
    appName="MyApp",
388
    tag="database,production"
389
)
390

391
# Configuration with CAS (Compare-And-Swap)
392
current_config = await config_service.get_config(
393
    ConfigParam(dataId="critical.config", group="DEFAULT_GROUP")
394
)
395
import hashlib
396
current_md5 = hashlib.md5(current_config.encode()).hexdigest()
397

398
param = ConfigParam(
399
    dataId="critical.config",
400
    group="DEFAULT_GROUP", 
401
    content="updated content",
402
    md5=current_md5  # Only update if current content matches this MD5
403
)
404
success = await config_service.publish_config(param)
405
```
406

407
## Advanced Features
408

409
### Encryption Support
410

411
When KMSConfig is enabled in ClientConfig, configurations are automatically encrypted/decrypted:
412

413
```python
414
from v2.nacos import ClientConfig, KMSConfig
415

416
# Configure KMS encryption
417
kms_config = KMSConfig(
418
    enabled=True,
419
    endpoint="https://kms.example.com",
420
    access_key="your-access-key",
421
    secret_key="your-secret-key"
422
)
423

424
client_config = ClientConfig(
425
    server_addresses="127.0.0.1:8848",
426
    kms_config=kms_config
427
)
428

429
config_service = await NacosConfigService.create_config_service(client_config)
430

431
# Configurations are automatically encrypted when published
432
# and decrypted when retrieved
433
param = ConfigParam(
434
    data_id="secret.config",
435
    content="sensitive data here"
436
)
437
await config_service.publish_config(param)  # Automatically encrypted
438

439
decrypted_config = await config_service.get_config(param)  # Automatically decrypted
440
```
441

442
### Configuration Filtering
443

444
Custom filters can be applied to configuration content:
445

446
```python
447
# Configuration filters are automatically applied based on ClientConfig
448
# They handle encryption, validation, and content transformation
449
# This is managed internally by the ConfigFilterChainManager
450
```
451

452
### Caching and Failover
453

454
The V2 API provides automatic caching with failover capabilities:
455

456
- **Local Cache**: Configurations are cached locally for improved performance
457
- **Failover Cache**: When server is unavailable, cached configurations are served
458
- **Cache Invalidation**: Cache is automatically updated when configurations change
459
- **Snapshot Support**: Persistent snapshots for disaster recovery
460

461
These features are enabled by default and managed automatically by the service.