Validation

from environs import env, validate, ValidationError, EnvValidationError

validate.OneOf(choices, error=None): ...      # Value must be one of the choices
validate.Range(min=None, max=None): ...      # Numeric range validation
validate.Length(min=None, max=None): ...     # String/list length validation
validate.Email(): ...                        # Email format validation
validate.URL(require_tld=True): ...          # URL format validation
validate.Regexp(regex, flags=0): ...         # Regular expression validation
validate.Equal(comparable): ...              # Equality validation
validate.NoneOf(iterable): ...               # Value must not be in iterable
validate.ContainsOnly(choices): ...          # All items must be in choices
validate.Predicate(method, error=None): ...  # Custom predicate function

import os
from environs import env, validate, ValidationError

# Choice validation
os.environ["NODE_ENV"] = "development"
node_env = env.str(
    "NODE_ENV", 
    validate=validate.OneOf(["development", "staging", "production"])
)  # => "development"

# Range validation
os.environ["MAX_CONNECTIONS"] = "50"
max_connections = env.int(
    "MAX_CONNECTIONS",
    validate=validate.Range(min=1, max=100)
)  # => 50

# String length validation
os.environ["API_KEY"] = "abc123def456"
api_key = env.str(
    "API_KEY",
    validate=validate.Length(min=8, max=64)
)  # => "abc123def456"

# Email validation
os.environ["ADMIN_EMAIL"] = "admin@example.com"
admin_email = env.str(
    "ADMIN_EMAIL",
    validate=validate.Email()
)  # => "admin@example.com"

# URL validation
os.environ["WEBHOOK_URL"] = "https://api.example.com/webhook"
webhook_url = env.str(
    "WEBHOOK_URL",
    validate=validate.URL()
)  # => "https://api.example.com/webhook"

import os
from environs import env, validate

# Multiple string validators
os.environ["USERNAME"] = "john_doe"
username = env.str(
    "USERNAME",
    validate=[
        validate.Length(min=3, max=20),
        validate.Regexp(r'^[a-zA-Z0-9_]+$', error="Username must contain only letters, numbers, and underscores")
    ]
)  # => "john_doe"

# Multiple numeric validators  
os.environ["PORT"] = "8080"
port = env.int(
    "PORT",
    validate=[
        validate.Range(min=1024, max=65535, error="Port must be between 1024 and 65535"),
        validate.NoneOf([3000, 5000], error="Port cannot be 3000 or 5000")
    ]
)  # => 8080

# List validation with item validation
os.environ["ALLOWED_HOSTS"] = "localhost,api.example.com,admin.example.com"
allowed_hosts = env.list(
    "ALLOWED_HOSTS",
    validate=validate.Length(min=1, max=10)  # Validate list length
)
# Individual items validated separately if needed

# Complex validation example
os.environ["DATABASE_NAME"] = "myapp_production"
db_name = env.str(
    "DATABASE_NAME",
    validate=[
        validate.Length(min=5, max=63),
        validate.Regexp(r'^[a-zA-Z][a-zA-Z0-9_]*$', error="Database name must start with letter"),
        validate.NoneOf(["test", "temp", "debug"], error="Reserved database names not allowed")
    ]
)  # => "myapp_production"

import os
from environs import env, ValidationError

def validate_positive_even(value):
    """Custom validator for positive even numbers."""
    if value <= 0:
        raise ValidationError("Value must be positive")
    if value % 2 != 0:
        raise ValidationError("Value must be even")
    return value

def validate_version_format(value):
    """Custom validator for semantic version format."""
    import re
    if not re.match(r'^\d+\.\d+\.\d+$', value):
        raise ValidationError("Version must be in format X.Y.Z")
    return value

def validate_file_extension(extensions):
    """Factory function for file extension validation."""
    def validator(value):
        if not any(value.endswith(ext) for ext in extensions):
            raise ValidationError(f"File must have one of these extensions: {', '.join(extensions)}")
        return value
    return validator

# Usage examples
os.environ["WORKER_COUNT"] = "4"
worker_count = env.int(
    "WORKER_COUNT",
    validate=validate_positive_even
)  # => 4

os.environ["APP_VERSION"] = "1.2.3"
app_version = env.str(
    "APP_VERSION",
    validate=validate_version_format
)  # => "1.2.3"

os.environ["CONFIG_FILE"] = "settings.json"
config_file = env.str(
    "CONFIG_FILE",
    validate=validate_file_extension(['.json', '.yaml', '.yml'])
)  # => "settings.json"

# Combining custom and built-in validators
os.environ["BACKUP_COUNT"] = "6"
backup_count = env.int(
    "BACKUP_COUNT",
    validate=[
        validate.Range(min=1, max=10),
        validate_positive_even
    ]
)  # => 6

import os
from environs import Env, EnvValidationError, validate

# Set up invalid environment variables
os.environ["INVALID_PORT"] = "999"           # Too low
os.environ["INVALID_EMAIL"] = "not-email"    # Invalid format  
os.environ["INVALID_ENV"] = "invalid"        # Not in allowed choices

