marier-nico/event-processor

Pythonic event-processing library based on decorators

event-processor's Issues

Support parsing JSON within events

The feature might be used like this :

input = {
	"Records": [
        {
            "Sns": {
                "Message": json.dumps({"key": "value"})
            }
        }
    ]   
}

@event_processor.processor({"Records.0.Sns.Message.key": "value"}, autoparse=True)
def my_processor(_event):
    pass

The autoparse argument could make the processors try to parse string to a dict if there's more to go. For example, there's a key after Message, but Message is a string, so we try to json.loads the value and keep going.

The default should be False.

For JSON parsing, consider using https://pypi.org/project/jsoncomment/ for more flexible parsing (perhaps with a flexible_json keyword argument in the decorator. Though, this feature should be optional and opt-in (use extras_require). Thanks to @mobouchardcoveo for this suggestion!

Support flitering in arrays

Using this feature might look a little bit like this :

input = {
	"Records": [
        {
            "Sns": {
                "Message": "msg"
            }
        }
    ]   
}

@event_processor.processor({"Records.0.Sns.Message": "msg"})
def my_processor(_event):
    pass

If event["Records"] is indeed an array, but the index is invalid for that array, simply do not call the processor. No error should be raised for a bad index.

If event["Records"] is a dict that contains the key 0, then treat the filter as any other filter which does not deal with arrays.

Add a hook system

It should be possible to hook into the library to run some code before or after processors. One nice way to do it would be similar to python decorators, by essentially wrapping the processor in a new function.

Sample Usage

def my_hook(processor: Processor):
    start = time.time_ns()
    processor()  # The hook needs to run the processor
    end = time.time_ns()

    print(f"The processor {processor.name} took {end - start}ns to run")

Use-Cases

• Collect metrics
• Monitoring for failures
• Standardized logging
• Observability (opentelemetry)
• Escape hatch to support extending event-processor in otherwise unsupported ways
• Dynamically decide to abort calling a processor
• etc.

Caveats

• How are processor dependencies resolved with this system? Should we use functools.partial to make sure the hooks don't need to deal with them?
• Does this really fit into the current project architecture? Would re-architecting the project be beneficial to other features as well?

Add py.typed