Examples

Real-world examples showing vein in action.

Multi-Stage Serverless API

Managing configuration across development, staging, and production:

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:

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:

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:

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:

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:

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:

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:

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")