CLI 验证模式与 Maybe Monad

发布: (2026年1月9日 GMT+8 09:39)
8 min read
原文: Dev.to

Source: Dev.to

示例:验证配置文件路径

输入必须通过多个检查:

  • 路径存在
  • 它是文件(而不是目录)
  • 可读
  • 具有 .json 扩展名
  • 包含具有必需键的有效 JSON
import os
import json

def validate_config_file(path: str) -> dict:
    if not os.path.exists(path):
        raise ValueError(f"{path} does not exist")
    if not os.path.isfile(path):
        raise ValueError(f"{path} is not a file")
    if not os.access(path, os.R_OK):
        raise ValueError(f"{path} is not readable")
    if not path.endswith('.json'):
        raise ValueError(f"{path} must be .json")
    try:
        with open(path) as f:
            config = json.load(f)
    except json.JSONDecodeError as e:
        raise ValueError(f"Invalid JSON: {e}")
    required = ['database', 'api_key']
    missing = [k for k in required if k not in config]
    if missing:
        raise ValueError(f"Missing keys: {missing}")
    return config

函数可以工作,但存在结构性问题:

  • 每个检查都是独立的 if‑语句,添加或删除检查需要修改多个位置。
  • JSON 解析位于嵌套的 try/except 中。
  • 测试需要模拟文件系统状态或创建实际文件。

Maybe Monad 解决方案

Maybe monad 将每个验证步骤表示为返回 Success(value)Failure(error) 的函数。步骤通过 bind 组合:如果前一步成功,则把值传递给下一个函数;如果前一步失败,则短路,不再继续。

from valid8r import parsers, validators
from valid8r.core.maybe import Maybe, Success, Failure

def validate_port(text: str) -> Maybe[int]:
    """Parse and validate a port number in range 1‑65535."""
    return (
        parsers.parse_int(text)
        .bind(validators.minimum(1))
        .bind(validators.maximum(65535))
    )

# Usage
result = validate_port("8080")
match result:
    case Success(port):
        print(f"Starting server on port {port}")
    case Failure(error):
        print(f"Invalid port: {error}")

每个验证器都是接受一个值并返回 Maybe[T] 的函数。bind 方法会解包 Success,将值传递给下一个函数,并返回其结果。如果当前结果是 Failurebind 会原样返回该结果。

链的工作方式

输入步骤结果
"8080"parsers.parse_intSuccess(8080)
.bind(validators.minimum(1))Success(8080)
.bind(validators.maximum(65535))Success(8080)
"70000"parsers.parse_intSuccess(70000)
.bind(validators.minimum(1))Success(70000)
.bind(validators.maximum(65535))Failure("Value must be at most 65535")

链在第一次失败时停止;后续的验证器不会被执行。

Source:

使用逻辑运算符组合验证器

Validator 类支持 &(与)、|(或)和 ~(非)运算符,以构建更复杂的规则。

from valid8r import parsers, validators
from valid8r.core.maybe import Maybe

def validate_username(text: str) -> Maybe[str]:
    """Username: 3‑20 chars, alphanumeric with underscores."""
    return parsers.parse_str(text).bind(
        validators.length(3, 20)
        & validators.matches_regex(r'^[a-zA-Z0-9_]+$')
    )

def validate_age(text: str) -> Maybe[int]:
    """Age: positive integer, max 150."""
    return parsers.parse_int(text).bind(
        validators.minimum(0) & validators.maximum(150)
    )
  • & 创建一个仅在 两个 验证器都通过时才通过的验证器。
  • |任意一个 验证器通过时即通过。
  • ~ 对验证器取反(即,当被包装的验证器失败时成功)。
from pathlib import Path
from valid8r import parsers, validators
from valid8r.core.maybe import Maybe

def validate_output_path(text: str) -> Maybe[Path]:
    """Output path: must be a directory **or** a writable file."""
    return (
        parsers.parse_path(text, resolve=True)
        .bind(validators.exists())
        .bind(validators.is_dir() | (validators.is_file() & validators.is_writable()))
    )

def validate_safe_upload(text: str) -> Maybe[Path]:
    """Upload: must exist, be readable, **NOT** be executable."""
    return parsers.parse_path(text).bind(
        validators.exists()
        & validators.is_readable()
        & ~validators.is_executable()
    )

~validators.is_executable() 验证器在文件 不是 可执行时成功,从而实现对原始检查的取反。

文件系统验证管道

文件系统验证通常需要多个顺序检查:存在性、类型、权限以及内容约束。Maybe 模式通过可组合的管道来处理这些检查。

from pathlib import Path
from valid8r import parsers, validators
from valid8r.core.maybe import Maybe, Success, Failure

def validate_config_file(path_str: str) -> Maybe[Path]:
    """Validate configuration file: exists, readable, YAML/JSON, under 1 MB."""
    return (
        parsers.parse_path(path_str, expand_user=True, resolve=True)
        .bind(validators.exists())
        .bind(validators.is_file())
        .bind(validators.is_readable())
        .bind(validators.has_extension(['.yaml', '.yml', '.json']))
        .bind(validators.max_size(1024 * 1024))
    )

