Custom Parsers and Validation

def add_parser(self, name: str, func: Callable):
    """
    Register a new parser method with the given name.
    
    Parameters:
    - name: str, method name for the parser (must not conflict with existing methods)
    - func: callable, parser function that receives the raw string value
    
    Raises:
    ParserConflictError: If name conflicts with existing method
    
    Notes:
    The parser function should accept a string value and return the parsed result.
    It can raise EnvError or ValidationError for invalid input.
    """

import os
from environs import env, EnvError
import re

# Custom email validator parser
def parse_email(value):
    """Parse and validate email addresses."""
    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    if not re.match(email_pattern, value):
        raise EnvError(f"Invalid email format: {value}")
    return value.lower()

# Register the custom parser
env.add_parser("email", parse_email)

# Use the custom parser
os.environ["ADMIN_EMAIL"] = "ADMIN@EXAMPLE.COM"
admin_email = env.email("ADMIN_EMAIL")  # => "admin@example.com"

# Custom URL slug parser
def parse_slug(value):
    """Parse and validate URL slugs."""
    slug_pattern = r'^[a-z0-9]+(?:-[a-z0-9]+)*$'
    if not re.match(slug_pattern, value):
        raise EnvError(f"Invalid slug format: {value}")
    return value

env.add_parser("slug", parse_slug)

os.environ["ARTICLE_SLUG"] = "my-awesome-article"
slug = env.slug("ARTICLE_SLUG")  # => "my-awesome-article"

# Custom coordinate parser
def parse_coordinates(value):
    """Parse latitude,longitude coordinates."""
    try:
        lat_str, lon_str = value.split(',')
        lat, lon = float(lat_str.strip()), float(lon_str.strip())
        if not (-90 <= lat <= 90):
            raise ValueError("Latitude must be between -90 and 90")
        if not (-180 <= lon <= 180):
            raise ValueError("Longitude must be between -180 and 180")
        return (lat, lon)
    except (ValueError, TypeError) as e:
        raise EnvError(f"Invalid coordinates format: {value}") from e

env.add_parser("coordinates", parse_coordinates)

os.environ["LOCATION"] = "40.7128, -74.0060"
location = env.coordinates("LOCATION")  # => (40.7128, -74.0060)

def parser_for(self, name: str):
    """
    Decorator that registers a new parser method with the given name.
    
    Parameters:
    - name: str, method name for the parser
    
    Returns:
    Decorator function that registers the decorated function as a parser
    
    Notes:
    The decorated function should accept a string value and return parsed result.
    """

import os
from environs import env, EnvError
import ipaddress

# Register IP address parser using decorator
@env.parser_for("ip_address")
def parse_ip_address(value):
    """Parse and validate IP addresses."""
    try:
        return ipaddress.ip_address(value)
    except ipaddress.AddressValueError as e:
        raise EnvError(f"Invalid IP address: {value}") from e

# Register network parser
@env.parser_for("network")
def parse_network(value):
    """Parse and validate network CIDR notation."""
    try:
        return ipaddress.ip_network(value, strict=False)
    except ipaddress.NetmaskValueError as e:
        raise EnvError(f"Invalid network: {value}") from e

# Use custom parsers
os.environ["SERVER_IP"] = "192.168.1.100"
os.environ["ALLOWED_NETWORK"] = "10.0.0.0/8"

server_ip = env.ip_address("SERVER_IP")        # => IPv4Address('192.168.1.100')
allowed_net = env.network("ALLOWED_NETWORK")   # => IPv4Network('10.0.0.0/8')

print(f"Server IP: {server_ip}")
print(f"Network contains server: {server_ip in allowed_net}")

# Complex data structure parser
@env.parser_for("connection_string")
def parse_connection_string(value):
    """Parse database connection strings."""
    # Format: "host:port/database?param1=value1&param2=value2"
    try:
        # Split main parts
        if '?' in value:
            main_part, params_part = value.split('?', 1)
        else:
            main_part, params_part = value, ""
        
        # Parse host:port/database
        host_port, database = main_part.split('/', 1)
        if ':' in host_port:
            host, port = host_port.split(':', 1)
            port = int(port)
        else:
            host, port = host_port, 5432
        
        # Parse parameters
        params = {}
        if params_part:
            for param in params_part.split('&'):
                key, val = param.split('=', 1)
                params[key] = val
        
        return {
            'host': host,
            'port': port,
            'database': database,
            'params': params
        }
    except (ValueError, TypeError) as e:
        raise EnvError(f"Invalid connection string: {value}") from e

# Use complex parser
os.environ["DATABASE_URL"] = "localhost:5432/myapp?ssl=require&timeout=30"
db_config = env.connection_string("DATABASE_URL")
# => {'host': 'localhost', 'port': 5432, 'database': 'myapp', 
#     'params': {'ssl': 'require', 'timeout': '30'}}

def add_parser_from_field(self, name: str, field_cls: Type[Field]):
    """
    Register a new parser method from a marshmallow Field class.
    
    Parameters:
    - name: str, method name for the parser
    - field_cls: marshmallow Field class or subclass
    
    Notes:
    The field class should implement _deserialize method for parsing.
    Enables full marshmallow validation capabilities and error handling.
    """

import os
from environs import env
import marshmallow as ma
from marshmallow import ValidationError
import re

