From 58aa88f715535b28211cee292c4a76d2c7d3fa17 Mon Sep 17 00:00:00 2001 From: John Peters Date: Fri, 8 Dec 2023 00:20:28 -0600 Subject: [PATCH 1/6] "Added as_json flag" --- memphis/functions.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/memphis/functions.py b/memphis/functions.py index 8937f1c..156b021 100644 --- a/memphis/functions.py +++ b/memphis/functions.py @@ -5,7 +5,8 @@ def create_function( event, event_handler: callable, - use_async: bool = False + use_async: bool = False, + as_json: bool = False ) -> None: """ This function creates a Memphis function and processes events with the passed-in event_handler function. @@ -83,6 +84,10 @@ async def handler(event): for message in event["messages"]: try: payload = base64.b64decode(bytes(message['payload'], encoding='utf-8')) + if as_json: + payload = str(payload, 'utf-8') + payload = json.loads(payload) + if use_async: processed_message, processed_headers = await event_handler(payload, message['headers'], event["inputs"]) else: From dfa95d7135bf869c91c121308ac6b6b07cb8115c Mon Sep 17 00:00:00 2001 From: John Peters Date: Sat, 9 Dec 2023 00:31:54 -0600 Subject: [PATCH 2/6] "When as_json is given, allows the return of a javascript object instead of a uint8 array" --- memphis/functions.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/memphis/functions.py b/memphis/functions.py index 156b021..41ff188 100644 --- a/memphis/functions.py +++ b/memphis/functions.py @@ -93,6 +93,9 @@ async def handler(event): else: processed_message, processed_headers = event_handler(payload, message['headers'], event["inputs"]) + if as_json: + processed_message = bytes(json.dumps(processed_message), encoding='utf-8') + if isinstance(processed_message, bytes) and isinstance(processed_headers, dict): processed_events["messages"].append({ "headers": processed_headers, From fee9a9f235b706d268fe6b760ac1714c31723ffd Mon Sep 17 00:00:00 2001 From: John Peters Date: Mon, 11 Dec 2023 02:10:17 -0600 Subject: [PATCH 3/6] "Updated Readme, changed as_json to as_dict" --- README.md | 17 +++++++++++++++++ memphis/functions.py | 6 +++--- 2 files changed, 20 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 125f5d4..aac146c 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,23 @@ def event_handler(msg_payload, msg_headers, inputs): return bytes(json.dumps(as_json), encoding='utf-8'), msg_headers ``` +Instead of taking `msg_payload` as a bytes object, the as_dict flag can be used to have the JSON parsed to a dictionary for the user. + +```python +import json +import base64 +from memphis import create_function + +def handler(event, context): # The name of this file and this function should match the handler field in the memphis.yaml file in the following format . + return create_function(event, event_handler = event_handler) + +def event_handler(as_json, msg_headers, inputs): + + as_json['modified'] = True + + return as_json, msg_headers +``` + Memphis Functions support using Async functions through asyncio. When functions are async, set the use_async parameter to true. ```python import json diff --git a/memphis/functions.py b/memphis/functions.py index 41ff188..fd7613d 100644 --- a/memphis/functions.py +++ b/memphis/functions.py @@ -6,7 +6,7 @@ def create_function( event, event_handler: callable, use_async: bool = False, - as_json: bool = False + as_dict: bool = False ) -> None: """ This function creates a Memphis function and processes events with the passed-in event_handler function. @@ -84,7 +84,7 @@ async def handler(event): for message in event["messages"]: try: payload = base64.b64decode(bytes(message['payload'], encoding='utf-8')) - if as_json: + if as_dict: payload = str(payload, 'utf-8') payload = json.loads(payload) @@ -93,7 +93,7 @@ async def handler(event): else: processed_message, processed_headers = event_handler(payload, message['headers'], event["inputs"]) - if as_json: + if as_dict: processed_message = bytes(json.dumps(processed_message), encoding='utf-8') if isinstance(processed_message, bytes) and isinstance(processed_headers, dict): From e69bb54b27135e2095a7878cf9787b277f96a41b Mon Sep 17 00:00:00 2001 From: John Peters Date: Sat, 16 Dec 2023 01:30:18 -0600 Subject: [PATCH 4/6] "Added new branch common_errors" --- memphis/functions.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/memphis/functions.py b/memphis/functions.py index fd7613d..cdcf3b1 100644 --- a/memphis/functions.py +++ b/memphis/functions.py @@ -103,9 +103,6 @@ async def handler(event): }) elif processed_message is None and processed_headers is None: # filter out empty messages continue - elif processed_message is None or processed_headers is None: - err_msg = f"processed_messages is of type {type(processed_message)} and processed_headers is {type(processed_headers)}. Either both of these should be None or neither" - raise Exception(err_msg) else: err_msg = "The returned processed_message or processed_headers were not in the right format. processed_message must be bytes and processed_headers, dict" raise Exception(err_msg) From 097de413d7c54593bbeb38926823e0f7e461033c Mon Sep 17 00:00:00 2001 From: John Peters Date: Sat, 16 Dec 2023 01:36:59 -0600 Subject: [PATCH 5/6] "using dataclass to store errors..." --- memphis/errors.py | 6 ++++++ memphis/functions.py | 6 +++--- 2 files changed, 9 insertions(+), 3 deletions(-) create mode 100644 memphis/errors.py diff --git a/memphis/errors.py b/memphis/errors.py new file mode 100644 index 0000000..f98584a --- /dev/null +++ b/memphis/errors.py @@ -0,0 +1,6 @@ +from dataclasses import dataclass + +@dataclass +class Errors: + invalid_types = "The returned processed_message or processed_headers were not in the right format. processed_message must be bytes and processed_headers, dict" + conversion_error = "Returned message types from user function are not able to be converted into JSON:" \ No newline at end of file diff --git a/memphis/functions.py b/memphis/functions.py index ae10373..6545c0c 100644 --- a/memphis/functions.py +++ b/memphis/functions.py @@ -1,6 +1,7 @@ import json import base64 import asyncio +from errors import Errors def create_function( event, @@ -106,8 +107,7 @@ async def handler(event): elif processed_message is None and processed_headers is None: # filter out empty messages continue else: - err_msg = "The returned processed_message or processed_headers were not in the right format. processed_message must be bytes and processed_headers, dict" - raise Exception(err_msg) + raise Exception(Errors.invalid_types) except Exception as e: processed_events["failed_messages"].append({ "headers": message["headers"], @@ -118,6 +118,6 @@ async def handler(event): try: return json.dumps(processed_events, cls=EncodeBase64).encode('utf-8') except Exception as e: - return f"Returned message types from user function are not able to be converted into JSON: {e}" + return f"{Errors.conversion_error} {e}" return asyncio.run(handler(event)) From ee917c25f6a2f0988c99641ea21e5365c898c931 Mon Sep 17 00:00:00 2001 From: John Peters Date: Sat, 16 Dec 2023 01:49:10 -0600 Subject: [PATCH 6/6] "added more info after simplifying the errors into one" --- memphis/functions.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/memphis/functions.py b/memphis/functions.py index 6545c0c..1149a6b 100644 --- a/memphis/functions.py +++ b/memphis/functions.py @@ -107,6 +107,10 @@ async def handler(event): elif processed_message is None and processed_headers is None: # filter out empty messages continue else: + err_msg = f"""Either processed_message or processed_headers were of the wrong type. +processed_message should be of type bytes and processed_headers should be of type Dict. Ensure these types are correct. +processed_message is of type {type(processed_message)} and processed_headers if of type {type(processed_headers)}. +""" raise Exception(Errors.invalid_types) except Exception as e: processed_events["failed_messages"].append({ @@ -118,6 +122,6 @@ async def handler(event): try: return json.dumps(processed_events, cls=EncodeBase64).encode('utf-8') except Exception as e: - return f"{Errors.conversion_error} {e}" + return f"Returned message types from user function are not able to be converted into JSON: {str(e)}" return asyncio.run(handler(event))