# Create env with deferred validation
env = Env(eager=False)

# Parse variables (errors collected, not raised)
port = env.int(
    "INVALID_PORT",
    validate=validate.Range(min=1024, max=65535)
)

email = env.str(
    "INVALID_EMAIL", 
    validate=validate.Email()
)

environment = env.str(
    "INVALID_ENV",
    validate=validate.OneOf(["development", "staging", "production"])
)

# Validate all at once
try:
    env.seal()
except EnvValidationError as e:
    print("Validation errors found:")
    for var_name, errors in e.error_messages.items():
        print(f"  {var_name}: {', '.join(errors)}")
    # Handle errors appropriately

import os
from environs import env, validate, EnvValidationError

# Set invalid value
os.environ["INVALID_COUNT"] = "-5"

try:
    count = env.int(
        "INVALID_COUNT",
        validate=[
            validate.Range(min=0, max=100, error="Count must be between 0 and 100"),
            lambda x: x if x % 2 == 0 else ValidationError("Count must be even")
        ]
    )
except EnvValidationError as e:
    print(f"Validation failed for INVALID_COUNT: {e}")
    print(f"Error messages: {e.error_messages}")
    
    # Provide fallback value
    count = 10
    print(f"Using fallback value: {count}")

# Graceful degradation
def get_validated_config():
    """Get configuration with validation and fallbacks."""
    config = {}
    
    # Required setting with validation
    try:
        config['port'] = env.int(
            "PORT",
            validate=validate.Range(min=1024, max=65535)
        )
    except EnvValidationError:
        raise RuntimeError("Valid PORT is required")
    
    # Optional setting with validation and fallback
    try:
        config['max_workers'] = env.int(
            "MAX_WORKERS",
            validate=validate.Range(min=1, max=32)
        )
    except EnvValidationError as e:
        print(f"Invalid MAX_WORKERS: {e}, using default")
        config['max_workers'] = 4
    
    return config

# Usage
os.environ["PORT"] = "8080"
os.environ["MAX_WORKERS"] = "invalid"

try:
    config = get_validated_config()
    print(f"Configuration: {config}")
except RuntimeError as e:
    print(f"Configuration error: {e}")

import os
from environs import env, validate, ValidationError

def validate_database_url(url):
    """Validate database URL format and supported schemes."""
    from urllib.parse import urlparse
    
    parsed = urlparse(url)
    supported_schemes = ['postgresql', 'mysql', 'sqlite', 'oracle']
    
    if parsed.scheme not in supported_schemes:
        raise ValidationError(f"Unsupported database scheme. Use: {', '.join(supported_schemes)}")
    
    if parsed.scheme != 'sqlite' and not parsed.hostname:
        raise ValidationError("Database URL must include hostname")
        
    return url

def validate_json_config(value):
    """Validate that string contains valid JSON configuration."""
    import json
    try:
        config = json.loads(value)
        if not isinstance(config, dict):
            raise ValidationError("JSON must be an object")
        
        # Validate required keys
        required_keys = ['name', 'version']
        for key in required_keys:
            if key not in config:
                raise ValidationError(f"JSON must contain '{key}' field")
                
        return value
    except json.JSONDecodeError:
        raise ValidationError("Value must be valid JSON")

# Environment-specific validation
def create_environment_validator():
    """Create validator based on current environment."""
    current_env = os.environ.get('NODE_ENV', 'development')
    
    if current_env == 'production':
        # Strict validation for production
        return [
            validate.Length(min=32, max=128),
            validate.Regexp(r'^[A-Za-z0-9+/=]+$', error="Must be base64 encoded")
        ]
    else:
        # Relaxed validation for development
        return validate.Length(min=8)

# Usage examples
os.environ["DATABASE_URL"] = "postgresql://user:pass@localhost:5432/mydb"
db_url = env.str(
    "DATABASE_URL",
    validate=validate_database_url
)

os.environ["APP_CONFIG"] = '{"name": "MyApp", "version": "1.0.0", "debug": true}'
app_config = env.str(
    "APP_CONFIG",
    validate=validate_json_config
)

os.environ["NODE_ENV"] = "production"
os.environ["SECRET_KEY"] = "YWJjZGVmZ2hpamtsbW5vcHFyc3R1dnd4eXoxMjM0NTY3ODkw"
secret_key = env.str(
    "SECRET_KEY",
    validate=create_environment_validator()
)

from typing import Any, Callable, List, Union
from marshmallow import ValidationError

ValidatorFunction = Callable[[Any], Any]
ValidatorList = List[ValidatorFunction]
Validator = Union[ValidatorFunction, ValidatorList]

class ValidationError(Exception):
    """Raised by validators when validation fails."""
    def __init__(self, message: str): ...

class EnvValidationError(Exception):
    """Raised when environment variable validation fails."""
    def __init__(self, message: str, error_messages): ...
    error_messages: dict | list  # Detailed error information