# Custom marshmallow field for phone numbers
class PhoneNumberField(ma.fields.Field):
    """Field for parsing and validating phone numbers."""
    
    def _serialize(self, value, attr, obj, **kwargs):
        if value is None:
            return None
        return str(value)
    
    def _deserialize(self, value, attr, data, **kwargs):
        if not isinstance(value, str):
            raise ValidationError("Phone number must be a string")
        
        # Remove all non-digit characters
        digits = re.sub(r'\D', '', value)
        
        # Validate US phone number format (10 digits)
        if len(digits) != 10:
            raise ValidationError("Phone number must have 10 digits")
        
        # Format as (XXX) XXX-XXXX
        return f"({digits[:3]}) {digits[3:6]}-{digits[6:]}"

# Register marshmallow field as parser
env.add_parser_from_field("phone", PhoneNumberField)

# Use field-based parser
os.environ["SUPPORT_PHONE"] = "1-800-555-0123"
support_phone = env.phone("SUPPORT_PHONE")  # => "(800) 555-0123"

# Custom field with validation options
class CreditCardField(ma.fields.Field):
    """Field for parsing credit card numbers with validation."""
    
    def _deserialize(self, value, attr, data, **kwargs):
        if not isinstance(value, str):
            raise ValidationError("Credit card number must be a string")
        
        # Remove spaces and dashes
        digits = re.sub(r'[\s-]', '', value)
        
        # Validate digits only
        if not digits.isdigit():
            raise ValidationError("Credit card number must contain only digits")
        
        # Validate length (13-19 digits for most cards)
        if not (13 <= len(digits) <= 19):
            raise ValidationError("Credit card number must be 13-19 digits")
        
        # Simple Luhn algorithm check
        def luhn_check(card_num):
            def digits_of(n):
                return [int(d) for d in str(n)]
            
            digits = digits_of(card_num)
            odd_digits = digits[-1::-2]
            even_digits = digits[-2::-2]
            checksum = sum(odd_digits)
            for d in even_digits:
                checksum += sum(digits_of(d * 2))
            return checksum % 10 == 0
        
        if not luhn_check(digits):
            raise ValidationError("Invalid credit card number")
        
        # Mask all but last 4 digits
        masked = '*' * (len(digits) - 4) + digits[-4:]
        return masked

env.add_parser_from_field("credit_card", CreditCardField)

# Use with validation
os.environ["PAYMENT_CARD"] = "4532 1488 0343 6467"  # Valid test number
card = env.credit_card("PAYMENT_CARD")  # => "************6467"

# Import validation functions
from environs import validate

# Common validators available:
# validate.Length(min=None, max=None)
# validate.Range(min=None, max=None)
# validate.OneOf(choices)
# validate.Regexp(regex)
# validate.URL()
# validate.Email()

import os
from environs import env, validate

# String length validation
os.environ["USERNAME"] = "john_doe"
username = env.str("USERNAME", validate=validate.Length(min=3, max=20))

# Numeric range validation
os.environ["PORT"] = "8080"
port = env.int("PORT", validate=validate.Range(min=1024, max=65535))

# Choice validation
os.environ["LOG_LEVEL"] = "INFO"
log_level = env.str("LOG_LEVEL", validate=validate.OneOf(["DEBUG", "INFO", "WARNING", "ERROR"]))

# Regular expression validation
os.environ["VERSION"] = "1.2.3"
version = env.str("VERSION", validate=validate.Regexp(r'^\d+\.\d+\.\d+$'))

# Multiple validators
os.environ["API_KEY"] = "sk_live_123456789abcdef"
api_key = env.str("API_KEY", validate=[
    validate.Length(min=20),
    validate.Regexp(r'^sk_(live|test)_[a-zA-Z0-9]+$')
])

# Custom validation function
def validate_positive_even(value):
    """Validate that number is positive and even."""
    if value <= 0:
        raise ValidationError("Value must be positive")
    if value % 2 != 0:
        raise ValidationError("Value must be even")
    return value

os.environ["BATCH_SIZE"] = "100"
batch_size = env.int("BATCH_SIZE", validate=validate_positive_even)

# Combining built-in and custom validators
os.environ["THREAD_COUNT"] = "8"
thread_count = env.int("THREAD_COUNT", validate=[
    validate.Range(min=1, max=32),
    validate_positive_even
])

from environs import env, ParserConflictError, EnvValidationError

# Handle parser name conflicts
try:
    env.add_parser("str", lambda x: x)  # Conflicts with built-in
except ParserConflictError as e:
    print(f"Parser conflict: {e}")
    # Use a different name
    env.add_parser("custom_str", lambda x: x.upper())

# Handle validation errors in custom parsers
@env.parser_for("positive_int")
def parse_positive_int(value):
    try:
        num = int(value)
        if num <= 0:
            raise EnvError("Value must be positive")
        return num
    except ValueError as e:
        raise EnvError(f"Invalid integer: {value}") from e

# Use with error handling
os.environ["INVALID_POSITIVE"] = "-5"
try:
    value = env.positive_int("INVALID_POSITIVE")
except EnvValidationError as e:
    print(f"Validation failed: {e}")

from typing import Callable, Any, Type
from marshmallow.fields import Field
from marshmallow import ValidationError

ParserFunction = Callable[[str], Any]
ValidatorFunction = Callable[[Any], Any]