什么是 CEL?
通用表达式语言(Common Expression Language,简称 CEL)是谷歌设计的一种非图灵完备的嵌入式策略和表达式语言。它的核心设计理念是“简洁、高速、安全、可移植”,特别适合嵌入到需要策略评估、数据验证和动态配置的系统中。
CEL 最吸引人的地方在于它的两大核心特性:无副作用和保证终止。这意味着 CEL 表达式执行时不会改变外部状态,只计算结果,而且永远不会陷入死循环,执行时间完全可控。这种设计让它成为 Kubernetes、Envoy 等高性能应用程序中的安全评估引擎。
cel-expr-python 是谷歌官方发布的 Python 实现,它不是从头开发的纯 Python 库,而是封装了官方的 C++ 核心。这样做有两个显著优势:一是保证了与 CEL 原生语义的高度一致,二是未来 C++ 核心的任何功能更新和性能优化都会自动同步到 Python 版本。
安装与基础使用
安装 cel-expr-python 非常简单,使用 pip 即可:
pip install cel-expr-python
安装完成后,让我们从一个最基础的例子开始。这个例子展示了 CEL 最核心的工作流程:创建环境、编译表达式、执行评估:
from cel_expr_python import cel# 第一步:创建环境并声明变量类型# 这一步告诉 CEL 我们会传入哪些变量及其类型cel_env = cel.NewEnv(variables={ "who": cel.Type.STRING, "age": cel.Type.INT})# 第二步:编译表达式# 编译后的表达式对象可以被重复使用,这是 CEL 性能优势的关键expr = cel_env.compile("'Hello, ' + who + '! You are ' + string(age) + ' years old.'")# 第三步:传入实际数据执行评估result = expr.eval(data={ "who": "World", "age": 25})print(result) # 输出: Hello, World! You are 25 years old.
注意编译步骤返回的表达式对象可以重复使用,只需在每次评估时传入不同的数据即可。这种“一次编译,多次评估”的模式特别适合需要高频评估的场景。
CEL 的类型系统
CEL 拥有丰富的类型系统,支持大多数常见的数据类型。在创建环境时,你需要声明变量类型,这样编译器才能进行类型检查:
from cel_expr_python import cel# 声明各种类型的变量cel_env = cel.NewEnv(variables={ "name": cel.Type.STRING, "count": cel.Type.INT, "price": cel.Type.DOUBLE, "is_active": cel.Type.BOOL, "tags": cel.Type.list(cel.Type.STRING), # 字符串列表 "metadata": cel.Type.map(cel.Type.STRING, cel.Type.DYN) # 动态类型的映射})
cel.Type.DYN 表示动态类型,当你无法确定值的具体类型时可以使用它。cel.Type.list() 和 cel.Type.map() 用于声明容器类型。
实战场景一:接口响应数据的复杂校验
在接口自动化测试中,我们经常需要验证复杂的 JSON 响应。传统方式需要编写大量嵌套的 if 语句,不仅代码冗长,而且难以维护。使用 CEL,我们可以将复杂的校验逻辑压缩为清晰简洁的表达式:
from cel_expr_python import cel# 定义校验环境validation_env = cel.NewEnv(variables={ "response": cel.Type.map(cel.Type.STRING, cel.Type.DYN)})# 编译校验表达式validation_expr = validation_env.compile(""" has(response.data) && has(response.data.users) && size(response.data.users) > 0 && response.data.users.all(u, has(u.id) && has(u.status) && u.status == 'active' && (!has(u.profile) || !has(u.profile.age) || (u.profile.age >= 18 && u.profile.age <= 100)) )""")# 模拟接口响应response_data = { "data": { "users": [ {"id": 1, "status": "active", "profile": {"age": 25}}, {"id": 2, "status": "active", "profile": {"age": 30}}, ] }}# 执行校验is_valid = validation_expr.eval(data={"response": response_data})print(f"Response is valid: {is_valid}") # 输出: Response is valid: True
这段代码中的 all() 函数是 CEL 提供的聚合函数,用于检查列表中的所有元素是否都满足条件。类似的函数还有 exists()、map() 和 filter()。
实战场景二:权限策略的动态测试
在测试权限系统时,我们经常需要验证用户是否满足各种复杂的访问条件。CEL 的表达式语法非常适合描述这类业务规则:
from cel_expr_python import cel# 定义权限检查环境permission_env = cel.NewEnv(variables={ "user": cel.Type.map(cel.Type.STRING, cel.Type.DYN), "coupon": cel.Type.map(cel.Type.STRING, cel.Type.DYN), "order": cel.Type.map(cel.Type.STRING, cel.Type.DYN)})# 编译优惠券使用规则can_use_coupon_expr = permission_env.compile(""" coupon.status == 'active' && coupon.valid_from <= now && coupon.valid_to >= now && coupon.min_order_amount <= order.amount && (size(coupon.allowed_user_ids) == 0 || user.id in coupon.allowed_user_ids) && (size(coupon.excluded_user_ids) == 0 || user.id not in coupon.excluded_user_ids) && user.vip_level >= coupon.required_vip_level""")# 测试用例test_cases = [ { "name": "普通用户使用有效优惠券", "user": {"id": 1, "vip_level": 1}, "coupon": { "status": "active", "valid_from": "2024-01-01", "valid_to": "2024-12-31", "min_order_amount": 100, "allowed_user_ids": [], "excluded_user_ids": [], "required_vip_level": 1 }, "order": {"amount": 150}, "expected": True }, { "name": "VIP不足的用户", "user": {"id": 2, "vip_level": 0}, "coupon": { "status": "active", "valid_from": "2024-01-01", "valid_to": "2024-12-31", "min_order_amount": 100, "allowed_user_ids": [], "excluded_user_ids": [], "required_vip_level": 1 }, "order": {"amount": 150}, "expected": False }]# 执行测试for test_case in test_cases: result = can_use_coupon_expr.eval(data=test_case) status = "PASS" if result == test_case["expected"] else "FAIL" print(f"{status}: {test_case['name']} -> {result}")
注意表达式中的 now 是 CEL 内置的当前时间变量,in 是成员检查操作符。这些内置特性让权限规则的编写变得非常自然。
实战场景三:跨语言项目的统一测试策略
在微服务架构中,不同服务可能使用不同编程语言。CEL 的一大优势是:你可以用 Go 或 Java 的工具链编译表达式,然后在 Python 中执行评估。这确保了不同服务使用完全一致的业务规则:
from cel_expr_python import cel# 假设这是从 Go 服务生成的序列化表达式(protobuf 格式)# 在实际场景中,你会从文件中读取这些二进制数据compiled_expr_proto = b"..." # 序列化的编译表达式# 反序列化并创建表达式对象expr = cel.deserialize_expr(compiled_expr_proto)# 在 Python 测试中复用完全相同的业务规则test_data = { "user_level": 3, "order_amount": 200, "coupon_type": "discount"}result = expr.eval(data=test_data)print(f"Rule evaluation result: {result}")
这种方式特别适合测试场景:你可以让后端开发人员用 Go 定义规则,测试人员用 Python 编写测试用例,而规则本身是完全一致的。
实战场景四:动态配置的自动化验证
在配置驱动的系统中,验证动态配置的有效性是一个常见需求。CEL 可以让验证逻辑与配置分离,变得易于维护:
from cel_expr_python import celimport yaml# 模拟功能开关配置feature_configs = { "new_checkout_flow": { "enabled": True, "percentage": 50, "allowed_envs": ["staging", "production"], "excluded_users": [1001, 1002], "start_time": "2024-01-01T00:00:00Z", "end_time": "2024-12-31T23:59:59Z" }, "experimental_ai": { "enabled": True, "percentage": 10, "allowed_envs": ["staging"], "excluded_users": [], "start_time": "2024-06-01T00:00:00Z" }}# 定义配置验证规则config_env = cel.NewEnv(variables={ "config": cel.Type.map(cel.Type.STRING, cel.Type.DYN), "user": cel.Type.map(cel.Type.STRING, cel.Type.DYN)})validation_expr = config_env.compile(""" config.enabled == true && config.percentage >= 0 && config.percentage <= 100 && (size(config.allowed_envs) == 0 || user.env in config.allowed_envs) && (size(config.excluded_users) == 0 || user.id not in config.excluded_users) && (!has(config.start_time) || now >= timestamp(config.start_time)) && (!has(config.end_time) || now <= timestamp(config.end_time))""")def validate_feature_config(feature_name, config, user): try: return validation_expr.eval(data={ "config": config, "user": user }) except Exception as e: print(f"Validation error for {feature_name}: {e}") return False# 测试不同用户场景test_users = [ {"id": 1, "env": "staging"}, {"id": 1001, "env": "staging"}, # 被排除的用户 {"id": 2, "env": "production"}, {"id": 3, "env": "development"} # 不允许的环境]for feature_name, config in feature_configs.items(): print(f"\nValidating {feature_name}:") for user in test_users: is_valid = validate_feature_config(feature_name, config, user) print(f" User {user['id']} in {user['env']}: {is_valid}")
这个例子展示了 CEL 如何处理时间戳、条件检查和集合操作,这些都是配置验证中常见的需求。
实战场景五:智能测试数据生成
在数据驱动测试中,生成符合特定规则的测试数据是一个挑战。CEL 可以帮助你定义数据约束,并验证生成的数据是否满足要求:
from cel_expr_python import celimport randomimport string# 定义数据生成规则generation_env = cel.NewEnv(variables={ "constraints": cel.Type.map(cel.Type.STRING, cel.Type.DYN)})# 编译数据验证规则data_validator = generation_env.compile(""" data.id > 0 && data.name.size() >= constraints.min_name_length && data.name.size() <= constraints.max_name_length && data.age >= constraints.min_age && data.age <= constraints.max_age && data.score >= 0 && data.score <= 100 && data.email.contains('@')""")def generate_random_name(min_len, max_len): length = random.randint(min_len, max_len) return ''.join(random.choices(string.ascii_letters, k=length))def generate_random_email(name): domains = ["example.com", "test.com", "demo.com"] return f"{name.lower()}@{random.choice(domains)}"def generate_valid_test_data(constraints): max_attempts = 100 for _ in range(max_attempts): # 生成随机数据 name = generate_random_name(constraints["min_name_length"], constraints["max_name_length"]) test_data = { "id": random.randint(1, 10000), "name": name, "age": random.randint(constraints["min_age"], constraints["max_age"]), "score": random.uniform(0, 100), "email": generate_random_email(name) } # 验证数据 if data_validator.eval(data={"data": test_data, "constraints": constraints}): return test_data raise Exception("Failed to generate valid test data")# 定义约束条件constraints = { "min_name_length": 3, "max_name_length": 10, "min_age": 18, "max_age": 65}# 生成多个有效的测试数据for i in range(5): test_data = generate_valid_test_data(constraints) print(f"Generated test data {i+1}: {test_data}")
CEL 表达式语法要点
CEL 的语法类似于 C 语言,但更简洁。以下是一些常用的语法元素:
条件判断:
# 使用三元运算符status = user.age >= 18 ? "adult" : "minor"
集合操作:
# 检查列表中是否有符合条件的元素has_vip = users.exists(u, u.vip_level >= 3)# 过滤列表active_users = users.filter(u, u.status == "active")# 映射列表user_names = users.map(u, u.name)
字符串操作:
full_name = first_name + " " + last_nameis_valid_email = email.contains("@") and email.endsWith(".com")
时间操作:
is_expired = expiration_time < nowis_within_range = start_time <= timestamp(request.time) <= end_time
性能特点
CEL 的评估速度非常快,官方宣称可以达到“纳秒到微秒”级别。这得益于两个因素:一是编译时优化,表达式在编译时就已经被转换为高效的内部表示;二是 C++ 核心的性能优势,Python 版本封装了这个高性能核心。
编译表达式并重复使用的模式是发挥 CEL 性能优势的关键:
# 正确做法:编译一次,重复使用expr = env.compile("user.age >= 18")for user in users: result = expr.eval(data={"user": user}) # 快速评估# 错误做法:每次都重新编译for user in users: expr = env.compile("user.age >= 18") # 浪费性能 result = expr.eval(data={"user": user})
总结
cel-expr-python 为 Python 开发者带来了谷歌在策略引擎领域的多年积累。它的核心价值在于将复杂的业务规则从代码中分离出来,以声明式的方式表达,同时保证了执行的安全性和性能。无论你是在做自动化测试、权限管理、配置验证,还是任何需要动态规则评估的场景,CEL 都值得你花时间了解。