def validate_upload_file(path_str: str) -> Maybe[Path]:
    """Validate uploaded file: PDF/DOCX, readable, under 10 MB."""
    return (
        parsers.parse_path(path_str)
        .bind(validators.exists())
        .bind(validators.is_file())
        .bind(validators.has_extension(['.pdf', '.docx']))
        .bind(validators.is_readable())
        .bind(validators.max_size(10 * 1024 * 1024))
    )

每个管道从左到右自然读取,任何失败都会短路后续链,返回带有帮助性错误信息的 Failure

验证管道

d(validators.is_readable()) \
    .bind(validators.has_extension(['.pdf', '.docx'])) \
    .bind(validators.max_size(10 * 1024 * 1024))

在 CLI 处理程序中的用法

def handle_upload(file_path: str) -> dict:
    match validate_upload_file(file_path):
        case Success(path):
            return {
                'status': 'success',
                'filename': path.name,
                'size': path.stat().st_size
            }
        case Failure(error):
            return {
                'status': 'error',
                'message': error
            }

parse_path 函数支持诸如 expand_user=True(将 ~ 展开为用户主目录)和 resolve=True(转换为绝对路径)等选项。这些会在验证开始前执行。

  • 添加或删除验证步骤只需更改 一行
  • 每个步骤都是独立可测试的。
  • 错误信息会自动传播,并带有哪个检查失败的上下文。

使用 ask 进行交互式提示

from valid8r import parsers, validators
from valid8r.prompt import ask

def get_user_config() -> dict:
    """Prompt user for configuration with validation and retry."""
    
    # Port with range validation, retry up to 3 times
    port_result = ask(
        "Enter port (1-65535): ",
        parser=parsers.parse_int,
        validator=validators.between(1, 65535),
        default=8080,
        retry=3
    )
    
    # Email with RFC validation, unlimited retries
    email_result = ask(
        "Enter email: ",
        parser=parsers.parse_email,
        retry=True
    )
    
    # Boolean with various formats accepted (yes/no, true/false, y/n, 1/0)
    debug_result = ask(
        "Enable debug mode? ",
        parser=parsers.parse_bool,
        default=False
    )
    
    return {
        'port':  port_result.value_or(8080),
        'email': email_result.value_or(None),
        'debug': debug_result.value_or(False)
    }

ask 的工作原理

  1. 显示提示 – 如果提供了默认值,则显示该默认值。
  2. 解析输入 – 使用提供的解析函数。
  3. 验证 – 运行验证器(如果有)。
  4. 失败时重试 – 当 retryTrue 或正整数时,显示错误并重复提示。
  5. 返回 – 包含最终结果的 Maybe[T]

自定义验证管道

使用 bind 组合带有验证器的解析器:

def custom_port_parser(text: str) -> Maybe[int]:
    return parsers.parse_int(text).bind(validators.between(1, 65535))

port_result = ask(
    "Enter port: ",
    parser=custom_port_parser,
    retry=True
)

Source:

Argparse 集成

type_from_parser 适配器将 valid8r 解析器连接到 argparse

import argparse
from valid8r import parsers, validators
from valid8r.integrations.argparse import type_from_parser
from valid8r.core.maybe import Maybe

def port_parser(text: str) -> Maybe[int]:
    return parsers.parse_int(text).bind(
        validators.minimum(1) & validators.maximum(65535)
    )

parser = argparse.ArgumentParser()
parser.add_argument(
    '--email',
    type=type_from_parser(parsers.parse_email),
    required=True,
    help='Email address'
)
parser.add_argument(
    '--port',
    type=type_from_parser(port_parser),
    default=8080,
    help='Port number (1-65535)'
)

args = parser.parse_args()
# args.email → EmailAddress(local='user', domain='example.com')
# args.port  → int

当验证失败时,argparse 会显示来自 Failure 的错误信息并以状态码 2 退出。该信息来源于验证器,而不是通用的类型转换错误。

Maybe‑Monad 模式的权衡

成本

IssueExplanation
认知负担不熟悉 monad 的开发者必须学习 bindmapSuccessFailure。函数式风格与典型的命令式 Python 不同。
堆栈追踪管道深处的失败会指向 bind 的内部实现,而不是具体的验证器,使调试更加困难。必须提供描述性的错误信息。
类型推断复杂的链式调用可能会让静态类型检查器困惑;显式的类型注解有助于缓解此问题。
对简单情况而言是大材小用对于简单的验证,直接使用 if not value: raise ValueError() 更加直观。

何时值得使用

  • 需要多个顺序验证步骤
  • 错误信息需要在管道中保持不变地传播
  • 验证逻辑应当能够单独进行测试
  • 相同的验证器在不同管道中被复用

Maybe monad 将分散的条件判断转化为可组合的管道。每个验证器都是一个函数 T → Maybe[T]bind 将它们串联起来。逻辑运算符(&|~)用于组合验证器,而 ask 则加入交互式提示及重试逻辑。

安装与资源

pip install valid8r
  • 源代码:
  • 测试版本: valid8r 1.25.0 on Python 3.12.
Back to Blog

相关文章

阅读更多 »