Advanced Features¶
This guide covers advanced vein features for production use.
Validation¶
Built-in Validators¶
vein provides several built-in validators:
from vein import config
from vein.validation.policies import (
RangeValidator,
RegexValidator,
TypeValidator,
RequiredValidator
)
# Numeric range validation
config.port = 8080
config.port.add_validator(RangeValidator(1, 65535))
# Regex pattern validation
config.email = "admin@company.com"
config.email.add_validator(RegexValidator(r".*@company\.com$"))
# Type validation
config.workers = 4
config.workers.add_validator(TypeValidator(int))
# Required fields
config.api_key.add_validator(RequiredValidator())
Pydantic Integration¶
Use your existing Pydantic models for validation:
from pydantic import BaseModel, Field
from vein import config
from vein.validation.policies import PydanticValidator
class DatabaseConfig(BaseModel):
host: str
port: int = Field(ge=1, le=65535)
username: str
password: str
ssl_enabled: bool = True
pool_size: int = Field(default=10, ge=1, le=100)
# Apply Pydantic validation
config.database.add_validator(PydanticValidator(DatabaseConfig))
# This will validate against the schema
config.database.update({
"host": "localhost",
"port": 5432,
"username": "admin",
"password": "secret"
})
Custom Validators¶
Create your own validation logic:
from vein.validation.policies import ValidationPolicy
class EnvironmentValidator(ValidationPolicy):
def __init__(self, allowed_environments):
self.allowed = allowed_environments
def validate(self, value):
if value not in self.allowed:
raise ValueError(f"Environment must be one of: {self.allowed}")
return value
config.environment = "production"
config.environment.add_validator(
EnvironmentValidator(["development", "staging", "production"])
)
Caching Strategies¶
TTL (Time-To-Live) Cache¶
Cache values for a specific duration:
from vein import config
from vein.sources.aws import from_parameter_store
from vein.cache import Cache
from vein.cache.strategies import TTLCacheStrategy
# Cache for 5 minutes
config.update(
from_parameter_store(
"/myapp/prod/",
cache=Cache(TTLCacheStrategy(minutes=5))
)
)
LRU (Least Recently Used) Cache¶
Cache with size limits:
from vein.cache.strategies import LRUCacheStrategy
# Keep 100 most recently used values
config.update(
from_parameter_store(
"/myapp/prod/",
cache=Cache(LRUCacheStrategy(max_size=100))
)
)
Custom Cache Strategy¶
Implement your own caching logic:
from vein.cache.strategies import CacheStrategy
class BusinessHoursCacheStrategy(CacheStrategy):
def should_refresh(self, key, cached_time):
# Refresh more frequently during business hours
now = datetime.now()
if 9 <= now.hour < 17: # 9 AM - 5 PM
return (now - cached_time).seconds > 300 # 5 minutes
else:
return (now - cached_time).seconds > 3600 # 1 hour
Event Handling¶
Change Events¶
React to configuration changes:
from vein import config
# Listen to specific key
@config.on_change("database.host")
def handle_db_change(event):
print(f"Database host changed from {event.old_value} to {event.new_value}")
# Reconnect to database
reconnect_database(event.new_value)
# Listen to pattern
@config.on_change("features.*")
def handle_feature_change(event):
print(f"Feature flag {event.key} changed")
clear_feature_cache()
# Listen to all changes
@config.on_change("*")
def log_all_changes(event):
logger.info(f"Config change: {event.key} = {event.new_value}")
Event Object Properties¶
@config.on_change("api_key")
def handle_change(event):
# Available properties:
event.key # "api_key"
event.old_value # Previous value
event.new_value # New value
event.timestamp # When the change occurred
event.source # Where the change came from
State Tracking¶
Enable History¶
Track configuration changes over time:
from vein import config
# Enable state tracking
config.enable_state_tracking()
# Make some changes
config.api_key = "key1"
config.api_key = "key2"
config.api_key = "key3"
# Get history
history = config.api_key.get_history()
for change in history:
print(f"{change.timestamp}: {change.old_value} -> {change.new_value}")
Rollback Changes¶
Revert to previous states:
# Rollback specific key
config.api_key.rollback(steps=1) # Go back one change
# Rollback entire config to timestamp
config.rollback_to(timestamp="2024-01-10T10:00:00")
# Rollback with safety check
if config.can_rollback("api_key", steps=2):
config.api_key.rollback(steps=2)
AWS Integration¶
AppConfig with Polling¶
Automatically poll for configuration updates:
from vein import config
from vein.sources.aws import from_appconfig
# Poll every 30 seconds
config.update(
from_appconfig(
application="my-app",
environment="production",
profile="backend-config",
poll_interval=30 # seconds
)
)
Parameter Store with Prefix¶
Load all parameters with a prefix:
from vein.sources.aws import from_parameter_store
# Load all parameters under /myapp/prod/
config.update(
from_parameter_store(
prefix="/myapp/prod/",
recursive=True,
decrypt=True # Decrypt SecureString parameters
)
)
Secrets Manager¶
Load secrets with automatic JSON parsing:
from vein.sources.aws import from_secrets_manager
# Load and parse JSON secret
config.database.update(
from_secrets_manager("prod/database/credentials")
)
# Now use the parsed values
print(config.database.username)
print(config.database.password)
Error Handling¶
Retry Policies¶
Configure retry behavior for remote sources:
from vein import config
from vein.backoff.policies import ExponentialBackoff
config.update(
from_parameter_store(
"/myapp/prod/",
retry_policy=ExponentialBackoff(
initial_interval=1,
max_interval=60,
max_retries=5
)
)
)
Fallback Values¶
Provide defaults when sources fail:
from vein import config
# Set defaults before loading
config.api_url = "http://localhost:8000" # fallback
config.timeout = 30 # fallback
try:
config.update(from_appconfig("my-app", "prod"))
except Exception as e:
logger.warning(f"Failed to load config: {e}, using defaults")
Performance Optimization¶
Lazy Loading¶
Load configuration values only when accessed:
from vein import config
from vein.sources import LazySource
# Expensive operation only runs when accessed
config.expensive_data = LazySource(
lambda: fetch_large_dataset()
)
# Not loaded yet
print("Config initialized")
# Loads now
data = config.expensive_data.value
Batch Updates¶
Update multiple values efficiently:
# Batch updates are atomic and efficient
config.update({
"api": {
"url": "https://api.example.com",
"key": "secret",
"version": "v2"
},
"database": {
"host": "db.example.com",
"port": 5432
}
})
Next Steps¶
Review the API Reference for complete API documentation
Check the Flow Architecture to understand the underlying engine
Join our community for support and updates