JSON based events router, inspired from an AWS Lambda project.
Project description
JSON Events Router 
Uses simple yaml
based rules to take action on JSON
events. Uses jsonpath to scan the event message and regex
for includes
and excludes
conditionals.
Note!
jsonrouter
currently converts all matching field values tostr
ings forre
gex comparision so be aware of this when expectingint
to be returned.
Install
pip install jsonrouter
TL;DR
import json import yaml from jsonrouter import JsonMatchEngine # load rules from file or string configs = yaml.load(''' rules: - name: example-rule routers: - name: print-router vars: - name: the-name jsonpath: $.name ''') # an example json record json_string = ''' { "name": "jsonrouter", "type": "jsonpath matcher and router", "why": { "because": "it's easy" } } ''' def print_router(data): # a trivial example router function print(json.dumps(data, indent=4)) # explicitly declare your registered routers for security registered_routers = { 'print-router': print_router } eng = JsonMatchEngine(configs, registered_routers)
Use the engine to find matches:
In [2]: matches = eng.route_matches(json_string)
{
"name": "example-rule",
"routers": [
{
"name": "print-router"
}
],
"vars": {
"the-name": "jsonrouter"
},
"template": "",
"record": {
"name": "jsonrouter",
"type": "jsonpath matcher and router",
"why": {
"because": "it's easy"
}
}
}
The contents of matches
:
In [3]: matches
Out[3]:
[{'name': 'example-rule',
'routers': [{'name': 'print-router'}],
'vars': {'the-name': 'jsonrouter'},
'template': '',
'record': {'name': 'jsonrouter',
'type': 'jsonpath matcher and router',
'why': {'because': "it's easy"}}}]
Rule Config
Anatomy of the config.
# `rules` is the root of the config # note: rules can have same name! rules: # required - name: notification # required routers: # required - name: slack # required # all fields besides name are optional but may be required in the router channel: my-channel # optional vars: # required - name: type # required jsonpath: $..Type # required, except for constants: see bellow includes: ['.*'] # optional, default `['.*']` includes all excludes: [] # optional, default `[]` excludes nothing # `template` is optional template: | This {type} just came in
Constants
You can define a constant var by providing value
field only
vars: - name: my-constant value: my constant value
Basic Usage
Simple capture examples.
rules: - name: notification routers: - name: slack channel: my-channel vars: - name: type jsonpath: $..Type # include anything includes: ['.*'] # exclude empty excludes: ['^$'] template: | This {type} just came in
Advanced Regex
Match Part of String ()
rules: - name: console_login routers: - name: slack channel: my-channel vars: - name: detail-type jsonpath: $..detail-type includes: ['AWS Console Sign In via CloudTrail'] excludes: ['^$'] - name: user jsonpath: $..principalId # Match part of string includes: ['.*:(.*)'] excludes: ['^$'] - name: account jsonpath: $..account includes: ['.*'] excludes: ['^$'] template: | Yo! {user} just signed in to {account}.
Match in String with Group Name
Use (?P<variable_name>)
to capture patterns within the matched field.
This will override any naming collisions with
vars:name
you set in the yaml. It merges the rule vars with matched name(s) declared in the regex where named regex take precedence
rules: - name: console_login routers: - name: slack channel: my-channel vars: - name: detail-type jsonpath: $..detail-type includes: ['AWS Console Sign In via CloudTrail'] excludes: ['^$'] - name: user jsonpath: $..principalId # Match part of string with variable names includes: ['(?P<stuff>.*):(?P<user>.*)'] excludes: ['^$'] - name: account jsonpath: $..account includes: ['.*'] excludes: ['^$'] template: | Yo! {user} just signed in to {account}. This {stuff} was before the user.
Lambda Example
Since jsonrouter
started from an AWS Lambda SNS parsing project it has a jsonify_string
method that converts the SNS message field from a sting to a nested dict
.
import json import yaml from jsonrouter import JsonMatchEngine, jsonify_string def slack(webhook): # whatever logic you want in here pass with open('rules.yaml', 'r') as f: configs = yaml.safe_load(f) registered_routers = { 'slack': slack } eng = JsonMatchEngine(configs, registered_routers) def handler(event, context): # Main lambda handler function eng.route_matches(jsonify_string(event)['Records'])
Development
# local tests run from root of project
python3 -m pytest -v
# to see print output use `-s`
python3 -m pytest -v -s
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Filename, size | File type | Python version | Upload date | Hashes |
---|---|---|---|---|
Filename, size jsonrouter-0.4.9-py3-none-any.whl (8.0 kB) | File type Wheel | Python version py3 | Upload date | Hashes View |
Filename, size jsonrouter-0.4.9.tar.gz (7.8 kB) | File type Source | Python version None | Upload date | Hashes View |
Hashes for jsonrouter-0.4.9-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | a9000e8181ebfdd1c4b44a902a2a6abb38b20ec8aefd7e7aad9c23a45148f6a6 |
|
MD5 | bb582879fd59f71f2257ae64e537d142 |
|
BLAKE2-256 | 33804536b5e3b4939f7850298035032799b37571a10d443ea48abf3231f3feb4 |