CLI 验证模式与 Maybe Monad
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,将值传递给下一个函数,并返回其结果。如果当前结果是 Failure,bind 会原样返回该结果。
链的工作方式
| 输入 | 步骤 | 结果 |
|---|---|---|
"8080" | parsers.parse_int | Success(8080) |
.bind(validators.minimum(1)) | Success(8080) | |
.bind(validators.maximum(65535)) | Success(8080) | |
"70000" | parsers.parse_int | Success(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 的工作原理
- 显示提示 – 如果提供了默认值,则显示该默认值。
- 解析输入 – 使用提供的解析函数。
- 验证 – 运行验证器(如果有)。
- 失败时重试 – 当
retry为True或正整数时,显示错误并重复提示。 - 返回 – 包含最终结果的
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 模式的权衡
成本
| Issue | Explanation |
|---|---|
| 认知负担 | 不熟悉 monad 的开发者必须学习 bind、map、Success 和 Failure。函数式风格与典型的命令式 Python 不同。 |
| 堆栈追踪 | 管道深处的失败会指向 bind 的内部实现,而不是具体的验证器,使调试更加困难。必须提供描述性的错误信息。 |
| 类型推断 | 复杂的链式调用可能会让静态类型检查器困惑;显式的类型注解有助于缓解此问题。 |
| 对简单情况而言是大材小用 | 对于简单的验证,直接使用 if not value: raise ValueError() 更加直观。 |
何时值得使用
- 需要多个顺序验证步骤。
- 错误信息需要在管道中保持不变地传播。
- 验证逻辑应当能够单独进行测试。
- 相同的验证器在不同管道中被复用。
Maybe monad 将分散的条件判断转化为可组合的管道。每个验证器都是一个函数 T → Maybe[T],bind 将它们串联起来。逻辑运算符(&、|、~)用于组合验证器,而 ask 则加入交互式提示及重试逻辑。
安装与资源
pip install valid8r
- 源代码:
- 测试版本:
valid8r 1.25.0on Python 3.12.