Examples ======== Real-world examples showing vein in action. Multi-Stage Serverless API ------------------------- Managing configuration across development, staging, and production: .. code-block:: python from vein import config from vein.sources.aws import from_appconfig from vein.sources.memory import from_env import os # Load the stage from the environment config.stage = from_env("STAGE") # Load configuration for your stage config.update(from_appconfig("my-api", config.stage)) # Use the configuration @app.route("/api/v1/process") def process_request(): # Auto-refreshing rate limits if request_count > config.rate_limit: return {"error": "Rate limit exceeded"}, 429 # Feature flags without redeploy if config.features.new_algorithm: return process_v2(request.data) return process_v1(request.data) ML Model Configuration --------------------- Type-safe parameters with validation: .. code-block:: python from vein import config from vein.validation.policies import RangeValidator # Type-safe model parameters config.temperature = 0.8 config.temperature.add_validator(RangeValidator(0.0, 2.0)) config.max_tokens = 1000 config.model_name = "gpt-4" # Dynamic model switching @config.on_change("model_name") def reload_model(event): global model model = load_model(event.new_value) logger.info(f"Switched to model: {event.new_value}") Database Connection Management ----------------------------- Secure credential management with automatic recycling: .. code-block:: python from vein import config from vein.sources.aws import from_secrets_manager # Load secrets securely config.update(from_secrets_manager("prod/database")) # Automatic connection recycling on config change @config.on_change("database.connection_string") def update_connection_pool(event): connection_pool.reconfigure(event.new_value) # Use it naturally db = create_connection( host=config.database.host, port=config.database.port, password=config.database.password ) Feature Flags & A/B Testing -------------------------- Real-time feature flag updates: .. code-block:: python from vein import config from vein.sources.aws import from_appconfig from vein.cache import Cache from vein.cache.strategies import TTLCacheStrategy # Poll for updates every 30 seconds config.features = from_appconfig( "feature-flags", "production", cache=Cache(TTLCacheStrategy(seconds=30)) ) def should_show_new_ui(user_id: str) -> bool: # Real-time feature flag updates if not config.features.new_ui_enabled: return False # Percentage rollout if hash(user_id) % 100 < config.features.new_ui_percentage: return True return False Microservice Configuration ------------------------- Layer configurations from multiple sources: .. code-block:: python from vein import config from pydantic import BaseModel from vein.validation.policies import PydanticValidator from vein.sources import from_env, from_yaml # Define your schema with Pydantic class ServiceConfig(BaseModel): service_name: str port: int timeout_seconds: int = 30 retry_attempts: int = 3 # Load from multiple sources config.update(from_env()) # Environment variables config.update(from_yaml("config/base.yml")) # Base config config.update(from_yaml(f"config/{ENVIRONMENT}.yml")) # Environment overrides # Add validation config.service.add_validator(PydanticValidator(ServiceConfig)) # AWS native integration if ENVIRONMENT == "production": config.update(from_appconfig("myservice", "prod")) Environment Management --------------------- Switch between environments dynamically: .. code-block:: python from vein import Config # Create environment-specific configs config = Config( dev={"api_url": "http://localhost:8000", "debug": True}, staging={"api_url": "https://staging.api.com", "debug": True}, prod={"api_url": "https://api.com", "debug": False} ) # Switch environments dynamically config.set_environment("prod") print(config.api_url) # https://api.com print(config.debug) # False Loading from Files ----------------- Layer configurations from multiple files: .. code-block:: python from vein import config from vein.sources.files import from_yaml, from_json # Load base configuration config.update(from_yaml("config/base.yml")) # Load secrets config.update(from_json("config/secrets.json")) # Layer environment-specific configs for env in ["base", os.environ.get("ENV", "dev"), "local"]: config_file = f"config/{env}.yml" if os.path.exists(config_file): config.update(from_yaml(config_file)) FastAPI Integration ------------------ Using vein with FastAPI: .. code-block:: python from fastapi import FastAPI, Depends from vein import config from vein.sources import from_env app = FastAPI() # Load configuration config.update(from_env("APP_")) config.database.url = from_env("DATABASE_URL") # Dependency for request-scoped config def get_config(): return config @app.get("/health") def health_check(cfg=Depends(get_config)): return { "status": "healthy", "version": cfg.version, "environment": cfg.environment } @app.on_event("startup") async def startup_event(): # Validate configuration on startup if not config.database.url: raise ValueError("DATABASE_URL not configured")