Skip to content

Instantly share code, notes, and snippets.

@sduff
Last active March 30, 2022 23:24
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save sduff/dbe1960ae3ca285ecc95742f9d27407c to your computer and use it in GitHub Desktop.
Save sduff/dbe1960ae3ca285ecc95742f9d27407c to your computer and use it in GitHub Desktop.
Simple rules engine in python using eval and exec
#!/usr/local/bin/python3
rules = [
[
# A required field and default value
'not "replication_factor" in configuration',
'configuration["replication_factor"] = 1'
],
[
# production environments require at least 6 partitions
'metadata["environment"] == "production" and configuration["num_partitions"] <= 6',
'configuration["num_partitions"] = 6'
],
[
# non-prod environments require at least 3 partitions
'metadata["environment"] != "production" and configuration["num_partitions"] <= 3',
'configuration["num_partitions"] = 3 '
],
[
# topic names must be at least 5 characters in length
'len(configuration["topic"]) < 5',
'raise Exception("Topic name is too short")'
],
[
# topic names must start with the team
'not configuration["topic"].startswith(metadata["team"])',
'configuration["topic"] = "%s.%s"%(metadata["team"],configuration["topic"])'
],
[
# WARNING: This approach is a security risk! Something like this can delete files!
'True',
'open("rules.txt","w")'
]
]
def apply_rules(rules, config):
for rule, action in rules:
e = eval( rule, {}, { 'metadata':config["metadata"], 'configuration': config["configuration"] })
if e:
exec( action, {}, { 'metadata':config["metadata"], 'configuration': config["configuration"] })
# Some examples
config1 = {
"action": "CREATE",
"configuration":
{
"topic": "event_streaming_101",
"num_partitions": 3,
"replication_factor": 5
},
"metadata":
{
"environment": "production",
"domain": "highered",
"team": "avengers",
"project": "2022-midyear"
}
}
print("\n#1 user supplied configuration")
print(config1)
try:
apply_rules(rules, config1)
print("#1 implemented configuration")
print(config1)
except Exception as e:
print("#1 not implemented")
print(e)
config2 = {
"action": "CREATE",
"configuration":
{
"topic": "event_streaming_101",
"num_partitions": 3
},
"metadata":
{
"environment": "development",
"domain": "highered",
"team": "avengers",
"project": "2022-midyear"
}
}
print("\n#2 user supplied configuration")
print(config2)
try:
apply_rules(rules, config2)
print("#2 implemented configuration")
print(config2)
except Exception as e:
print("#2 not implemented")
print(e)
config3 = {
"action": "CREATE",
"configuration":
{
"topic": "this",
"num_partitions": 6
},
"metadata":
{
"environment": "production",
"domain": "highered",
"team": "avengers",
"project": "2022-midyear"
}
}
print("\n#3 user supplied configuration")
print(config3)
try:
apply_rules(rules, config3)
print("#3 implemented configuration")
print(config3)
except Exception as e:
print("#3 not implemented")
print(e)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment