Repository: hopeit-git/hopeit.engine (language: python)

Issue:
feature(plugins):Support using storage plugins to dump data from streams to persistent storage
MOTIVATION:
==========
In many applications, when data is writen to stream, it is actually persisted most of the times., or many events need to persiste they returned information what creates a lot of bolierplate.

IDEA:
====
Add the ability to call storage plugins directly from write_stream configuration section of events

TBC



Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index b4ce6a1d..117f2596 100644
--- a/README.md
+++ b/README.md
@@ -119,7 +119,7 @@ Microservices architecture is proven to be an efficient way that allows systems
 
 #### Consejo Nacional de Bomberos Voluntarios de la República Argentina
 
-National Council of Volunteer Firefighters adopted hopeit.engine to create new applications, including APIs for data integration, nation-wide real-time resources monitoring, Firefighters Academy, COVID-19 dashboards, to mention a few examples. hopeit.engine leverages for this organization the ability to ease development of new features at fast speed and allows them to integrate onto existing software and decouple compoenents using Microservices architecture. 
+National Council of Volunteer Firefighters adopted hopeit.engine to create new applications, including APIs for data integration, nation-wide real-time resources monitoring, Firefighters Academy, COVID-19 dashboards, to mention a few examples. hopeit.engine leverages for this organization the ability to ease development of new features at fast speed and allows them to integrate onto existing software and decouple components using Microservices architecture. 
 
 <img align="left" width="56" height="56" src="docs/source/readme/capsfe.png">
 
@@ -208,6 +208,10 @@ The framework is stable and running in production by early adopters with no majo
 It's also being tested and used in new projects. Now we are focusing in bug-fixing,
 complete requested new features and optimize small pieces of the software.
 
+- **APRIL 2022**: `hopeit.engine` `0.15.0` adds support to persist data directly from streams with zero-code approach using 
+`hopeit.fs_storage` plugin. Initially it supports saving data in batches to the file system using folders as partitions.
+This can be easily extended to persist in databases or cloud storage services like S3 or GCS.
+
 - **JANUARY 2022**: `hopeit.engine` `0.13.0` ensures support for Python 3.10 and updates `aiohttp` to latest version. 
 `0.14.0` adds support to return `web.StreamResponse` in chunks. 
 
diff --git a/apps/benchmark/simple-benchmark/api/openapi.json b/apps/benchmark/simple-benchmark/api/openapi.json
index 5d5772dd..65f49330 100644
--- a/apps/benchmark/simple-benchmark/api/openapi.json
+++ b/apps/benchmark/simple-benchmark/api/openapi.json
@@ -6,7 +6,7 @@
     "description": "Simple Benchmark App"
   },
   "paths": {
-    "/api/simple-benchmark/0x14/give-me-something": {
+    "/api/simple-benchmark/0x15/give-me-something": {
       "get": {
         "summary": "Simple Benchamrck: Give me Something",
         "description": "Loads Something from disk",
@@ -53,11 +53,11 @@
           }
         },
         "tags": [
-          "simple_benchmark.0x14"
+          "simple_benchmark.0x15"
         ]
       }
     },
-    "/api/simple-benchmark/0x14/query-something-fs": {
+    "/api/simple-benchmark/0x15/query-something-fs": {
       "get": {
         "summary": "Simple Benchamrck: Query Something",
         "description": "Loads Something from disk",
@@ -114,11 +114,11 @@
           }
         },
         "tags": [
-          "simple_benchmark.0x14"
+          "simple_benchmark.0x15"
         ]
       }
     },
-    "/api/simple-benchmark/0x14/save-something-fs": {
+    "/api/simple-benchmark/0x15/save-something-fs": {
       "post": {
         "summary": "Simple Benchmark: Save Something",
         "description": "Creates and saves Something",
@@ -176,11 +176,11 @@
           }
         },
         "tags": [
-          "simple_benchmark.0x14"
+          "simple_benchmark.0x15"
         ]
       }
     },
-    "/api/simple-benchmark/0x14/query-something-redis": {
+    "/api/simple-benchmark/0x15/query-something-redis": {
       "get": {
         "summary": "Simple Benchamrck: Query Something",
         "description": "Loads Something from disk",
@@ -237,11 +237,11 @@
           }
         },
         "tags": [
-          "simple_benchmark.0x14"
+          "simple_benchmark.0x15"
         ]
       }
     },
-    "/api/simple-benchmark/0x14/save-something-redis": {
+    "/api/simple-benchmark/0x15/save-something-redis": {
       "post": {
         "summary": "Simple Benchmark: Save Something",
         "description": "Creates and saves Something",
@@ -299,7 +299,7 @@
           }
         },
         "tags": [
-          "simple_benchmark.0x14"
+          "simple_benchmark.0x15"
         ]
       }
     }
diff --git a/apps/benchmark/simple-benchmark/src/simple_benchmark/query_something_fs.py b/apps/benchmark/simple-benchmark/src/simple_benchmark/query_something_fs.py
index a709aa68..2d21a048 100644
--- a/apps/benchmark/simple-benchmark/src/simple_benchmark/query_something_fs.py
+++ b/apps/benchmark/simple-benchmark/src/simple_benchmark/query_something_fs.py
@@ -10,7 +10,7 @@
 from hopeit.app.api import event_api
 from hopeit.app.context import EventContext, PostprocessHook
 from hopeit.app.logger import app_extra_logger
-from hopeit.fs_storage import FileStorage
+from hopeit.fs_storage import FileStorage, FileStorageSettings
 from model import Something, StatusType, Status, SomethingNotFound
 
 __steps__ = ['load', 'update_status_history']
@@ -32,7 +32,10 @@
 async def __init_event__(context):
     global fs
     if fs is None:
-        fs = FileStorage(path=str(context.env['fs']['data_path']))
+        settings: FileStorageSettings = context.settings(
+            key="fs_storage", datatype=FileStorageSettings
+        )
+        fs = FileStorage.with_settings(settings)
 
 
 async def load(payload: None, context: EventContext, *,
diff --git a/apps/benchmark/simple-benchmark/src/simple_benchmark/save_something_fs.py b/apps/benchmark/simple-benchmark/src/simple_benchmark/save_something_fs.py
index 0976ad14..f17dd26e 100644
--- a/apps/benchmark/simple-benchmark/src/simple_benchmark/save_something_fs.py
+++ b/apps/benchmark/simple-benchmark/src/simple_benchmark/save_something_fs.py
@@ -8,7 +8,7 @@
 from hopeit.app.api import event_api
 from hopeit.app.logger import app_extra_logger
 from hopeit.app.context import EventContext
-from hopeit.fs_storage import FileStorage
+from hopeit.fs_storage import FileStorage, FileStorageSettings
 
 from model import Something, User, SomethingParams
 
@@ -32,7 +32,10 @@
 async def __init_event__(context):
     global fs
     if fs is None:
-        fs = FileStorage(path=str(context.env['fs']['data_path']))
+        settings: FileStorageSettings = context.settings(
+            key="fs_storage", datatype=FileStorageSettings
+        )
+        fs = FileStorage.with_settings(settings)
 
 
 async def create_something(payload: SomethingParams, context: EventContext) -> Something:
diff --git a/apps/examples/client-example/api/openapi.json b/apps/examples/client-example/api/openapi.json
index 79f07ef9..189262dd 100644
--- a/apps/examples/client-example/api/openapi.json
+++ b/apps/examples/client-example/api/openapi.json
@@ -1,12 +1,12 @@
 {
   "openapi": "3.0.3",
   "info": {
-    "version": "0.14",
+    "version": "0.15",
     "title": "Client Example",
     "description": "Client Example"
   },
   "paths": {
-    "/api/config-manager/0x14/runtime-apps-config": {
+    "/api/config-manager/0x15/runtime-apps-config": {
       "get": {
         "summary": "Config Manager: Runtime Apps Config",
         "description": "Returns the runtime config for the Apps running on this server",
@@ -62,11 +62,11 @@
           }
         },
         "tags": [
-          "config_manager.0x14"
+          "config_manager.0x15"
         ]
       }
     },
-    "/api/config-manager/0x14/cluster-apps-config": {
+    "/api/config-manager/0x15/cluster-apps-config": {
       "get": {
         "summary": "Config Manager: Cluster Apps Config",
         "description": "Handle remote access to runtime configuration for a group of hosts",
@@ -122,11 +122,11 @@
           }
         },
         "tags": [
-          "config_manager.0x14"
+          "config_manager.0x15"
         ]
       }
     },
-    "/api/client-example/0x14/count-and-save": {
+    "/api/client-example/0x15/count-and-save": {
       "get": {
         "summary": "Client Example: Count Objects and Save new one",
         "description": "Count all available Something objects connecting to simple-example app",
@@ -193,7 +193,7 @@
           }
         },
         "tags": [
-          "client_example.0x14"
+          "client_example.0x15"
         ],
         "security": [
           {
@@ -482,6 +482,9 @@
           "route": {
             "type": "string"
           },
+          "impl": {
+            "type": "string"
+          },
           "connections": {
             "type": "array",
             "items": {
@@ -516,10 +519,17 @@
               "type": "string"
             },
             "default": []
+          },
+          "dataobjects": {
+            "type": "array",
+            "items": {
+              "type": "string"
+            },
+            "default": []
           }
         },
         "x-module-name": "hopeit.app.config",
-        "description": "\n    Event descriptor\n    "
+        "description": "\n    Event Descriptor: condfigures event implementation\n\n    :field: type, EventType: type of event i.e.: GET, POST, MULTIPART, STREAM, SERVICE\n    :field: plug_mode, EventPlugMode: defines whether an event defined in a plugin is created in the\n        current app (ON_APP) or it will be created in the original plugin (STANDALONE, default)\n    :field: route, optional str: custom route for endpoint. If not specified route will be derived\n        from `/api/app_name/app_version/event_name`\n    :field: impl, optional str: custom event implementation Python module. If not specified, module\n        with same same as event will be imported.\n    :field: connections, list of EventConnection: specifies dependencies on other apps/endpoints,\n        that can be used by client plugins to call events on external apps\n    :field: read_stream, optional ReadStreamDescriptor: specifies source stream to read from.\n        Valid only for STREAM events.\n    :field: write_stream, optional WriteStreamDescriptor: for any type of events, resultant dataobjects will\n        be published to the specified stream.\n    :field: auth, list of AuthType: supported authentication schemas for this event. If not specified\n        application default will be used.\n    :field: setting_keys, list of str: by default EventContext will have access to the settings section\n        with the same name of the event using `settings = context.settings(datatype=MySettingsType)`.\n        In case additional sections are needed to be accessed from\n        EventContext, then a list of setting keys, including the name of the event if needed,\n        can be specified here. Then access to a `custom` key can be done using\n        `custom_settings = context.settings(key=\"customer\", datatype=MyCustomSettingsType)`\n    :field: dataobjects, list of str: list of full qualified dataobject types that this event can process.\n        When not specified, the engine will inspect the module implementation and find all datatypes supported\n        as payload in the functions defined as `__steps__`. In case of generic functions that support\n        `payload: DataObject` argument, then a list of full qualified datatypes must be specified here.\n    "
       },
       "EventConnection": {
         "type": "object",
@@ -646,7 +656,7 @@
           },
           "engine_version": {
             "type": "string",
-            "default": "0.14.2"
+            "default": "0.15.0"
           }
         },
         "x-module-name": "hopeit.server.config",
diff --git a/apps/examples/simple-example/api/openapi.json b/apps/examples/simple-example/api/openapi.json
index 11233391..395129f2 100644
--- a/apps/examples/simple-example/api/openapi.json
+++ b/apps/examples/simple-example/api/openapi.json
@@ -1,12 +1,12 @@
 {
   "openapi": "3.0.3",
   "info": {
-    "version": "0.14",
+    "version": "0.15",
     "title": "Simple Example",
     "description": "Simple Example"
   },
   "paths": {
-    "/api/basic-auth/0x14/decode": {
+    "/api/basic-auth/0x15/decode": {
       "get": {
         "summary": "Basic Auth: Decode",
         "description": "Returns decoded auth info",
@@ -44,7 +44,7 @@
           }
         },
         "tags": [
-          "basic_auth.0x14"
+          "basic_auth.0x15"
         ],
         "security": [
           {
@@ -53,7 +53,7 @@
         ]
       }
     },
-    "/api/config-manager/0x14/runtime-apps-config": {
+    "/api/config-manager/0x15/runtime-apps-config": {
       "get": {
         "summary": "Config Manager: Runtime Apps Config",
         "description": "Returns the runtime config for the Apps running on this server",
@@ -109,11 +109,11 @@
           }
         },
         "tags": [
-          "config_manager.0x14"
+          "config_manager.0x15"
         ]
       }
     },
-    "/api/config-manager/0x14/cluster-apps-config": {
+    "/api/config-manager/0x15/cluster-apps-config": {
       "get": {
         "summary": "Config Manager: Cluster Apps Config",
         "description": "Handle remote access to runtime configuration for a group of hosts",
@@ -169,11 +169,11 @@
           }
         },
         "tags": [
-          "config_manager.0x14"
+          "config_manager.0x15"
         ]
       }
     },
-    "/api/simple-example/0x14/list-somethings": {
+    "/api/simple-example/0x15/list-somethings": {
       "get": {
         "summary": "Simple Example: List Objects",
         "description": "Lists all available Something objects",
@@ -243,7 +243,7 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
@@ -252,7 +252,7 @@
         ]
       }
     },
-    "/api/simple-example/0x14/query-something": {
+    "/api/simple-example/0x15/query-something": {
       "get": {
         "summary": "Simple Example: Query Something",
         "description": "Loads Something from disk",
@@ -266,6 +266,15 @@
               "type": "string"
             }
           },
+          {
+            "name": "partition_key",
+            "in": "query",
+            "required": true,
+            "description": "Partition folder in `YYYY/MM/DD/HH` format",
+            "schema": {
+              "type": "string"
+            }
+          },
           {
             "name": "X-Track-Request-Id",
             "in": "header",
@@ -329,7 +338,7 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
@@ -350,6 +359,15 @@
               "type": "string"
             }
           },
+          {
+            "name": "partition_key",
+            "in": "query",
+            "required": true,
+            "description": "Partition folder in `YYYY/MM/DD/HH` format",
+            "schema": {
+              "type": "string"
+            }
+          },
           {
             "name": "X-Track-Request-Id",
             "in": "header",
@@ -424,7 +442,7 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
@@ -433,7 +451,7 @@
         ]
       }
     },
-    "/api/simple-example/0x14/save-something": {
+    "/api/simple-example/0x15/save-something": {
       "post": {
         "summary": "Simple Example: Save Something",
         "description": "Creates and saves Something",
@@ -530,7 +548,7 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
@@ -539,7 +557,7 @@
         ]
       }
     },
-    "/api/simple-example/0x14/download-something": {
+    "/api/simple-example/0x15/download-something": {
       "get": {
         "summary": "Simple Example: Download Something",
         "description": "Download image file. The PostprocessHook return the requested file as stream.",
@@ -626,7 +644,7 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
@@ -635,7 +653,7 @@
         ]
       }
     },
-    "/api/simple-example/0x14/download-something-streamed": {
+    "/api/simple-example/0x15/download-something-streamed": {
       "get": {
         "summary": "Simple Example: Download Something Streamed",
         "description": "Download streamd created content as file.\nThe PostprocessHook return the requested resource as stream using `prepare_stream_response`.",
@@ -703,11 +721,11 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ]
       }
     },
-    "/api/simple-example/0x14/upload-something": {
+    "/api/simple-example/0x15/upload-something": {
       "post": {
         "summary": "Simple Example: Multipart Upload files",
         "description": "Upload files using Multipart form request",
@@ -845,7 +863,7 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
@@ -854,7 +872,7 @@
         ]
       }
     },
-    "/api/simple-example/0x14/streams/something-event": {
+    "/api/simple-example/0x15/streams/something-event": {
       "post": {
         "summary": "Simple Example: Something Event",
         "description": "Submits a Something object to a stream to be processed asynchronously by process-events app event",
@@ -923,7 +941,7 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
@@ -932,7 +950,7 @@
         ]
       }
     },
-    "/api/simple-example/0x14/collector/query-concurrently": {
+    "/api/simple-example/0x15/collector/query-concurrently": {
       "post": {
         "summary": "Simple Example: Query Concurrently",
         "description": "Loads 2 Something objects concurrently from disk and combine the results\nusing `collector` steps constructor (instantiating an `AsyncCollector`)",
@@ -1004,7 +1022,7 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
@@ -1013,7 +1031,7 @@
         ]
       }
     },
-    "/api/simple-example/0x14/collector/collect-spawn": {
+    "/api/simple-example/0x15/collector/collect-spawn": {
       "post": {
         "summary": "Simple Example: Collect and Spawn",
         "description": "Loads 2 Something objects concurrently from disk, combine the results\nusing `collector` steps constructor (instantiating an `AsyncCollector`)\nthen spawn the items found individually into a stream",
@@ -1091,7 +1109,7 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
@@ -1100,7 +1118,7 @@
         ]
       }
     },
-    "/api/simple-example/0x14/shuffle/spawn-event": {
+    "/api/simple-example/0x15/shuffle/spawn-event": {
       "post": {
         "summary": "Simple Example: Spawn Event",
         "description": "This example will spawn 3 data events, those are going to be send to a stream using SHUFFLE\nand processed in asynchronously / in parallel if multiple nodes are available",
@@ -1178,7 +1196,7 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
@@ -1187,7 +1205,7 @@
         ]
       }
     },
-    "/api/simple-example/0x14/shuffle/parallelize-event": {
+    "/api/simple-example/0x15/shuffle/parallelize-event": {
       "post": {
         "summary": "Simple Example: Parallelize Event",
         "description": "This example will spawn 2 copies of payload data, those are going to be send to a stream using SHUFFLE\nand processed in asynchronously / in parallel if multiple nodes are available,\nthen submitted to other stream to be updated and saved",
@@ -1265,7 +1283,7 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
@@ -1274,7 +1292,7 @@
         ]
       }
     },
-    "/api/simple-example/0x14/basic-auth/0x14/login": {
+    "/api/simple-example/0x15/basic-auth/0x15/login": {
       "get": {
         "summary": "Basic Auth: Login",
         "description": "Handles users login using basic-auth\nand generate access tokens for external services invoking apps\nplugged in with basic-auth plugin.",
@@ -1342,7 +1360,7 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
@@ -1351,7 +1369,7 @@
         ]
       }
     },
-    "/api/simple-example/0x14/basic-auth/0x14/refresh": {
+    "/api/simple-example/0x15/basic-auth/0x15/refresh": {
       "get": {
         "summary": "Basic Auth: Refresh",
         "description": "This event can be used for obtain new access token and update refresh token (http cookie),\nwith no need to re-login the user if there is a valid refresh token active.",
@@ -1419,16 +1437,16 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
-            "simple_example.0x14.refresh": []
+            "simple_example.0x15.refresh": []
           }
         ]
       }
     },
-    "/api/simple-example/0x14/basic-auth/0x14/logout": {
+    "/api/simple-example/0x15/basic-auth/0x15/logout": {
       "get": {
         "summary": "Basic Auth: Logout",
         "description": "Invalidates previous refresh cookies.",
@@ -1505,11 +1523,11 @@
           }
         },
         "tags": [
-          "simple_example.0x14"
+          "simple_example.0x15"
         ],
         "security": [
           {
-            "simple_example.0x14.refresh": []
+            "simple_example.0x15.refresh": []
           }
         ]
       }
@@ -1856,6 +1874,9 @@
           "route": {
             "type": "string"
           },
+          "impl": {
+            "type": "string"
+          },
           "connections": {
             "type": "array",
             "items": {
@@ -1890,10 +1911,17 @@
               "type": "string"
             },
             "default": []
+          },
+          "dataobjects": {
+            "type": "array",
+            "items": {
+              "type": "string"
+            },
+            "default": []
           }
         },
         "x-module-name": "hopeit.app.config",
-        "description": "\n    Event descriptor\n    "
+        "description": "\n    Event Descriptor: condfigures event implementation\n\n    :field: type, EventType: type of event i.e.: GET, POST, MULTIPART, STREAM, SERVICE\n    :field: plug_mode, EventPlugMode: defines whether an event defined in a plugin is created in the\n        current app (ON_APP) or it will be created in the original plugin (STANDALONE, default)\n    :field: route, optional str: custom route for endpoint. If not specified route will be derived\n        from `/api/app_name/app_version/event_name`\n    :field: impl, optional str: custom event implementation Python module. If not specified, module\n        with same same as event will be imported.\n    :field: connections, list of EventConnection: specifies dependencies on other apps/endpoints,\n        that can be used by client plugins to call events on external apps\n    :field: read_stream, optional ReadStreamDescriptor: specifies source stream to read from.\n        Valid only for STREAM events.\n    :field: write_stream, optional WriteStreamDescriptor: for any type of events, resultant dataobjects will\n        be published to the specified stream.\n    :field: auth, list of AuthType: supported authentication schemas for this event. If not specified\n        application default will be used.\n    :field: setting_keys, list of str: by default EventContext will have access to the settings section\n        with the same name of the event using `settings = context.settings(datatype=MySettingsType)`.\n        In case additional sections are needed to be accessed from\n        EventContext, then a list of setting keys, including the name of the event if needed,\n        can be specified here. Then access to a `custom` key can be done using\n        `custom_settings = context.settings(key=\"customer\", datatype=MyCustomSettingsType)`\n    :field: dataobjects, list of str: list of full qualified dataobject types that this event can process.\n        When not specified, the engine will inspect the module implementation and find all datatypes supported\n        as payload in the functions defined as `__steps__`. In case of generic functions that support\n        `payload: DataObject` argument, then a list of full qualified datatypes must be specified here.\n    "
       },
       "EventConnection": {
         "type": "object",
@@ -2020,7 +2048,7 @@
           },
           "engine_version": {
             "type": "string",
-            "default": "0.14.2"
+            "default": "0.15.0"
           }
         },
         "x-module-name": "hopeit.server.config",
@@ -2262,6 +2290,10 @@
           },
           "item2_id": {
             "type": "string"
+          },
+          "partition_key": {
+            "type": "string",
+            "default": ""
           }
         },
         "x-module-name": "model",
@@ -2277,10 +2309,10 @@
         "type": "http",
         "scheme": "bearer"
       },
-      "simple_example.0x14.refresh": {
+      "simple_example.0x15.refresh": {
         "type": "apiKey",
         "in": "cookie",
-        "name": "simple_example.0x14.refresh"
+        "name": "simple_example.0x15.refresh"
       }
     }
   },
diff --git a/apps/examples/simple-example/config/app-config.json b/apps/examples/simple-example/config/app-config.json
index 6f41767d..e9e50bc2 100644
--- a/apps/examples/simple-example/config/app-config.json
+++ b/apps/examples/simple-example/config/app-config.json
@@ -14,11 +14,11 @@
     "cors_origin": "*"
   },
   "env" : {
-    "fs": {
-      "data_path": "/tmp/{auto}/"
+    "storage": {
+      "base_path": "/tmp/hopeit/"
     },
     "upload_something": {
-      "save_path": "/tmp/{auto}/",
+      "save_path": "{env.storage.base_path}/{auto}/",
       "chunk_size": 16384
     }
   },
@@ -53,22 +53,36 @@
       "logging": {
         "extra_fields": ["something_id"]
       }
+    },
+    "fs_storage": {
+      "path": "{env.storage.base_path}/{auto}",
+      "partition_dateformat": "%Y/%m/%d/%H/"
+    },
+    "storage.save_events_fs": {
+      "path": "{env.storage.base_path}/{auto}",
+      "partition_dateformat": "%Y/%m/%d/%H/",
+      "flush_seconds": 60.0,
+      "flush_max_size": 100
     }
   },
   "events": {
     "list_somethings": {
-      "type": "GET"
+      "type": "GET",
+      "setting_keys": ["fs_storage"]
     },
     "query_something": {
       "type": "GET",
-      "route": "{app.name}/{app.version}/query_something"
+      "route": "{app.name}/{app.version}/query_something",
+      "setting_keys": ["fs_storage"]
     },
     "query_something_extended": {
       "type": "POST",
-      "route": "{app.name}/{app.version}/query_something"
+      "route": "{app.name}/{app.version}/query_something",
+      "setting_keys": ["fs_storage"]
     },
     "save_something" : {
-      "type": "POST"
+      "type": "POST",
+      "setting_keys": ["fs_storage"]
     },
     "download_something": {
       "type": "GET"
@@ -86,7 +100,7 @@
         "name": "{events.streams.something_event.write_stream.name}"
       }
     },
-    "streams.something_event" : {
+    "streams.something_event": {
       "type": "POST",
       "write_stream": {
         "name": "{auto}",
@@ -99,28 +113,45 @@
         "name": "{events.streams.something_event.write_stream.name}",
         "consumer_group": "{auto}",
         "queues": ["high-prio", "AUTO"]
-      }
+      },
+      "setting_keys": ["fs_storage"]
     },
     "collector.query_concurrently": {
-      "type": "POST"
+      "type": "POST",
+      "setting_keys": ["fs_storage"]
     },
-    "collector.collect_spawn" : {
+    "collector.collect_spawn": {
       "type": "POST",
       "write_stream": {
         "name": "{events.streams.something_event.write_stream.name}"
-      }
+      },
+      "setting_keys": ["fs_storage"]
     },
-    "shuffle.spawn_event" : {
+    "shuffle.spawn_event": {
       "type": "POST",
       "write_stream": {
         "name": "{events.streams.something_event.write_stream.name}"
-      }
+      },
+      "setting_keys": ["fs_storage"]
     },
-    "shuffle.parallelize_event" : {
+    "shuffle.parallelize_event": {
       "type": "POST",
       "write_stream": {
         "name": "{events.streams.something_event.write_stream.name}"
-      }
+      },
+      "setting_keys": ["fs_storage"]
+    },
+    "storage.save_events_fs": {
+      "type": "STREAM",
+      "read_stream": {
+        "name": "{events.streams.something_event.write_stream.name}",
+        "consumer_group": "{auto}",
+        "queues": ["high-prio", "AUTO"]
+      },
+      "impl": "hopeit.fs_storage.events.stream_batch_storage",
+      "dataobjects": [
+        "model.Something"
+      ]
     }
   }
 }
diff --git a/apps/examples/simple-example/src/model/__init__.py b/apps/examples/simple-example/src/model/__init__.py
index aa89fca6..a535cde9 100644
--- a/apps/examples/simple-example/src/model/__init__.py
+++ b/apps/examples/simple-example/src/model/__init__.py
@@ -66,6 +66,7 @@ class ItemsInfo:
     """
     item1_id: str
     item2_id: str
+    partition_key: str = ""
 
 
 @dataobject(event_id='payload.id', event_ts='payload.status.ts')
diff --git a/apps/examples/simple-example/src/simple_example/collector/collect_spawn.py b/apps/examples/simple-example/src/simple_example/collector/collect_spawn.py
index c160d378..d7e6f78d 100644
--- a/apps/examples/simple-example/src/simple_example/collector/collect_spawn.py
+++ b/apps/examples/simple-example/src/simple_example/collector/collect_spawn.py
@@ -16,7 +16,7 @@
 from hopeit.dataobjects import dataobject
 from hopeit.server.collector import Collector
 from hopeit.server.steps import SHUFFLE
-from hopeit.fs_storage import FileStorage
+from hopeit.fs_storage import FileStorage, FileStorageSettings
 from model import ItemsInfo, Something, SomethingNotFound
 
 
@@ -50,7 +50,10 @@ class ItemsCollected:
 async def __init_event__(context):
     global fs
     if fs is None:
-        fs = FileStorage(path=str(context.env['fs']['data_path']))
+        settings: FileStorageSettings = context.settings(
+            key="fs_storage", datatype=FileStorageSettings
+        )
+        fs = FileStorage.with_settings(settings)
 
 
 async def load_first(collector: Collector, context: EventContext) -> Union[Something, SomethingNotFound]:
@@ -67,7 +70,7 @@ async def load_first(collector: Collector, context: EventContext) -> Union[Somet
     item_id = items_to_read.item1_id
     await asyncio.sleep(0.1)
     logger.info(context, "load_second", extra=extra(something_id=item_id, path=fs.path))
-    something = await fs.get(key=item_id, datatype=Something)
+    something = await fs.get(key=item_id, datatype=Something, partition_key=items_to_read.partition_key)
     if something is None:
         logger.warning(context, "item not found", extra=extra(something_id=item_id, path=fs.path))
         return SomethingNotFound(str(fs.path), item_id)
@@ -88,7 +91,7 @@ async def load_second(collector: Collector, context: EventContext) -> Union[Some
     item_id = items_to_read.item2_id
     await asyncio.sleep(0.1)
     logger.info(context, "load_first", extra=extra(something_id=item_id, path=fs.path))
-    something = await fs.get(key=item_id, datatype=Something)
+    something = await fs.get(key=item_id, datatype=Something, partition_key=items_to_read.partition_key)
     if something is None:
         logger.warning(context, "item not found", extra=extra(something_id=item_id, path=fs.path))
         return SomethingNotFound(str(fs.path), item_id)
diff --git a/apps/examples/simple-example/src/simple_example/collector/query_concurrently.py b/apps/examples/simple-example/src/simple_example/collector/query_concurrently.py
index dfc28335..4e69f77e 100644
--- a/apps/examples/simple-example/src/simple_example/collector/query_concurrently.py
+++ b/apps/examples/simple-example/src/simple_example/collector/query_concurrently.py
@@ -12,7 +12,7 @@
 from hopeit.app.events import collector_step
 from hopeit.app.logger import app_extra_logger
 from hopeit.server.collector import Collector
-from hopeit.fs_storage import FileStorage
+from hopeit.fs_storage import FileStorage, FileStorageSettings
 from model import ItemsInfo, Something, SomethingNotFound
 
 
@@ -33,7 +33,10 @@
 async def __init_event__(context):
     global fs
     if fs is None:
-        fs = FileStorage(path=str(context.env['fs']['data_path']))
+        settings: FileStorageSettings = context.settings(
+            key="fs_storage", datatype=FileStorageSettings
+        )
+        fs = FileStorage.with_settings(settings)
 
 
 async def load_first(collector: Collector, context: EventContext) -> Union[Something, SomethingNotFound]:
@@ -50,7 +53,7 @@ async def load_first(collector: Collector, context: EventContext) -> Union[Somet
     item_id = items_to_read.item1_id
     await asyncio.sleep(0.1)
     logger.info(context, "load_second", extra=extra(something_id=item_id, path=fs.path))
-    something = await fs.get(key=item_id, datatype=Something)
+    something = await fs.get(key=item_id, datatype=Something, partition_key=items_to_read.partition_key)
     if something is None:
         logger.warning(context, "item not found", extra=extra(something_id=item_id, path=fs.path))
         return SomethingNotFound(str(fs.path), item_id)
@@ -71,7 +74,7 @@ async def load_second(collector: Collector, context: EventContext) -> Union[Some
     item_id = items_to_read.item2_id
     await asyncio.sleep(0.1)
     logger.info(context, "load_first", extra=extra(something_id=item_id, path=fs.path))
-    something = await fs.get(key=item_id, datatype=Something)
+    something = await fs.get(key=item_id, datatype=Something, partition_key=items_to_read.partition_key)
     if something is None:
         logger.warning(context, "item not found", extra=extra(something_id=item_id, path=fs.path))
         return SomethingNotFound(str(fs.path), item_id)
diff --git a/apps/examples/simple-example/src/simple_example/download_something.py b/apps/examples/simple-example/src/simple_example/download_something.py
index d9b03e8d..29199fb9 100644
--- a/apps/examples/simple-example/src/simple_example/download_something.py
+++ b/apps/examples/simple-example/src/simple_example/download_something.py
@@ -39,7 +39,7 @@ async def find_image(payload: None, context: EventContext, *, file_name: str) ->
     Find image file
     """
     src_file_path = f"./apps/examples/simple-example/resources/{file_name}"
-    tgt_file_path = os.path.join(str(context.env['fs']['data_path']), file_name)
+    tgt_file_path = os.path.join(str(context.env['storage']['base_path']), file_name)
     try:
         shutil.copy(src_file_path, tgt_file_path)
     except IOError as msg:
diff --git a/apps/examples/simple-example/src/simple_example/list_somethings.py b/apps/examples/simple-example/src/simple_example/list_somethings.py
index 6119af31..5ef8e39f 100644
--- a/apps/examples/simple-example/src/simple_example/list_somethings.py
+++ b/apps/examples/simple-example/src/simple_example/list_somethings.py
@@ -8,7 +8,7 @@
 from hopeit.app.api import event_api
 from hopeit.app.context import EventContext
 from hopeit.app.logger import app_extra_logger
-from hopeit.fs_storage import FileStorage
+from hopeit.fs_storage import FileStorage, FileStorageSettings
 from model import Something
 
 __steps__ = ['load_all']
@@ -30,15 +30,29 @@
 async def __init_event__(context):
     global fs
     if fs is None:
-        fs = FileStorage(path=str(context.env['fs']['data_path']))
+        settings: FileStorageSettings = context.settings(
+            key="fs_storage", datatype=FileStorageSettings
+        )
+        fs = FileStorage.with_settings(settings)
 
 
 async def load_all(payload: None, context: EventContext, wildcard: str = '*') -> List[Something]:
+    """
+    Load objects that match the given wildcard
+    """
     assert fs
     logger.info(context, "load_all", extra=extra(path=fs.path))
     items: List[Something] = []
-    for item_id in await fs.list_objects(wildcard):
-        something = await fs.get(key=item_id, datatype=Something)
+    for item_loc in await fs.list_objects(wildcard):
+        something = await fs.get(
+            key=item_loc.item_id, datatype=Something, partition_key=item_loc.partition_key
+        )
         if something is not None:
             items.append(something)
+        else:
+            logger.warning(context, "Item not found", extra=extra(
+                path=fs.path,
+                partition_key=item_loc.partition_key,
+                item_id=item_loc.item_id
+            ))
     return items
diff --git a/apps/examples/simple-example/src/simple_example/query_something.py b/apps/examples/simple-example/src/simple_example/query_something.py
index 04e0d467..1e220e06 100644
--- a/apps/examples/simple-example/src/simple_example/query_something.py
+++ b/apps/examples/simple-example/src/simple_example/query_something.py
@@ -9,7 +9,7 @@
 from hopeit.app.api import event_api
 from hopeit.app.context import EventContext, PostprocessHook
 from hopeit.app.logger import app_extra_logger
-from hopeit.fs_storage import FileStorage
+from hopeit.fs_storage import FileStorage, FileStorageSettings
 from model import Something, StatusType, Status, SomethingNotFound
 
 __steps__ = ['load', 'update_status_history']
@@ -17,7 +17,8 @@
 __api__ = event_api(
     summary="Simple Example: Query Something",
     query_args=[
-        ('item_id', str, 'Item Id to read')
+        ('item_id', str, 'Item Id to read'),
+        ('partition_key', str, 'Partition folder in `YYYY/MM/DD/HH` format')
     ],
     responses={
         200: (Something, "Something object returned when found"),
@@ -32,11 +33,15 @@
 async def __init_event__(context):
     global fs
     if fs is None:
-        fs = FileStorage(path=str(context.env['fs']['data_path']))
+        settings: FileStorageSettings = context.settings(
+            key="fs_storage", datatype=FileStorageSettings
+        )
+        fs = FileStorage.with_settings(settings)
 
 
 async def load(payload: None, context: EventContext, *,
-               item_id: str, update_status: bool = False) -> Union[Something, SomethingNotFound]:
+               item_id: str, partition_key: str,
+               update_status: bool = False) -> Union[Something, SomethingNotFound]:
     """
     Loads json file from filesystem as `Something` instance
 
@@ -48,10 +53,10 @@ async def load(payload: None, context: EventContext, *,
     """
     assert fs
     logger.info(context, "load", extra=extra(something_id=item_id, path=fs.path))
-    something = await fs.get(key=item_id, datatype=Something)
+    something = await fs.get(key=item_id, datatype=Something, partition_key=partition_key)
     if something is None:
         logger.warning(context, "item not found", extra=extra(something_id=item_id, path=fs.path))
-        return SomethingNotFound(str(fs.path), item_id)
+        return SomethingNotFound(str(fs.path / partition_key), item_id)
     return something
 
 
diff --git a/apps/examples/simple-example/src/simple_example/query_something_extended.py b/apps/examples/simple-example/src/simple_example/query_something_extended.py
index e077a868..8fcd1278 100644
--- a/apps/examples/simple-example/src/simple_example/query_something_extended.py
+++ b/apps/examples/simple-example/src/simple_example/query_something_extended.py
@@ -9,7 +9,7 @@
 from hopeit.app.api import event_api
 from hopeit.app.context import EventContext, PostprocessHook
 from hopeit.app.logger import app_extra_logger
-from hopeit.fs_storage import FileStorage
+from hopeit.fs_storage import FileStorage, FileStorageSettings
 from common.validation import validate
 from model import Something, Status, SomethingNotFound
 
@@ -18,7 +18,8 @@
 __api__ = event_api(
     summary="Simple Example: Query Something Extended",
     query_args=[
-        ('item_id', str, 'Item Id to read')
+        ('item_id', str, 'Item Id to read'),
+        ('partition_key', str, 'Partition folder in `YYYY/MM/DD/HH` format')
     ],
     payload=(Status, "Status change for the retrieved object"),
     responses={
@@ -34,11 +35,15 @@
 async def __init_event__(context):
     global fs
     if fs is None:
-        fs = FileStorage(path=str(context.env['fs']['data_path']))
+        settings: FileStorageSettings = context.settings(
+            key="fs_storage", datatype=FileStorageSettings
+        )
+        fs = FileStorage.with_settings(settings)
 
 
 async def load(payload: Status, context: EventContext, *,
-               item_id: str, update_status: bool = False) -> Union[Something, SomethingNotFound]:
+               item_id: str, partition_key: str,
+               update_status: bool = False) -> Union[Something, SomethingNotFound]:
     """
     Loads json file from filesystem as `Something` instance,
     sets status to specified payload
@@ -51,10 +56,10 @@ async def load(payload: Status, context: EventContext, *,
     """
     assert fs
     logger.info(context, "load", extra=extra(something_id=item_id, path=fs.path))
-    something = await fs.get(key=item_id, datatype=Something)
+    something = await fs.get(key=item_id, datatype=Something, partition_key=partition_key)
     if something is None:
         logger.warning(context, "item not found", extra=extra(something_id=item_id, path=fs.path))
-        return SomethingNotFound(str(fs.path), item_id)
+        return SomethingNotFound(str(fs.path / partition_key), item_id)
     if something.status and (something.status not in something.history):
         something.history.append(something.status)
     something.status = payload
diff --git a/apps/examples/simple-example/src/simple_example/save_something.py b/apps/examples/simple-example/src/simple_example/save_something.py
index f22d1ee9..406ee9b8 100644
--- a/apps/examples/simple-example/src/simple_example/save_something.py
+++ b/apps/examples/simple-example/src/simple_example/save_something.py
@@ -8,7 +8,7 @@
 from hopeit.app.api import event_api
 from hopeit.app.logger import app_extra_logger
 from hopeit.app.context import EventContext, PreprocessHook
-from hopeit.fs_storage import FileStorage
+from hopeit.fs_storage import FileStorage, FileStorageSettings
 
 from model import Something, User, SomethingParams
 from common.validation import validate
@@ -33,7 +33,10 @@
 async def __init_event__(context):
     global fs
     if fs is None:
-        fs = FileStorage(path=str(context.env['fs']['data_path']))
+        settings: FileStorageSettings = context.settings(
+            key="fs_storage", datatype=FileStorageSettings
+        )
+        fs = FileStorage.with_settings(settings)
 
 
 # pylint: disable=invalid-name
diff --git a/apps/examples/simple-example/src/simple_example/service/something_generator.py b/apps/examples/simple-example/src/simple_example/service/something_generator.py
index 05b79e96..bf14c534 100644
--- a/apps/examples/simple-example/src/simple_example/service/something_generator.py
+++ b/apps/examples/simple-example/src/simple_example/service/something_generator.py
@@ -4,26 +4,17 @@
 Creates and publish Something object every 10 seconds
 """
 import asyncio
-from typing import Optional
 import random
 
 from hopeit.app.events import Spawn
 from hopeit.app.logger import app_extra_logger
 from hopeit.app.context import EventContext
-from hopeit.fs_storage import FileStorage
 
 from model import Something, User, SomethingParams
 
 __steps__ = ['create_something']
 
 logger, extra = app_extra_logger()
-fs: Optional[FileStorage] = None
-
-
-async def __init_event__(context: EventContext):
-    global fs
-    if fs is None:
-        fs = FileStorage(path=str(context.env['fs']['data_path']))
 
 
 async def __service__(context: EventContext) -> Spawn[SomethingParams]:
diff --git a/apps/examples/simple-example/src/simple_example/shuffle/parallelize_event.py b/apps/examples/simple-example/src/simple_example/shuffle/parallelize_event.py
index 661e936d..1f3331da 100644
--- a/apps/examples/simple-example/src/simple_example/shuffle/parallelize_event.py
+++ b/apps/examples/simple-example/src/simple_example/shuffle/parallelize_event.py
@@ -14,7 +14,7 @@
 from hopeit.dataobjects import dataobject
 from hopeit.app.events import Spawn, SHUFFLE
 from hopeit.app.logger import app_extra_logger
-from hopeit.fs_storage import FileStorage
+from hopeit.fs_storage import FileStorage, FileStorageSettings
 from model import Something, SomethingStored, Status, StatusType
 
 logger, extra = app_extra_logger()
@@ -48,7 +48,10 @@ class SecondPart:
 async def __init_event__(context):
     global fs
     if fs is None:
-        fs = FileStorage(path=str(context.env['fs']['data_path']))
+        settings: FileStorageSettings = context.settings(
+            key="fs_storage", datatype=FileStorageSettings
+        )
+        fs = FileStorage.with_settings(settings)
 
 
 async def fork_something(payload: Something, context: EventContext) -> Spawn[Union[FirstPart, SecondPart]]:
diff --git a/apps/examples/simple-example/src/simple_example/shuffle/spawn_event.py b/apps/examples/simple-example/src/simple_example/shuffle/spawn_event.py
index 16a48b4b..646a175b 100644
--- a/apps/examples/simple-example/src/simple_example/shuffle/spawn_event.py
+++ b/apps/examples/simple-example/src/simple_example/shuffle/spawn_event.py
@@ -11,7 +11,7 @@
 from hopeit.app.context import EventContext, PostprocessHook
 from hopeit.app.events import Spawn, SHUFFLE
 from hopeit.app.logger import app_extra_logger
-from hopeit.fs_storage import FileStorage
+from hopeit.fs_storage import FileStorage, FileStorageSettings
 from model import Something, SomethingStored, Status, StatusType
 
 logger, extra = app_extra_logger()
@@ -33,7 +33,10 @@
 async def __init_event__(context):
     global fs
     if fs is None:
-        fs = FileStorage(path=str(context.env['fs']['data_path']))
+        settings: FileStorageSettings = context.settings(
+            key="fs_storage", datatype=FileStorageSettings
+        )
+        fs = FileStorage.with_settings(settings)
 
 
 async def spawn_many_events(payload: Something, context: EventContext) -> Spawn[Something]:
@@ -72,7 +75,7 @@ def update_status(payload: Something, context: EventContext) -> Something:
     if payload.status:
         payload.history.append(payload.status)
     payload.status = Status(
-        ts=datetime.now(),
+        ts=datetime.now(tz=timezone.utc),
         type=StatusType.PROCESSED
     )
     return payload
diff --git a/apps/examples/simple-example/src/simple_example/streams/process_events.py b/apps/examples/simple-example/src/simple_example/streams/process_events.py
index a8ac3887..0f538702 100644
--- a/apps/examples/simple-example/src/simple_example/streams/process_events.py
+++ b/apps/examples/simple-example/src/simple_example/streams/process_events.py
@@ -10,7 +10,7 @@
 
 from hopeit.app.logger import app_extra_logger
 from hopeit.app.context import EventContext
-from hopeit.fs_storage import FileStorage
+from hopeit.fs_storage import FileStorage, FileStorageSettings
 from model import Something, SomethingStored, Status, StatusType
 
 __steps__ = ['update_status', 'save']
@@ -23,7 +23,10 @@
 async def __init_event__(context):
     global fs
     if fs is None:
-        fs = FileStorage(path=str(context.env['fs']['data_path']))
+        settings: FileStorageSettings = context.settings(
+            key="fs_storage", datatype=FileStorageSettings
+        )
+        fs = FileStorage.with_settings(settings)
 
 
 def update_status(payload: Something, context: EventContext) -> Something:
diff --git a/apps/tutorials/05-streaming-app/my_streaming_app/process_message.py b/apps/tutorials/05-streaming-app/my_streaming_app/process_message.py
index 3c553513..07f3ac49 100644
--- a/apps/tutorials/05-streaming-app/my_streaming_app/process_message.py
+++ b/apps/tutorials/05-streaming-app/my_streaming_app/process_message.py
@@ -8,7 +8,7 @@
 
 from hopeit.app.context import EventContext
 from hopeit.app.logger import app_extra_logger
-from hopeit.toolkit.storage.fs import FileStorage
+from hopeit.fs_storage import FileStorage
 
 from .data_model import Status, MyMessage
 
diff --git a/docs/source/benchmark/dedicated.rst b/docs/source/benchmark/dedicated.rst
index d651e30c..b18160f2 100644
--- a/docs/source/benchmark/dedicated.rst
+++ b/docs/source/benchmark/dedicated.rst
@@ -12,11 +12,11 @@ real time") for every request in the 99th percentile. The result is a rate (requ
 
 Three endpoints are evaluated in each test scenario
 
-* http://hostname:port/api/simple-benchmark/0x14/give-me-something retrieves information from objects randomly created
+* http://hostname:port/api/simple-benchmark/0x15/give-me-something retrieves information from objects randomly created
 
-* http://hostname:port/api/simple-benchmark/0x14/query-something-redis returns information from objects stored on a Redis server (memory store)
+* http://hostname:port/api/simple-benchmark/0x15/query-something-redis returns information from objects stored on a Redis server (memory store)
 
-* http://hostname:port/api/simple-benchmark/0x14/query-something-fs retrieves information from objects stored on a file system (SSD drive)
+* http://hostname:port/api/simple-benchmark/0x15/query-something-fs retrieves information from objects stored on a file system (SSD drive)
 
 Scenarios:
 __________
@@ -68,13 +68,13 @@ Now we are ready to start the simple-benchmark app to run the tests
 
 Latency (HdrHistogram - Uncorrected Latency, measured without taking delayed starts into account) results:
 
-./wrk2/wrk -t8 -c34 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x14/give-me-something?item_id=string"
+./wrk2/wrk -t8 -c34 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x15/give-me-something?item_id=string"
  p99 30.38ms @ 1794.29 req/s
 
-./wrk2/wrk -t6 -c32 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x14/query-something-redis?item_id=string"
+./wrk2/wrk -t6 -c32 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x15/query-something-redis?item_id=string"
  p99 27.18ms @ 1226.51 req/s
 
-./wrk2/wrk -t4 -c18 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x14/query-something-fs?item_id=string"
+./wrk2/wrk -t4 -c18 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x15/query-something-fs?item_id=string"
  p99 31.12ms @ 803.38 req/s
 
 2 . Docker, single/multiple instances of hopeit.engine
@@ -103,22 +103,22 @@ Latency (HdrHistogram - Uncorrected Latency, measured without taking delayed sta
 
 4 hopeit.engine instances behind a reverse proxy
 
-./wrk2/wrk -t8 -c40 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x14/give-me-something?item_id=string"
+./wrk2/wrk -t8 -c40 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x15/give-me-something?item_id=string"
  p99 27.77ms @ 2587.95 req/s
 
-./wrk2/wrk -t6 -c32 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x14/query-something-redis?item_id=string"
+./wrk2/wrk -t6 -c32 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x15/query-something-redis?item_id=string"
  p99 28.67ms @ 1810.14 req/s
 
-./wrk2/wrk -t4 -c20 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x14/query-something-fs?item_id=string"
+./wrk2/wrk -t4 -c20 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x15/query-something-fs?item_id=string"
  p99 30.29ms @ 1241.39 req/s
 
 1 hopeit.engine instance
 
-./wrk2/wrk -t8 -c34 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x14/give-me-something?item_id=string"
+./wrk2/wrk -t8 -c34 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x15/give-me-something?item_id=string"
  p99 26.48ms @ 1336.18 req/s
 
-./wrk2/wrk -t6 -c24 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x14/query-something-redis?item_id=string"
+./wrk2/wrk -t6 -c24 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x15/query-something-redis?item_id=string"
  p99 29.14ms @ 908.93 req/s
 
-./wrk2/wrk -t4 -c14 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x14/query-something-fs?item_id=string"
+./wrk2/wrk -t4 -c14 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x15/query-something-fs?item_id=string"
  p99 24.77ms @ 624.77 req/s
\ No newline at end of file
diff --git a/docs/source/release-notes.rst b/docs/source/release-notes.rst
index 7398cba9..9de18b3f 100644
--- a/docs/source/release-notes.rst
+++ b/docs/source/release-notes.rst
@@ -1,6 +1,22 @@
 Release Notes
 =============
 
+
+Version 0.15.0
+______________
+-Plugins:
+  - fs_storage: event implementation to support persisting directly from a stream into disk by adding event to configuration:
+  using ```"impl": "hopeit.fs_storage.events.stream_batch_storage"``` in event configuration.
+
+-Engine:
+  - Support for custom implementation for events, enable to reuse code provided in external libraries or plugins as events
+  in an app, like `hopeit.fs_storage.events.stream_batch_storage` to persist data from streams to disks directly.
+  - Support for generic `DataObject` as a type for step payload. Enable creating generic events that can be reused
+  among apps.
+  - Extended EventDescriptor configuration, to provide a list of dataobject types that can be handled by generic events
+  using `DataObject` payload.
+
+
 Version 0.14.2
 ______________
 - Fix: removed global security section from generate openapi file to allow per event configuration to take precedence
diff --git a/engine/requirements-dev.txt b/engine/requirements-dev.txt
index f81da13c..1e1ff420 100644
--- a/engine/requirements-dev.txt
+++ b/engine/requirements-dev.txt
@@ -11,5 +11,6 @@ coverage-badge
 mypy
 wheel
 aiofiles
+types-aiofiles
 nest-asyncio
 
diff --git a/engine/requirements.lock b/engine/requirements.lock
index fb5e3c94..5c2d887f 100644
--- a/engine/requirements.lock
+++ b/engine/requirements.lock
@@ -1,61 +1,34 @@
-aiofiles==0.8.0
 aiohttp==3.8.1
 aiohttp-cors==0.7.0
-aiohttp-swagger3==0.7.0
+aiohttp-swagger3==0.7.1
 aioredis==2.0.1
 aiosignal==1.2.0
-astroid==2.9.3
 async-timeout==4.0.2
 attrs==21.4.0
-certifi==2020.6.20
+certifi==2021.10.8
 cffi==1.15.0
-charset-normalizer==2.0.10
-click==8.0.3
-coverage==6.3
-coverage-badge==1.1.0
-cryptography==36.0.1
-dataclasses-jsonschema==2.15.0
+charset-normalizer==2.0.12
+click==8.1.2
+cryptography==36.0.2
+dataclasses-jsonschema==2.15.1
 deepdiff==5.7.0
 fastjsonschema==2.15.3
-flake8==4.0.1
 frozenlist==1.3.0
 idna==3.3
-iniconfig==1.1.1
-isort==5.10.1
 jsonschema==4.4.0
-lazy-object-proxy==1.7.1
 lz4==3.1.10
-mccabe==0.6.1
 multidict==6.0.2
-mypy==0.931
 mypy-extensions==0.4.3
 ordered-set==4.0.2
-packaging==21.3
-platformdirs==2.4.1
-pluggy==1.0.0
-py==1.11.0
-pycodestyle==2.8.0
 pycparser==2.21
-pyflakes==2.4.0
 PyJWT==2.3.0
-pylint==2.12.2
-pyparsing==3.0.7
 pyrsistent==0.18.1
-pytest==6.2.5
-pytest-aiohttp==1.0.3
-pytest-asyncio==0.17.2
-pytest-cov==3.0.0
-pytest-mock==3.7.0
-pytest-order==1.0.1
 python-dateutil==2.8.2
 PyYAML==6.0
 six==1.16.0
 strict-rfc3339==0.7
 stringcase==1.2.0
-toml==0.10.2
-tomli==2.0.0
 typing-inspect==0.7.1
-typing_extensions==4.0.1
-watchdog==2.1.6
-wrapt==1.13.3
+typing_extensions==4.1.1
+watchdog==2.1.7
 yarl==1.7.2
diff --git a/engine/src/hopeit/app/api.py b/engine/src/hopeit/app/api.py
index af523e30..f582a047 100644
--- a/engine/src/hopeit/app/api.py
+++ b/engine/src/hopeit/app/api.py
@@ -17,7 +17,7 @@
            'event_api',
            'app_base_route_name']
 
-ArgType = TypeVar("ArgType", bound=Type)
+ArgType = TypeVar("ArgType", bound=Type)  # pylint: disable=invalid-name
 ArgDef = Union[str, Tuple[str, ArgType], Tuple[str, ArgType, str]]
 PayloadDef = Union[Type, Tuple[Type, str]]
 
diff --git a/engine/src/hopeit/app/config.py b/engine/src/hopeit/app/config.py
index 62c2035e..e2a4be63 100644
--- a/engine/src/hopeit/app/config.py
+++ b/engine/src/hopeit/app/config.py
@@ -297,16 +297,44 @@ class EventConnection:
 @dataclass
 class EventDescriptor:
     """
-    Event descriptor
+    Event Descriptor: configures event implementation
+
+    :field: type, EventType: type of event i.e.: GET, POST, MULTIPART, STREAM, SERVICE
+    :field: plug_mode, EventPlugMode: defines whether an event defined in a plugin is created in the
+        current app (ON_APP) or it will be created in the original plugin (STANDALONE, default)
+    :field: route, optional str: custom route for endpoint. If not specified route will be derived
+        from `/api/app_name/app_version/event_name`
+    :field: impl, optional str: custom event implementation Python module. If not specified, module
+        with same same as event will be imported.
+    :field: connections, list of EventConnection: specifies dependencies on other apps/endpoints,
+        that can be used by client plugins to call events on external apps
+    :field: read_stream, optional ReadStreamDescriptor: specifies source stream to read from.
+        Valid only for STREAM events.
+    :field: write_stream, optional WriteStreamDescriptor: for any type of events, resultant dataobjects will
+        be published to the specified stream.
+    :field: auth, list of AuthType: supported authentication schemas for this event. If not specified
+        application default will be used.
+    :field: setting_keys, list of str: by default EventContext will have access to the settings section
+        with the same name of the event using `settings = context.settings(datatype=MySettingsType)`.
+        In case additional sections are needed to be accessed from
+        EventContext, then a list of setting keys, including the name of the event if needed,
+        can be specified here. Then access to a `custom` key can be done using
+        `custom_settings = context.settings(key="customer", datatype=MyCustomSettingsType)`
+    :field: dataobjects, list of str: list of full qualified dataobject types that this event can process.
+        When not specified, the engine will inspect the module implementation and find all datatypes supported
+        as payload in the functions defined as `__steps__`. In case of generic functions that support
+        `payload: DataObject` argument, then a list of full qualified datatypes must be specified here.
     """
     type: EventType
     plug_mode: EventPlugMode = EventPlugMode.STANDALONE
     route: Optional[str] = None
+    impl: Optional[str] = None
     connections: List[EventConnection] = field(default_factory=list)
     read_stream: Optional[ReadStreamDescriptor] = None
     write_stream: Optional[WriteStreamDescriptor] = None
     auth: List[AuthType] = field(default_factory=list)
     setting_keys: List[str] = field(default_factory=list)
+    dataobjects: List[str] = field(default_factory=list)
 
     def __post_init__(self):
         if self.read_stream:
diff --git a/engine/src/hopeit/app/context.py b/engine/src/hopeit/app/context.py
index a8149730..4ab5d6d9 100644
--- a/engine/src/hopeit/app/context.py
+++ b/engine/src/hopeit/app/context.py
@@ -50,7 +50,7 @@ def __init__(self, *,
         self.settings = settings
         base_event = event_name.split('$')[0]
         self.event_info: EventDescriptor = plugin_config.events[base_event]
-        self.creation_ts: datetime = datetime.now().astimezone(tz=timezone.utc)
+        self.creation_ts: datetime = datetime.now(tz=timezone.utc)
         self.auth_info = auth_info
         track_fields = [
             'track.operation_id',
diff --git a/engine/src/hopeit/dataobjects/__init__.py b/engine/src/hopeit/dataobjects/__init__.py
index 93e26acc..3cc464ee 100644
--- a/engine/src/hopeit/dataobjects/__init__.py
+++ b/engine/src/hopeit/dataobjects/__init__.py
@@ -78,7 +78,7 @@ def event_ts(self) -> Optional[datetime]:
 
 DataObject = TypeVar("DataObject", bound=JsonSchemaMixin)
 EventPayload = Union[str, int, float, bool, dict, set, list, DataObject]
-EventPayloadType = TypeVar("EventPayloadType")
+EventPayloadType = TypeVar("EventPayloadType")  # pylint: disable=invalid-name
 
 
 class BinaryAttachment:
diff --git a/engine/src/hopeit/dataobjects/payload.py b/engine/src/hopeit/dataobjects/payload.py
index ba9a3d07..619cf9dc 100644
--- a/engine/src/hopeit/dataobjects/payload.py
+++ b/engine/src/hopeit/dataobjects/payload.py
@@ -37,12 +37,14 @@ def from_json(json_str: Union[str, bytes],
             return datatype(json.loads(json_str).get(key))  # type: ignore
         if datatype in _COLLECTION_TYPES:
             return datatype(json.loads(json_str))  # type: ignore
-        assert getattr(datatype, 'from_json'), \
-            f"{datatype} should be annotated with @dataobject"
         try:
             return datatype.from_json(json_str, validate=datatype.__data_object__['validate'])  # type: ignore
         except ValidationError as e:
             raise ValueError(f"Cannot read JSON: type={datatype} validation_error={str(e)}") from e
+        except Exception:
+            assert getattr(datatype, 'from_json'), \
+                f"{datatype} should be annotated with @dataobject"
+            raise  # Raises unexpected exceptions, if assert block does not catch missing @dataobject
 
     @staticmethod
     def from_obj(data: Union[dict, list],
diff --git a/engine/src/hopeit/server/api.py b/engine/src/hopeit/server/api.py
index 4400709b..a1efacde 100644
--- a/engine/src/hopeit/server/api.py
+++ b/engine/src/hopeit/server/api.py
@@ -441,8 +441,8 @@ def _update_api_schemas(app_config: AppConfig):
     """
     assert spec is not None
     schemas = spec['components'].get('schemas', {})
-    for event_name in app_config.events.keys():
-        event_schemas = _generate_schemas(app_config, event_name)
+    for event_name, event_info in app_config.events.items():
+        event_schemas = _generate_schemas(app_config, event_name, event_info)
         for name, event_schema in event_schemas.items():
             if name in runtime_schemas:
                 if not event_schema == schemas.get(name):
@@ -509,7 +509,9 @@ def _update_api_paths(app_config: AppConfig, plugin: Optional[AppConfig] = None)
         method = METHOD_MAPPING.get(event_info.type)
         if method is None:
             continue
-        event_api_spec = _extract_event_api_spec(app_config if plugin is None else plugin, event_name)
+        event_api_spec = _extract_event_api_spec(
+            app_config if plugin is None else plugin, event_name, event_info
+        )
         if event_api_spec is None:
             event_api_spec = paths.get(route, {}).get(method)
         if event_api_spec is None and _options.get('generate_mode'):
@@ -592,11 +594,11 @@ def _set_path_security(event_api_spec: dict, app_config: AppConfig, event_info:
         event_api_spec['security'] = security
 
 
-def _extract_event_api_spec(app_config: AppConfig, event_name: str) -> Optional[dict]:
+def _extract_event_api_spec(app_config: AppConfig, event_name: str, event_info: EventDescriptor) -> Optional[dict]:
     """
     Extract __api__ definition from event implementation
     """
-    module = find_event_handler(app_config=app_config, event_name=event_name)
+    module = find_event_handler(app_config=app_config, event_name=event_name, event_info=event_info)
     if hasattr(module, '__api__'):
         method_spec = getattr(module, '__api__')
         if isinstance(method_spec, dict):
@@ -605,11 +607,11 @@ def _extract_event_api_spec(app_config: AppConfig, event_name: str) -> Optional[
     return None
 
 
-def _generate_schemas(app_config: AppConfig, event_name: str) -> dict:
+def _generate_schemas(app_config: AppConfig, event_name: str, event_info: EventDescriptor) -> dict:
     """
     Generate all schemas for a given event, based on steps signatures
     """
-    module = find_event_handler(app_config=app_config, event_name=event_name)
+    module = find_event_handler(app_config=app_config, event_name=event_name, event_info=event_info)
     steps = extract_module_steps(module)
     schemas: dict = {}
     for _, step_info in steps:
diff --git a/engine/src/hopeit/server/config.py b/engine/src/hopeit/server/config.py
index aa9f8bca..376ff8d4 100644
--- a/engine/src/hopeit/server/config.py
+++ b/engine/src/hopeit/server/config.py
@@ -24,7 +24,7 @@
 
 
 DEFAULT_STR = "<<DEFAULT>>"
-ConfigType = TypeVar("ConfigType")
+ConfigType = TypeVar("ConfigType")  # pylint: disable=invalid-name
 
 
 @dataobject
diff --git a/engine/src/hopeit/server/engine.py b/engine/src/hopeit/server/engine.py
index febc9c30..fd73c201 100644
--- a/engine/src/hopeit/server/engine.py
+++ b/engine/src/hopeit/server/engine.py
@@ -8,14 +8,15 @@
 from datetime import datetime, timezone
 from typing import Awaitable, Optional, Dict, List, Union, Tuple, Any
 
-from hopeit.server.imports import find_event_handler
-from hopeit.server.steps import split_event_stages, event_and_step, extract_module_steps, effective_steps
+from hopeit.server.imports import find_datobject_type, find_event_handler
+from hopeit.server.steps import split_event_stages, event_and_step, extract_module_steps, \
+    effective_steps
 from hopeit.toolkit import auth
 from hopeit.app.config import AppConfig, EventSettings, EventType, ReadStreamDescriptor, EventDescriptor, \
     StreamQueue, StreamQueueStrategy
 from hopeit.app.context import EventContext, PostprocessHook, PreprocessHook
 from hopeit.app.client import register_app_connections, stop_app_connections
-from hopeit.dataobjects import EventPayload
+from hopeit.dataobjects import DataObject, EventPayload
 from hopeit.server.config import ServerConfig
 from hopeit.server.events import EventHandler, get_event_settings, get_runtime_settings
 from hopeit.streams import stream_auth_info, StreamEvent, StreamOSError, StreamManager
@@ -380,7 +381,7 @@ async def read_stream(self, *,
                     consumer_group=stream_info.consumer_group
                 )
 
-            datatypes = self._find_stream_datatype_handlers(event_name)
+            datatypes = self._find_stream_datatype_handlers(event_name, event_config)
             log_info['name'] = stream_info.name
             log_info['consumer_group'] = stream_info.consumer_group
             assert not self._running[event_name].locked(), f"Event already running {event_name}"
@@ -452,7 +453,7 @@ def _service_event_context(self, event_name: str, event_settings: EventSettings,
         if previous_context is None:
             track_ids = {
                 'track.request_id': str(uuid.uuid4()),
-                'track.request_ts': datetime.now().astimezone(timezone.utc).isoformat()
+                'track.request_ts': datetime.now(tz=timezone.utc).isoformat()
             }
         else:
             track_ids = previous_context.track_ids
@@ -493,7 +494,8 @@ async def service_loop(self, *,
                         extra=extra(prefix='service.', **log_info))
             await asyncio.sleep(wait)
         logger.info(__name__, "Starting service...", extra=extra(prefix='service.', **log_info))
-        impl = find_event_handler(app_config=self.app_config, event_name=event_name)
+        event_config = self.effective_events[event_name]
+        impl = find_event_handler(app_config=self.app_config, event_name=event_name, event_info=event_config)
         service_handler = getattr(impl, '__service__')
         assert service_handler is not None, \
             f"{event_name} must implement method `__service__(context) -> Spawn[...]` to run as a service"
@@ -547,27 +549,47 @@ async def stop_event(self, event_name: str):
 
     @staticmethod
     def _config_effective_events(app_config: AppConfig) -> Dict[str, EventDescriptor]:
+        """
+        Return effective events computed from user app config.
+
+        Effective events could be result of splitting a single event in stages,
+        using the "SHUFFLE" keyword, that will internaally generate 2 events.
+        Or for STREAMS that implementes the `__service__` method, both a STREAM
+        and a SERVICE event will be generated.
+        """
         effective_events: Dict[str, EventDescriptor] = {}
         for event_name, event_info in app_config.events.items():
-            impl = find_event_handler(app_config=app_config, event_name=event_name)
+            impl = find_event_handler(app_config=app_config, event_name=event_name, event_info=event_info)
+            # Add events resultatn of splitting steps on SHUFFLE (stages)
             splits = split_event_stages(app_config.app, event_name, event_info, impl)
             effective_events.update(**splits)
+            # Add associated SERVICE events to streams
+            if event_info.type == EventType.STREAM and hasattr(impl, "__service__"):
+                effective_events[f"{event_name}$__service__"] = EventDescriptor(
+                    type=EventType.SERVICE,
+                    connections=event_info.connections,
+                    impl=event_info.impl,
+                )
         return effective_events
 
-    def _find_stream_datatype_handlers(self, event_name: str) -> Dict[str, type]:
+    def _find_stream_datatype_handlers(self, event_name: str, event_info: EventDescriptor) -> Dict[str, type]:
         """
         Computes a dictionary of `{datatype name: datatype class}` that event steps
         can be handle when consuming from an stream.
         """
         base_event, _ = event_and_step(event_name)
-        impl = find_event_handler(app_config=self.app_config, event_name=base_event)
+        impl = find_event_handler(app_config=self.app_config, event_name=base_event, event_info=event_info)
         all_steps = extract_module_steps(impl)
         steps = effective_steps(event_name, all_steps)
         datatypes = {}
         for _, _, step in steps:
             _, datatype, _, _ = step
             if hasattr(datatype, '__stream_event__'):
-                datatypes[datatype.__name__] = datatype
+                datatypes[f"{datatype.__module__}.{datatype.__qualname__}"] = datatype
+            elif datatype is DataObject:
+                for type_name in event_info.dataobjects:
+                    datatype = find_datobject_type(type_name)
+                    datatypes[f"{datatype.__module__}.{datatype.__qualname__}"] = datatype
         if len(datatypes) == 0:
             raise NotImplementedError(f"No data types found to read from stream in event={event_name}. "
                                       "Dataclasses must be decorated with `@dataobject` to be used in streams")
diff --git a/engine/src/hopeit/server/events.py b/engine/src/hopeit/server/events.py
index 4bcd615e..1217b57e 100644
--- a/engine/src/hopeit/server/events.py
+++ b/engine/src/hopeit/server/events.py
@@ -49,9 +49,9 @@ def load_modules(self, effective_events: Dict[str, EventDescriptor]):
         """
         Load and setup event implementation modules. Set logger and load settings.
         """
-        for event_name in effective_events.keys():
+        for event_name, event_info in effective_events.items():
             base_event, _ = event_and_step(event_name)
-            module = find_event_handler(app_config=self.app_config, event_name=base_event)
+            module = find_event_handler(app_config=self.app_config, event_name=base_event, event_info=event_info)
             steps = extract_module_steps(module)
             self.modules[base_event] = (module, False, steps)
             self.preprocess_handlers[base_event] = extract_preprocess_handler(module)
diff --git a/engine/src/hopeit/server/imports.py b/engine/src/hopeit/server/imports.py
index 102eb04b..e3ba7fa5 100644
--- a/engine/src/hopeit/server/imports.py
+++ b/engine/src/hopeit/server/imports.py
@@ -3,21 +3,30 @@
 """
 from importlib import import_module
 from types import ModuleType
+from typing import Type
 
-from hopeit.app.config import AppConfig
+from hopeit.app.config import AppConfig, EventDescriptor
 from hopeit.server.names import module_name
 
+from hopeit.dataobjects import DataObject
+
 __all__ = ['find_event_handler']
 
 
 def find_event_handler(*, app_config: AppConfig,
-                       event_name: str) -> ModuleType:
+                       event_name: str,
+                       event_info: EventDescriptor) -> ModuleType:
     """
     Returns the initialized module implementing the event business logic.
     """
-    imps = app_config.engine.import_modules \
-        if app_config.engine and app_config.engine.import_modules \
+    if event_info.impl:
+        return import_module(event_info.impl)
+
+    imps = (
+        app_config.engine.import_modules
+        if app_config.engine and app_config.engine.import_modules
         else [app_config.app.name]
+    )
 
     errors = []
     for imp in imps:
@@ -27,3 +36,15 @@ def find_event_handler(*, app_config: AppConfig,
         except ImportError as e:
             errors.append(e)
     raise ImportError(f"Cannot import {event_name} from {imps}", *errors)
+
+
+def find_datobject_type(qual_type_name: str) -> Type[DataObject]:
+    mod_name,  type_name = (
+        '.'.join(qual_type_name.split('.')[:-1]),
+        qual_type_name.split('.')[-1]
+    )
+    module = import_module(mod_name)
+    datatype = getattr(module, type_name)
+    assert hasattr(datatype, "__data_object__"), \
+        f"Type {qual_type_name} must be annotated with `@dataobject`."
+    return datatype
diff --git a/engine/src/hopeit/server/metrics.py b/engine/src/hopeit/server/metrics.py
index 5a3c3f3f..f9b628c9 100644
--- a/engine/src/hopeit/server/metrics.py
+++ b/engine/src/hopeit/server/metrics.py
@@ -43,7 +43,7 @@ def stream_metrics(context: EventContext):
 
 def _calc_event_metrics(context: EventContext):
     duration = 1000.0 * (
-        datetime.now().astimezone(tz=timezone.utc) - context.creation_ts
+        datetime.now(tz=timezone.utc) - context.creation_ts
     ).total_seconds()
     return {
         'duration': f"{duration:.3f}"
@@ -73,7 +73,7 @@ def _calc_stream_metrics(context: EventContext):
     if request_ts:
         request_dt = datetime.fromisoformat(request_ts)
         elapsed = 1000.0 * (
-            datetime.now().astimezone(timezone.utc) - request_dt
+            datetime.now(tz=timezone.utc) - request_dt
         ).total_seconds()
         result['request_elapsed'] = f"{elapsed:.3f}"
     return result
@@ -84,7 +84,7 @@ class StreamStats:
     Helper class to keep stream consuming stats
     """
     def __init__(self):
-        self.run_ts: datetime = datetime.now()
+        self.run_ts: datetime = datetime.now(tz=timezone.utc)
         self.start_ts: Optional[datetime] = None
         self.from_ts: Optional[datetime] = None
         self.event_count: int = 0
@@ -94,8 +94,8 @@ def __init__(self):
 
     def ensure_start(self):
         if self.start_ts is None:
-            self.start_ts = datetime.now()
-            self.from_ts = datetime.now()
+            self.start_ts = datetime.now(tz=timezone.utc)
+            self.from_ts = datetime.now(tz=timezone.utc)
             self.event_count: int = 0
             self.error_count: int = 0
             self.total_event_count = 0
@@ -121,7 +121,7 @@ def calc(self) -> Dict[str, Union[int, float]]:
         """
         assert self.start_ts is not None and self.from_ts is not None, \
             "StreamStats not initialized. Call `ensure_start()`"
-        now = datetime.now()
+        now = datetime.now(tz=timezone.utc)
         total_elapsed_td = now - self.start_ts
         total_elapsed = 1000.0 * total_elapsed_td.total_seconds()
         avg_rate = (1000.0 * self.total_event_count / total_elapsed) if total_elapsed else 0.0
diff --git a/engine/src/hopeit/server/steps.py b/engine/src/hopeit/server/steps.py
index 6e25ee82..69a113e3 100644
--- a/engine/src/hopeit/server/steps.py
+++ b/engine/src/hopeit/server/steps.py
@@ -2,7 +2,7 @@
 Handling sequencing and execution of steps from events
 """
 import asyncio
-from datetime import datetime
+from datetime import datetime, timezone
 from functools import partial
 from types import ModuleType
 from typing import Type, Dict, Any, Optional, Callable, Tuple, AsyncGenerator, List, Union
@@ -11,7 +11,7 @@
 from hopeit.app.config import AppConfig, AppDescriptor, EventDescriptor, EventType, \
     ReadStreamDescriptor, StreamQueueStrategy, WriteStreamDescriptor
 from hopeit.app.context import EventContext
-from hopeit.dataobjects import EventPayload, copy_payload, EventPayloadType
+from hopeit.dataobjects import DataObject, EventPayload, copy_payload, EventPayloadType
 from hopeit.server.imports import find_event_handler
 from hopeit.server.logger import engine_logger, extra_logger
 from hopeit.server.names import auto_path
@@ -126,7 +126,10 @@ def extract_input_type(impl: ModuleType, from_step: Optional[str] = None) -> Typ
 def event_and_step(event_name: str) -> Tuple[str, Optional[str]]:
     comps = event_name.split('$')
     if len(comps) > 1:
-        return comps[0], comps[1]
+        return (
+            comps[0],
+            comps[1] if comps[1][0] != "_" else None  # Ignore special suffixes after '$' (i.e. `$__service__`)
+        )
     return event_name, None
 
 
@@ -199,7 +202,7 @@ async def execute_steps(steps: StepExecutionList,
     It will try to find next step in configuration order that matches input type of the payload,
     and will updated the payload and invoke next valid step.
     """
-    start_ts = datetime.now()
+    start_ts = datetime.now(tz=timezone.utc)
     step_delay = context.settings.stream.step_delay / 1000.0
     throttle_ms = context.settings.stream.throttle_ms
     i, func, is_spawn = _find_next_step(payload, steps, from_index=0)
@@ -242,20 +245,37 @@ async def invoke_single_step(func: Callable, *,
     return await _invoke_step(payload=payload, func=func, context=context, **kwargs)
 
 
-def _find_next_step(payload: Optional[EventPayload],
-                    steps: StepExecutionList,
-                    from_index: int) -> Tuple[int, Optional[Callable], bool]:
+def _find_next_step(
+    payload: Optional[EventPayload],
+    steps: StepExecutionList,
+    from_index: int
+) -> Tuple[int, Optional[Callable], bool]:
     """
     Finds next step to exectute in pending_steps list, base on the payload data type
     """
+    # Returns in case no more steps are available
     if from_index >= len(steps):
         return -1, None, False
+    # Try to match specific payload datatypes incluiding None
     for i, _, step_info in steps[from_index:]:
         func, input_type, _, is_iterable = step_info
         if input_type is None and payload is None:
             return i, func, is_iterable
-        if input_type is not None and isinstance(payload, input_type):
+        if (
+            input_type is not None
+            and input_type is not DataObject
+            and isinstance(payload, input_type)
+        ):
+            return i, func, is_iterable
+    # Return if there was no match for None payload
+    if payload is None:
+        return -1, None, False
+    # Finally try to match generic `payload: DataObject` argument in step
+    for i, _, step_info in steps[from_index:]:
+        func, input_type, _, is_iterable = step_info
+        if input_type is DataObject and payload is not None:
             return i, func, is_iterable
+    # No step input datatype matches current payload
     return -1, None, False
 
 
@@ -265,7 +285,7 @@ async def _throttle(context: EventContext, throttle_ms: int, start_ts: datetime)
     """
     delay = 0.0
     if throttle_ms:
-        elapsed_td = datetime.now() - start_ts
+        elapsed_td = datetime.now(tz=timezone.utc) - start_ts
         elapsed = 1000.0 * elapsed_td.total_seconds()
         delay = max(0.0, 1.0 * throttle_ms - elapsed)
     if delay > 0.0:
@@ -310,6 +330,7 @@ def split_event_stages(app: AppDescriptor,
             type=event_type,
             read_stream=read_stream,
             connections=event_info.connections,
+            impl=event_info.impl,
             write_stream=WriteStreamDescriptor(
                 name=intermediate_stream,
                 queue_strategy=StreamQueueStrategy.PROPAGATE
@@ -324,9 +345,9 @@ def split_event_stages(app: AppDescriptor,
     return effective_events
 
 
-def find_datatype_handler(*, app_config: AppConfig, event_name: str):
+def find_datatype_handler(*, app_config: AppConfig, event_name: str, event_info: EventDescriptor):
     base_event, from_step = event_and_step(event_name)
-    impl = find_event_handler(app_config=app_config, event_name=base_event)
+    impl = find_event_handler(app_config=app_config, event_name=base_event, event_info=event_info)
     datatype = extract_input_type(impl, from_step=from_step)
     return datatype
 
diff --git a/engine/src/hopeit/server/version.py b/engine/src/hopeit/server/version.py
index b837c49e..4cc93276 100644
--- a/engine/src/hopeit/server/version.py
+++ b/engine/src/hopeit/server/version.py
@@ -8,7 +8,7 @@
 import sys
 
 ENGINE_NAME = "hopeit.engine"
-ENGINE_VERSION = "0.14.2"
+ENGINE_VERSION = "0.15.0"
 
 # Major.Minor version to be used in App versions and Api endpoints for Apps/Plugins
 APPS_API_VERSION = '.'.join(ENGINE_VERSION.split('.')[0:2])
diff --git a/engine/src/hopeit/server/web.py b/engine/src/hopeit/server/web.py
index 1912d47c..4a3c3d83 100644
--- a/engine/src/hopeit/server/web.py
+++ b/engine/src/hopeit/server/web.py
@@ -239,19 +239,19 @@ def _setup_app_event_routes(app_engine: AppEngine,
         if event_info.type == EventType.POST:
             web_server.add_routes([
                 _create_post_event_route(
-                    app_engine, plugin=plugin, event_name=event_name, override_route_name=event_info.route
+                    app_engine, plugin=plugin, event_name=event_name, event_info=event_info
                 )
             ])
         elif event_info.type == EventType.GET:
             web_server.add_routes([
                 _create_get_event_route(
-                    app_engine, plugin=plugin, event_name=event_name, override_route_name=event_info.route
+                    app_engine, plugin=plugin, event_name=event_name, event_info=event_info
                 )
             ])
         elif event_info.type == EventType.MULTIPART:
             web_server.add_routes([
                 _create_multipart_event_route(
-                    app_engine, plugin=plugin, event_name=event_name, override_route_name=event_info.route
+                    app_engine, plugin=plugin, event_name=event_name, event_info=event_info
                 )
             ])
         elif event_info.type == EventType.STREAM and plugin is None:
@@ -282,14 +282,14 @@ def _create_post_event_route(
         app_engine: AppEngine, *,
         plugin: Optional[AppEngine] = None,
         event_name: str,
-        override_route_name: Optional[str]) -> web.RouteDef:
+        event_info: EventDescriptor) -> web.RouteDef:
     """
     Creates route for handling POST event
     """
-    datatype = find_datatype_handler(app_config=app_engine.app_config, event_name=event_name)
+    datatype = find_datatype_handler(app_config=app_engine.app_config, event_name=event_name, event_info=event_info)
     route = app_route_name(app_engine.app_config.app, event_name=event_name,
                            plugin=None if plugin is None else plugin.app_config.app,
-                           override_route_name=override_route_name)
+                           override_route_name=event_info.route)
     logger.info(__name__, f"POST path={route} input={str(datatype)}")
     impl = plugin if plugin else app_engine
     handler = partial(_handle_post_invocation, app_engine, impl,
@@ -304,18 +304,18 @@ def _create_get_event_route(
         app_engine: AppEngine, *,
         plugin: Optional[AppEngine] = None,
         event_name: str,
-        override_route_name: Optional[str]) -> web.RouteDef:
+        event_info: EventDescriptor) -> web.RouteDef:
     """
     Creates route for handling GET requests
     """
     route = app_route_name(app_engine.app_config.app, event_name=event_name,
                            plugin=None if plugin is None else plugin.app_config.app,
-                           override_route_name=override_route_name)
+                           override_route_name=event_info.route)
     logger.info(__name__, f"GET path={route}")
     impl = plugin if plugin else app_engine
     handler = partial(_handle_get_invocation, app_engine, impl, event_name, _auth_types(impl, event_name))
     setattr(handler, '__closure__', None)
-    setattr(handler, '__code__', _handle_post_invocation.__code__)
+    setattr(handler, '__code__', _handle_get_invocation.__code__)
     api_handler = api.add_route('get', route, handler)
     return web.get(route, api_handler)
 
@@ -324,14 +324,14 @@ def _create_multipart_event_route(
         app_engine: AppEngine, *,
         plugin: Optional[AppEngine] = None,
         event_name: str,
-        override_route_name: Optional[str]) -> web.RouteDef:
+        event_info: EventDescriptor) -> web.RouteDef:
     """
     Creates route for handling MULTIPART event
     """
-    datatype = find_datatype_handler(app_config=app_engine.app_config, event_name=event_name)
+    datatype = find_datatype_handler(app_config=app_engine.app_config, event_name=event_name, event_info=event_info)
     route = app_route_name(app_engine.app_config.app, event_name=event_name,
                            plugin=None if plugin is None else plugin.app_config.app,
-                           override_route_name=override_route_name)
+                           override_route_name=event_info.route)
     logger.info(__name__, f"MULTIPART path={route} input={str(datatype)}")
     impl = plugin if plugin else app_engine
     handler = partial(_handle_multipart_invocation, app_engine, impl,
@@ -414,7 +414,7 @@ def _track_ids(request: web.Request) -> Dict[str, str]:
     return {
         'track.operation_id': str(uuid.uuid4()),
         'track.request_id': str(uuid.uuid4()),
-        'track.request_ts': datetime.now().astimezone(timezone.utc).isoformat(),
+        'track.request_ts': datetime.now(tz=timezone.utc).isoformat(),
         **{
             "track." + snakecase(k[8:].lower()): v
             for k, v in request.headers.items() if k.lower().startswith('x-track-')
diff --git a/engine/src/hopeit/streams/__init__.py b/engine/src/hopeit/streams/__init__.py
index 2e5a45b6..2f44c2da 100644
--- a/engine/src/hopeit/streams/__init__.py
+++ b/engine/src/hopeit/streams/__init__.py
@@ -178,7 +178,7 @@ def _consumer_id(self) -> str:
         :return: str, concatenating current UTC ISO datetime, host name, process id
             and this StreamManager instance id
         """
-        ts = datetime.now().astimezone(tz=timezone.utc).isoformat()
+        ts = datetime.now(tz=timezone.utc).isoformat()
         host = socket.gethostname()
         pid = os.getpid()
         mgr = id(self)
diff --git a/plugins/auth/basic-auth/src/hopeit/basic_auth/login.py b/plugins/auth/basic-auth/src/hopeit/basic_auth/login.py
index 2b828ac2..e32bfda9 100644
--- a/plugins/auth/basic-auth/src/hopeit/basic_auth/login.py
+++ b/plugins/auth/basic-auth/src/hopeit/basic_auth/login.py
@@ -33,7 +33,7 @@ async def login(payload: None, context: EventContext) -> AuthInfoExtended:
     Returns a new access and refresh token for a set of given basic-auth credentials
     """
     assert context.auth_info['allowed']
-    now = datetime.now().astimezone(timezone.utc)
+    now = datetime.now(tz=timezone.utc)
     if context.auth_info['auth_type'] == AuthType.BASIC:
         data = base64.b64decode(context.auth_info['payload'].encode()).decode()
         user_info = ContextUserInfo(
diff --git a/plugins/auth/basic-auth/src/hopeit/basic_auth/refresh.py b/plugins/auth/basic-auth/src/hopeit/basic_auth/refresh.py
index 5c0ed05f..897a9125 100644
--- a/plugins/auth/basic-auth/src/hopeit/basic_auth/refresh.py
+++ b/plugins/auth/basic-auth/src/hopeit/basic_auth/refresh.py
@@ -31,7 +31,7 @@ async def refresh(payload: None, context: EventContext) -> AuthInfoExtended:
     Returns a new access and refresh tokens, from a request containing a valid refresh token.
     """
     assert context.auth_info['allowed']
-    now = datetime.now().astimezone(timezone.utc)
+    now = datetime.now(tz=timezone.utc)
     if context.auth_info['auth_type'] == AuthType.REFRESH:
         user_info = ContextUserInfo(
             id=context.auth_info['payload']['id'],
diff --git a/plugins/build/ci-static-plugins.sh b/plugins/build/ci-static-plugins.sh
index c1871cbc..a42d9948 100644
--- a/plugins/build/ci-static-plugins.sh
+++ b/plugins/build/ci-static-plugins.sh
@@ -48,9 +48,9 @@ code+=$?
 echo "storage/fs"
 export MYPYPATH=engine/src/:plugins/storage/fs/src/ && python3 -m mypy --namespace-packages -p hopeit.fs_storage
 code+=$?
-export MYPYPATH=engine/src/:plugins/storage/fs/src/ && python3 -m mypy --namespace-packages plugins/storage/fs/test/unit/
+export MYPYPATH=engine/src/:plugins/storage/fs/src/ && python3 -m mypy --namespace-packages plugins/storage/fs/test/unit/ plugins/storage/fs/test/integration/
 code+=$?
-python3 -m flake8 --max-line-length=120 plugins/storage/fs/src/hopeit/ plugins/storage/fs/test/unit/
+python3 -m flake8 --max-line-length=120 plugins/storage/fs/src/hopeit/ plugins/storage/fs/test/unit/ plugins/storage/fs/test/integration/
 code+=$?
 python3 -m pylint plugins/storage/fs/src/hopeit/fs_storage/
 code+=$?
diff --git a/plugins/ops/apps-visualizer/api/openapi.json b/plugins/ops/apps-visualizer/api/openapi.json
index 3a8ea51d..4217b162 100644
--- a/plugins/ops/apps-visualizer/api/openapi.json
+++ b/plugins/ops/apps-visualizer/api/openapi.json
@@ -1,12 +1,12 @@
 {
   "openapi": "3.0.3",
   "info": {
-    "version": "0.14",
+    "version": "0.15",
     "title": "Simple Example",
     "description": "Simple Example"
   },
   "paths": {
-    "/api/config-manager/0x14/runtime-apps-config": {
+    "/api/config-manager/0x15/runtime-apps-config": {
       "get": {
         "summary": "Config Manager: Runtime Apps Config",
         "description": "Returns the runtime config for the Apps running on this server",
@@ -62,11 +62,11 @@
           }
         },
         "tags": [
-          "config_manager.0x14"
+          "config_manager.0x15"
         ]
       }
     },
-    "/api/config-manager/0x14/cluster-apps-config": {
+    "/api/config-manager/0x15/cluster-apps-config": {
       "get": {
         "summary": "Config Manager: Cluster Apps Config",
         "description": "Handle remote access to runtime configuration for a group of hosts",
@@ -122,7 +122,7 @@
           }
         },
         "tags": [
-          "config_manager.0x14"
+          "config_manager.0x15"
         ]
       }
     },
@@ -209,11 +209,11 @@
           }
         },
         "tags": [
-          "apps_visualizer.0x14"
+          "apps_visualizer.0x15"
         ]
       }
     },
-    "/api/apps-visualizer/0x14/apps/events-graph": {
+    "/api/apps-visualizer/0x15/apps/events-graph": {
       "get": {
         "summary": "App Visualizer: Events Graph Data",
         "description": "App Visualizer: Events Graph Data",
@@ -287,11 +287,11 @@
           }
         },
         "tags": [
-          "apps_visualizer.0x14"
+          "apps_visualizer.0x15"
         ]
       }
     },
-    "/api/apps-visualizer/0x14/event-stats/live": {
+    "/api/apps-visualizer/0x15/event-stats/live": {
       "get": {
         "summary": "App Visualizer: Live Stats",
         "description": "App Visualizer: Live Stats",
@@ -365,7 +365,7 @@
           }
         },
         "tags": [
-          "apps_visualizer.0x14"
+          "apps_visualizer.0x15"
         ]
       }
     }
@@ -649,6 +649,9 @@
           "route": {
             "type": "string"
           },
+          "impl": {
+            "type": "string"
+          },
           "connections": {
             "type": "array",
             "items": {
@@ -683,10 +686,17 @@
               "type": "string"
             },
             "default": []
+          },
+          "dataobjects": {
+            "type": "array",
+            "items": {
+              "type": "string"
+            },
+            "default": []
           }
         },
         "x-module-name": "hopeit.app.config",
-        "description": "\n    Event descriptor\n    "
+        "description": "\n    Event Descriptor: condfigures event implementation\n\n    :field: type, EventType: type of event i.e.: GET, POST, MULTIPART, STREAM, SERVICE\n    :field: plug_mode, EventPlugMode: defines whether an event defined in a plugin is created in the\n        current app (ON_APP) or it will be created in the original plugin (STANDALONE, default)\n    :field: route, optional str: custom route for endpoint. If not specified route will be derived\n        from `/api/app_name/app_version/event_name`\n    :field: impl, optional str: custom event implementation Python module. If not specified, module\n        with same same as event will be imported.\n    :field: connections, list of EventConnection: specifies dependencies on other apps/endpoints,\n        that can be used by client plugins to call events on external apps\n    :field: read_stream, optional ReadStreamDescriptor: specifies source stream to read from.\n        Valid only for STREAM events.\n    :field: write_stream, optional WriteStreamDescriptor: for any type of events, resultant dataobjects will\n        be published to the specified stream.\n    :field: auth, list of AuthType: supported authentication schemas for this event. If not specified\n        application default will be used.\n    :field: setting_keys, list of str: by default EventContext will have access to the settings section\n        with the same name of the event using `settings = context.settings(datatype=MySettingsType)`.\n        In case additional sections are needed to be accessed from\n        EventContext, then a list of setting keys, including the name of the event if needed,\n        can be specified here. Then access to a `custom` key can be done using\n        `custom_settings = context.settings(key=\"customer\", datatype=MyCustomSettingsType)`\n    :field: dataobjects, list of str: list of full qualified dataobject types that this event can process.\n        When not specified, the engine will inspect the module implementation and find all datatypes supported\n        as payload in the functions defined as `__steps__`. In case of generic functions that support\n        `payload: DataObject` argument, then a list of full qualified datatypes must be specified here.\n    "
       },
       "EventConnection": {
         "type": "object",
@@ -813,7 +823,7 @@
           },
           "engine_version": {
             "type": "string",
-            "default": "0.14.2"
+            "default": "0.15.0"
           }
         },
         "x-module-name": "hopeit.server.config",
diff --git a/plugins/ops/log-streamer/src/hopeit/log_streamer/__init__.py b/plugins/ops/log-streamer/src/hopeit/log_streamer/__init__.py
index 8a9bf91b..05af6987 100644
--- a/plugins/ops/log-streamer/src/hopeit/log_streamer/__init__.py
+++ b/plugins/ops/log-streamer/src/hopeit/log_streamer/__init__.py
@@ -3,7 +3,7 @@
 """
 import asyncio
 from typing import Dict, List, TextIO
-from datetime import datetime
+from datetime import datetime, timezone
 
 from watchdog.observers import Observer  # type: ignore
 from watchdog.events import FileSystemEventHandler  # type: ignore
@@ -151,7 +151,7 @@ async def _save_checkpoint(self, src_path: str, line: str):
             key = f'{self.context.app_key}.{src_path}.checkpoint'.replace('/', 'x')
             cp = Checkpoint(
                 line=line,
-                expire=int(datetime.now().timestamp()) + self.file_checkpoint_expire
+                expire=int(datetime.now(tz=timezone.utc).timestamp()) + self.file_checkpoint_expire
             )
             await self.checkpoint_storage.store(key, cp)
 
@@ -160,7 +160,7 @@ async def _load_checkpoint(self, src_path: str) -> str:
         cp = await self.checkpoint_storage.get(key, datatype=Checkpoint)
         if cp is None:
             return ''
-        if int(cp.expire) < int(datetime.now().timestamp()):
+        if int(cp.expire) < int(datetime.now(tz=timezone.utc).timestamp()):
             return ''
         return cp.line
 
@@ -171,7 +171,7 @@ async def _open_file(self, src_path: str) -> bool:
         """
         try:
             async with self.lock:
-                self.last_access[src_path] = datetime.now().timestamp()
+                self.last_access[src_path] = datetime.now(tz=timezone.utc).timestamp()
                 if self.open_files.get(src_path) is None:
                     checkpoint = await self._load_checkpoint(src_path)
                     logger.info(self.context, "Opening log file...", extra=extra(
@@ -204,7 +204,7 @@ def close_inactive_files(self):
         """
         Closes files that are inactive (deleted of with no activity)
         """
-        exp = datetime.now().timestamp()
+        exp = datetime.now(tz=timezone.utc).timestamp()
         for key, last_ts in list(self.last_access.items()):
             if (last_ts + self.file_open_timeout) < exp:
                 try:
diff --git a/plugins/storage/fs/src/hopeit/fs_storage/__init__.py b/plugins/storage/fs/src/hopeit/fs_storage/__init__.py
index c18f121a..8328dc3c 100644
--- a/plugins/storage/fs/src/hopeit/fs_storage/__init__.py
+++ b/plugins/storage/fs/src/hopeit/fs_storage/__init__.py
@@ -2,6 +2,7 @@
 Storage/persistence asynchronous stores and gets dataobjects from filesystem.
 
 """
+from dataclasses import dataclass
 import os
 from glob import glob
 from pathlib import Path
@@ -10,46 +11,105 @@
 
 import aiofiles  # type: ignore
 
-from hopeit.dataobjects import DataObject
+from hopeit.dataobjects import dataobject, DataObject
 from hopeit.dataobjects.payload import Payload
+from hopeit.fs_storage.partition import get_partition_key
 
-__all__ = ['FileStorage']
+__all__ = ['FileStorage',
+           'FileStorageSettings']
 
 SUFFIX = '.json'
 
 
+@dataobject
+@dataclass
+class FileStorageSettings:
+    """
+    File storage plugin config
+
+    :field: path, str: base path in file systems here data is saved
+    :field: partition_dateformat, optional str: date format to be used to prefix file name in order
+        to parition saved files to different subfolders based on event_ts(). i.e. "%Y/%m/%d"
+        will store each files in a folder `base_path/year/month/day/`
+    :field: flush_seconds, float: number of seconds to trigger a flush event to save all current
+        buffered partitions. Default 0 means flish is not triggered by time.
+    :field: fllush_max_size: max number of elements to keep in a partition before forcing a flush.
+        Default 1. A value of 0 will disable flushing by partition size.
+    """
+    path: str
+    partition_dateformat: Optional[str] = None
+    flush_seconds: float = 0.0
+    flush_max_size: int = 1
+
+
+@dataobject
+@dataclass
+class ItemLocator:
+    item_id: str
+    partition_key: Optional[str] = None
+
+
 class FileStorage(Generic[DataObject]):
     """
         Stores and retrieves dataobjects from filesystem
     """
-    def __init__(self, *, path: str):
+    def __init__(self, *, path: str, partition_dateformat: Optional[str] = None):
         """
         Setups a file storage
 
         :param path: str, base path to be used to store and retrieve objects
         """
         self.path: Path = Path(path)
+        os.makedirs(self.path.resolve().as_posix(), exist_ok=True)
+        self.partition_dateformat = (partition_dateformat or '').strip('/')
+
+    @classmethod
+    def with_settings(cls, settings: FileStorageSettings) -> "FileStorage":
+        return cls(
+            path=settings.path,
+            partition_dateformat=settings.partition_dateformat
+        )
 
-    async def list_objects(self, wildcard: str = '*') -> List[str]:
+    async def list_objects(self, wildcard: str = '*') -> List[ItemLocator]:
         """
         Retrieves list of objects keys from the file storage
 
         :param wilcard: allow filter the listing of objects
         :return: List of objects key
         """
-
-        path = str(self.path.absolute()) + '/' + wildcard + SUFFIX
-        return [Path(path).name[:-len(SUFFIX)] for path in glob(path)]
-
-    async def get(self, key: str, *, datatype: Type[DataObject]) -> Optional[DataObject]:
+        base_path = str(self.path.resolve())
+        path = base_path + '/' + wildcard + SUFFIX
+        n_part_comps = len(self.partition_dateformat.split('/'))
+        return [
+            self._get_item_locator(item_path, n_part_comps) for item_path in glob(path)
+        ]
+
+    def _get_item_locator(self, item_path: str, n_part_comps: int) -> ItemLocator:
+        comps = item_path.split("/")
+        if self.partition_dateformat:
+            partition_key = '/'.join(comps[-n_part_comps - 1:-1])
+        else:
+            partition_key = None
+        return ItemLocator(
+            item_id=comps[-1][:-len(SUFFIX)],
+            partition_key=partition_key
+        )
+
+    async def get(
+        self, key: str,
+        *, datatype: Type[DataObject],
+        partition_key: Optional[str] = None
+    ) -> Optional[DataObject]:
         """
         Retrieves value under specified key, converted to datatype
 
         :param key: str
         :param datatype: dataclass implementing @dataobject (@see DataObject)
+        :param parition_key: partition path to be appended to base path
         :return: instance of datatype or None if not found
         """
-        payload_str = await self._load_file(path=self.path, file_name=key + SUFFIX)
+        path = self.path / partition_key if partition_key else self.path
+        payload_str = await self._load_file(path=path, file_name=key + SUFFIX)
         if payload_str:
             return Payload.from_json(payload_str, datatype)
         return None
@@ -63,7 +123,11 @@ async def store(self, key: str, value: DataObject) -> str:
         :return: str, path where the object was stored
         """
         payload_str = Payload.to_json(value)
-        return await self._save_file(payload_str, path=self.path, file_name=key + SUFFIX)
+        path = self.path
+        if self.partition_dateformat:
+            path = path / get_partition_key(value, self.partition_dateformat)
+            os.makedirs(path.resolve().as_posix(), exist_ok=True)
+        return await self._save_file(payload_str, path=path, file_name=key + SUFFIX)
 
     @staticmethod
     async def _load_file(*, path: Path, file_name: str) -> Optional[str]:
@@ -88,7 +152,6 @@ async def _save_file(payload_str: str, *, path: Path, file_name: str) -> str:
         """
         file_path = path / file_name
         tmp_path = path / str(f".{uuid.uuid4()}")
-        os.makedirs(str(path), exist_ok=True)
         async with aiofiles.open(tmp_path, 'w') as f:  # type: ignore
             await f.write(payload_str)
             await f.flush()
diff --git a/plugins/storage/fs/src/hopeit/fs_storage/events/__init__.py b/plugins/storage/fs/src/hopeit/fs_storage/events/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/plugins/storage/fs/src/hopeit/fs_storage/events/py.typed b/plugins/storage/fs/src/hopeit/fs_storage/events/py.typed
new file mode 100644
index 00000000..f5642f79
--- /dev/null
+++ b/plugins/storage/fs/src/hopeit/fs_storage/events/py.typed
@@ -0,0 +1,1 @@
+Marker
diff --git a/plugins/storage/fs/src/hopeit/fs_storage/events/stream_batch_storage.py b/plugins/storage/fs/src/hopeit/fs_storage/events/stream_batch_storage.py
new file mode 100644
index 00000000..deae1aab
--- /dev/null
+++ b/plugins/storage/fs/src/hopeit/fs_storage/events/stream_batch_storage.py
@@ -0,0 +1,171 @@
+"""
+Stream Batch Storage
+
+This event implementation allows saving data from a stream to the file system.
+This event acts as a STREAM event and at the same time as a SERVICE,
+managed internally by hopeit.engine once configured (see example below).
+
+The STREAM event, consumed elements from a stream and keeps them in a memory buffer.
+The SERVICE part provides a loop to flush the buffer in a fixed time interval.
+
+Example app config:
+```
+{
+  ...
+  "settings": {
+    "storage.save_events_fs": {
+      "path": "/tmp/{auto}",
+      "partition_dateformat": "%Y/%m/%d/%H/",
+      "flush_seconds": 60.0,
+      "flush_max_size": 100
+    }
+  },
+  "events": {
+    "storage.save_events_fs": {
+      "type": "STREAM",
+      "read_stream": {
+        "name": "data_stream",
+        "consumer_group": "{auto}",
+      },
+      "impl": "hopeit.fs_storage.events.stream_batch_storage",
+      "dataobjects": [
+        "model.Something"
+      ]
+    }
+  }
+```
+
+This implementation will buffer a number of events for a given time, divided into partitions.
+Partitions are determined by the `event_ts()` returning function of a dataobject, i.e.:
+```
+@dataobject(event_ts='date_field')
+@dataclass
+class Something:
+    ...
+    date_field: datetime
+    ...
+```
+
+If event_ts is not set or date field is None, current timestamp at the moment the event
+is consumed in UTC will be used. In case event_ts is defined, it will be converted to UTC
+before used for partitioning. In case datetime field is naive (no timezone) it will
+be assumed to be in local timezone before converstion to UTC (Python `astimezone()` implementation)
+
+The maximum elements per partition to be kept in memory are set by the `flush_max_size` setting.
+The approximate regular interval to flush all partitions in buffer is set by `flush_seconds` setting.
+When flushing buffered items, this event implementation will create a folder
+per each partition, using the `partition_dateformat` string to create subfolders x time intervals:
+
+i.e. for a daily format `%Y/%m/%d` (default):
+```
+/save_path/
+ |--2022
+     |-- 03
+          |--01
+          |  |-- 55d6f2d4-865d-47a5-8cd3-bf04f7ac07f5.jsonlines
+          |  |-- 6d926417-6b16-49ad-86bd-897cbbeaa614.jsonlines
+          |  ...
+          |--02
+             ...
+```
+
+This way data can be filtered out on retrieval without the need to iterate
+all folders.
+
+Each generated files is a in `jsonlines` format (http://jsonlines.org) where
+each line is a valid single-line json object resulting of serializing the
+dataobjects consumed from the input stream.
+"""
+import asyncio
+import os
+import uuid
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Dict, List, Optional
+
+import aiofiles
+from hopeit.app.context import EventContext
+from hopeit.app.events import Spawn
+from hopeit.app.logger import app_extra_logger
+from hopeit.dataobjects import DataObject, dataobject
+from hopeit.dataobjects.payload import Payload
+from hopeit.fs_storage import FileStorageSettings
+from hopeit.fs_storage.partition import get_partition_key
+
+logger, extra = app_extra_logger()
+
+__steps__ = ['buffer_item', 'flush']
+
+
+@dataclass
+class Partition:
+    lock: asyncio.Lock = field(default_factory=asyncio.Lock)
+    items: List[DataObject] = field(default_factory=list)  # type: ignore
+
+
+SUFFIX = '.jsonlines'
+buffer: Dict[str, Partition] = {}
+buffer_lock: asyncio.Lock = asyncio.Lock()
+
+
+@dataobject
+@dataclass
+class FlushSignal:
+    partition_key: str
+
+
+async def __service__(context: EventContext) -> Spawn[FlushSignal]:
+    settings: FileStorageSettings = context.settings(datatype=FileStorageSettings)
+    if settings.flush_seconds:
+        while True:
+            await asyncio.sleep(settings.flush_seconds)
+            for partition_key in list(buffer.keys()):
+                yield FlushSignal(partition_key=partition_key)
+    else:
+        if settings.flush_max_size == 0:
+            logger.warning(
+                context,
+                "Flushing partitions by size and time are disabled."
+                "Specify either `flush_seconds` or `flush_max_size`"
+                "to enable flushing the buffer periodically"
+            )
+        else:
+            logger.info(context, "Flushing partitions by time disabled.")
+
+
+async def buffer_item(payload: DataObject, context: EventContext) -> Optional[FlushSignal]:
+    """
+    Consumes any Dataobject type from stream and put it local memory buffer to be flushed later
+    """
+    settings: FileStorageSettings = context.settings(datatype=FileStorageSettings)
+    partition_key = get_partition_key(payload, settings.partition_dateformat or "")
+    async with buffer_lock:
+        partition = buffer.get(partition_key, Partition())
+        buffer[partition_key] = partition
+    async with partition.lock:
+        partition.items.append(payload)
+    if settings.flush_max_size and len(partition.items) >= settings.flush_max_size:
+        return FlushSignal(partition_key=partition_key)
+    return None
+
+
+async def flush(signal: FlushSignal, context: EventContext):
+    logger.info(context, f"Flushing partition {signal.partition_key}...")
+    partition = buffer[signal.partition_key]
+    async with partition.lock:
+        if len(partition.items):
+            await _save_partition(signal.partition_key, partition.items, context)
+        async with buffer_lock:
+            del buffer[signal.partition_key]
+    logger.info(context, f"Flush {signal.partition_key} done.")
+
+
+async def _save_partition(partition_key: str, items: List[DataObject], context: EventContext):
+    settings = context.settings(datatype=FileStorageSettings)
+    path = Path(settings.path) / partition_key
+    file = path / f"{uuid.uuid4()}{SUFFIX}"
+    logger.info(context, f"Saving {file}...")
+    os.makedirs(path.resolve(), exist_ok=True)
+    async with aiofiles.open(file, 'w') as f:
+        for item in items:
+            await f.write(Payload.to_json(item) + "\n")
diff --git a/plugins/storage/fs/src/hopeit/fs_storage/partition.py b/plugins/storage/fs/src/hopeit/fs_storage/partition.py
new file mode 100644
index 00000000..dfbb84c9
--- /dev/null
+++ b/plugins/storage/fs/src/hopeit/fs_storage/partition.py
@@ -0,0 +1,16 @@
+"""
+FS Storage plugin package module
+"""
+from datetime import datetime, timezone
+
+from hopeit.dataobjects import DataObject
+
+
+def get_partition_key(payload: DataObject, partition_dateformat: str) -> str:
+    ts = _partition_timestamp(payload)
+    return ts.strftime(partition_dateformat.strip('/')) + '/'
+
+
+def _partition_timestamp(payload: DataObject) -> datetime:
+    ts = payload.event_ts() or datetime.now(tz=timezone.utc)  # type: ignore
+    return ts.astimezone(timezone.utc)
diff --git a/plugins/streams/redis/src/hopeit/redis_streams/__init__.py b/plugins/streams/redis/src/hopeit/redis_streams/__init__.py
index 2b810b17..10a3e707 100644
--- a/plugins/streams/redis/src/hopeit/redis_streams/__init__.py
+++ b/plugins/streams/redis/src/hopeit/redis_streams/__init__.py
@@ -166,7 +166,7 @@ async def read_stream(self, *,
                                          batch_size=len(batch), head=batch[0][0], tail=batch[-1][0]))
                 stream_events: List[Union[StreamEvent, Exception]] = []
                 for msg in batch:
-                    read_ts = datetime.now().astimezone(tz=timezone.utc).isoformat()
+                    read_ts = datetime.now(tz=timezone.utc).isoformat()
                     msg_type = msg[1][b'type'].decode()
                     datatype = datatypes.get(msg_type)
                     if datatype is None:
@@ -218,10 +218,11 @@ def _encode_message(self, payload: EventPayload,
             :event_ts: extracted from payload.event_ts() if defined, if not empty string
             :payload: json serialized payload
         """
+        datatype = type(payload)
         event_fields = {
             'id': payload.event_id(),  # type: ignore
-            'type': type(payload).__name__,
-            'submit_ts': datetime.now().astimezone(tz=timezone.utc).isoformat(),
+            'type': f"{datatype.__module__}.{datatype.__qualname__}",
+            'submit_ts': datetime.now(tz=timezone.utc).isoformat(),
             'event_ts': '',
             **{k: v or '' for k, v in track_ids.items()},
             'auth_info': base64.b64encode(json.dumps(auth_info).encode()),


Test patch:
diff --git a/apps/benchmark/simple-benchmark/test/benchmark-docker.sh b/apps/benchmark/simple-benchmark/test/benchmark-docker.sh
index ed468ff8..cd7d5602 100755
--- a/apps/benchmark/simple-benchmark/test/benchmark-docker.sh
+++ b/apps/benchmark/simple-benchmark/test/benchmark-docker.sh
@@ -1,18 +1,18 @@
 #!/usr/bin/env bash
-echo ./wrk2/wrk -t8 -c40 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x14/give-me-something?item_id=string"
-./wrk2/wrk -t8 -c40 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x14/give-me-something?item_id=string"
+echo ./wrk2/wrk -t8 -c40 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x15/give-me-something?item_id=string"
+./wrk2/wrk -t8 -c40 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x15/give-me-something?item_id=string"
 sleep 5
-echo ./wrk2/wrk -t6 -c32 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x14/query-something-redis?item_id=string"
-./wrk2/wrk -t6 -c32 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x14/query-something-redis?item_id=string"
+echo ./wrk2/wrk -t6 -c32 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x15/query-something-redis?item_id=string"
+./wrk2/wrk -t6 -c32 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x15/query-something-redis?item_id=string"
 sleep 5
-echo ./wrk2/wrk -t4 -c20 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x14/query-something-fs?item_id=string"
-./wrk2/wrk -t4 -c20 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x14/query-something-fs?item_id=string"
+echo ./wrk2/wrk -t4 -c20 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x15/query-something-fs?item_id=string"
+./wrk2/wrk -t4 -c20 -d60s -R4000 --u_latency "http://localhost:8025/api/simple-benchmark/0x15/query-something-fs?item_id=string"
 sleep 5
-echo ./wrk2/wrk -t8 -c34 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x14/give-me-something?item_id=string"
-./wrk2/wrk -t8 -c34 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x14/give-me-something?item_id=string"
+echo ./wrk2/wrk -t8 -c34 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x15/give-me-something?item_id=string"
+./wrk2/wrk -t8 -c34 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x15/give-me-something?item_id=string"
 sleep 5
-echo ./wrk2/wrk -t6 -c24 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x14/query-something-redis?item_id=string"
-./wrk2/wrk -t6 -c24 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x14/query-something-redis?item_id=string"
+echo ./wrk2/wrk -t6 -c24 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x15/query-something-redis?item_id=string"
+./wrk2/wrk -t6 -c24 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x15/query-something-redis?item_id=string"
 sleep 5
-echo ./wrk2/wrk -t4 -c14 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x14/query-something-fs?item_id=string"
-./wrk2/wrk -t4 -c14 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x14/query-something-fs?item_id=string"
+echo ./wrk2/wrk -t4 -c14 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x15/query-something-fs?item_id=string"
+./wrk2/wrk -t4 -c14 -d60s -R4000 --u_latency "http://localhost:8021/api/simple-benchmark/0x15/query-something-fs?item_id=string"
diff --git a/apps/benchmark/simple-benchmark/test/warm-up-fs.sh b/apps/benchmark/simple-benchmark/test/warm-up-fs.sh
index 2f0627e9..36634bd6 100755
--- a/apps/benchmark/simple-benchmark/test/warm-up-fs.sh
+++ b/apps/benchmark/simple-benchmark/test/warm-up-fs.sh
@@ -7,6 +7,6 @@ else
   host=localhost
 fi
 for i in {1..2000}; do
-  curl --silent --output /dev/null -d '{"id": "string", "user": "string"}' -H 'Content-Type: application/json' "http://$host:8021/api/simple-benchmark/0x14/save-something-fs"
+  curl --silent --output /dev/null -d '{"id": "string", "user": "string"}' -H 'Content-Type: application/json' "http://$host:8021/api/simple-benchmark/0x15/save-something-fs"
 done
 echo Done.
diff --git a/apps/examples/simple-example/test/integration/conftest.py b/apps/examples/simple-example/test/integration/conftest.py
index f4b35cb5..037593ac 100644
--- a/apps/examples/simple-example/test/integration/conftest.py
+++ b/apps/examples/simple-example/test/integration/conftest.py
@@ -1,10 +1,12 @@
-from datetime import datetime
+from datetime import datetime, timezone
 
 import pytest  # type: ignore
 from hopeit.testing.apps import config
 
 from model import Something, User, Status, StatusType, SomethingParams
 
+EVENT_TS = datetime.strptime("2020-05-01T00:00:00", "%Y-%m-%dT%H:%M:%S").astimezone(tz=timezone.utc)
+
 
 @pytest.fixture
 def something_example():
@@ -32,7 +34,7 @@ def something_submitted():
     return Something(
         id="test",
         user=User(id="u1", name="test_user"),
-        status=Status(ts=datetime.now(), type=StatusType.SUBMITTED)
+        status=Status(ts=EVENT_TS, type=StatusType.SUBMITTED)
     )
 
 
@@ -41,8 +43,8 @@ def something_processed():
     return Something(
         id="test",
         user=User(id="u1", name="test_user"),
-        status=Status(ts=datetime.now(), type=StatusType.PROCESSED),
-        history=[Status(ts=datetime.now(), type=StatusType.SUBMITTED)]
+        status=Status(ts=EVENT_TS, type=StatusType.PROCESSED),
+        history=[Status(ts=EVENT_TS, type=StatusType.SUBMITTED)]
     )
 
 
@@ -51,7 +53,7 @@ def something_with_status_example():
     return Something(
         id="test",
         user=User(id="u1", name="test_user"),
-        status=Status(ts=datetime.now(), type=StatusType.NEW)
+        status=Status(ts=EVENT_TS, type=StatusType.NEW)
     )
 
 
@@ -60,8 +62,8 @@ def something_with_status_submitted_example():
     return Something(
         id="test",
         user=User(id="u1", name="test_user"),
-        status=Status(ts=datetime.now(), type=StatusType.SUBMITTED),
-        history=[Status(ts=datetime.now(), type=StatusType.NEW)]
+        status=Status(ts=EVENT_TS, type=StatusType.SUBMITTED),
+        history=[Status(ts=EVENT_TS, type=StatusType.NEW)]
     )
 
 
@@ -70,9 +72,9 @@ def something_with_status_processed_example():
     return Something(
         id="test",
         user=User(id="u1", name="test_user"),
-        status=Status(ts=datetime.now(), type=StatusType.PROCESSED),
-        history=[Status(ts=datetime.now(), type=StatusType.NEW),
-                 Status(ts=datetime.now(), type=StatusType.SUBMITTED)]
+        status=Status(ts=EVENT_TS, type=StatusType.PROCESSED),
+        history=[Status(ts=EVENT_TS, type=StatusType.NEW),
+                 Status(ts=EVENT_TS, type=StatusType.SUBMITTED)]
     )
 
 
diff --git a/apps/examples/simple-example/test/integration/test_it_collect_spawn.py b/apps/examples/simple-example/test/integration/test_it_collect_spawn.py
index 5b485cb1..d8041b5e 100644
--- a/apps/examples/simple-example/test/integration/test_it_collect_spawn.py
+++ b/apps/examples/simple-example/test/integration/test_it_collect_spawn.py
@@ -18,8 +18,8 @@ def sample_file_ids():
     for test_id in ids:
         json_str = '{"id": "' + test_id + '", "user": {"id": "u1", "name": "test_user"}, ' \
                    + '"status": {"ts": "2020-05-01T00:00:00Z", "type": "NEW"}, "history": []}'
-        os.makedirs(f'/tmp/simple_example.{APP_VERSION}.fs.data_path/', exist_ok=True)
-        with open(f'/tmp/simple_example.{APP_VERSION}.fs.data_path/{test_id}.json', 'w') as f:
+        os.makedirs(f'/tmp/hopeit/simple_example.{APP_VERSION}.fs_storage.path/2020/05/01/00/', exist_ok=True)
+        with open(f'/tmp/hopeit/simple_example.{APP_VERSION}.fs_storage.path/2020/05/01/00/{test_id}.json', 'w') as f:
             f.write(json_str)
             f.flush()
     return ids
@@ -28,6 +28,7 @@ def sample_file_ids():
 @pytest.mark.asyncio
 async def test_find_two_items(app_config, sample_file_ids):  # noqa: F811
     payload = ItemsInfo(*sample_file_ids)
+    payload.partition_key = "2020/05/01/00/"
     result, pp_result, response = await execute_event(app_config=app_config,
                                                       event_name='collector.collect_spawn',
                                                       payload=payload,
@@ -42,6 +43,7 @@ async def test_find_two_items(app_config, sample_file_ids):  # noqa: F811
 @pytest.mark.asyncio
 async def test_find_one_item(app_config, sample_file_ids):  # noqa: F811
     payload = ItemsInfo(sample_file_ids[0], str(uuid.uuid4()))
+    payload.partition_key = "2020/05/01/00/"
     result, pp_result, response = await execute_event(app_config=app_config,
                                                       event_name='collector.collect_spawn',
                                                       payload=payload,
diff --git a/apps/examples/simple-example/test/integration/test_it_download_something.py b/apps/examples/simple-example/test/integration/test_it_download_something.py
index 43fc14c6..504bb43c 100644
--- a/apps/examples/simple-example/test/integration/test_it_download_something.py
+++ b/apps/examples/simple-example/test/integration/test_it_download_something.py
@@ -25,6 +25,6 @@ async def test_it_download_something(app_config):
         'Content-Disposition': f'attachment; filename="{file_name}"'
     }
     assert response.file_response == (
-        f"/tmp/simple_example.{APP_VERSION}.fs.data_path/{file_name}"
+        f"/tmp/hopeit/{file_name}"
     )
     assert response.content_type == 'image/png'
diff --git a/apps/examples/simple-example/test/integration/test_it_list_somethings.py b/apps/examples/simple-example/test/integration/test_it_list_somethings.py
index f6e9a0eb..356499e5 100644
--- a/apps/examples/simple-example/test/integration/test_it_list_somethings.py
+++ b/apps/examples/simple-example/test/integration/test_it_list_somethings.py
@@ -20,11 +20,11 @@ def sample_file_id():
                 + '"status": {"ts": "2020-05-01T00:00:00Z", "type": "NEW"}, "history": []}'
     json_str2 = '{"id": "' + test_id2 + '", "user": {"id": "u1", "name": "test_user"}, ' \
                 + '"status": {"ts": "2020-05-01T00:00:00Z", "type": "NEW"}, "history": []}'
-    os.makedirs(f'/tmp/simple_example.{APP_VERSION}.fs.data_path/', exist_ok=True)
-    with open(f'/tmp/simple_example.{APP_VERSION}.fs.data_path/{test_id1}.json', 'w') as f:
+    os.makedirs(f'/tmp/hopeit/simple_example.{APP_VERSION}.fs_storage.path/2020/05/01/00/', exist_ok=True)
+    with open(f'/tmp/hopeit/simple_example.{APP_VERSION}.fs_storage.path/2020/05/01/00/{test_id1}.json', 'w') as f:
         f.write(json_str1)
         f.flush()
-    with open(f'/tmp/simple_example.{APP_VERSION}.fs.data_path/{test_id2}.json', 'w') as f:
+    with open(f'/tmp/hopeit/simple_example.{APP_VERSION}.fs_storage.path/2020/05/01/00/{test_id2}.json', 'w') as f:
         f.write(json_str2)
         f.flush()
     return test_id
@@ -35,7 +35,7 @@ async def test_query_item(app_config, sample_file_id):  # noqa: F811
     results = await execute_event(app_config=app_config,
                                   event_name='list_somethings',
                                   payload=None,
-                                  wildcard=sample_file_id + '*')
+                                  wildcard=f"2020/05/01/00/{sample_file_id}*")
     assert len(results) == 2
     assert all(result.id.startswith(sample_file_id) for result in results)
     assert all(isinstance(result, Something) for result in results)
diff --git a/apps/examples/simple-example/test/integration/test_it_parallelize_event.py b/apps/examples/simple-example/test/integration/test_it_parallelize_event.py
index d4387492..632882b5 100644
--- a/apps/examples/simple-example/test/integration/test_it_parallelize_event.py
+++ b/apps/examples/simple-example/test/integration/test_it_parallelize_event.py
@@ -20,9 +20,9 @@ async def test_parallelize_event(monkeypatch, app_config,  # noqa: F811
         expected.status.ts = results[i].payload.status.ts
         for j in range(len(expected.history)):
             expected.history[j].ts = results[i].payload.history[j].ts
-    assert results[0].path == f"{app_config.env['fs']['data_path']}{first.id}.json"
+    assert results[0].path.split('/')[-1] == f"{first.id}.json"
     assert results[0].payload == first
-    assert results[1].path == f"{app_config.env['fs']['data_path']}{second.id}.json"
+    assert results[1].path.split('/')[-1] == f"{second.id}.json"
     assert results[1].payload == second
     assert msg == f"events submitted to stream: {app_config.events['shuffle.parallelize_event'].write_stream.name}"
     assert response.headers.get("X-Stream-Name") == app_config.events['shuffle.parallelize_event'].write_stream.name
diff --git a/apps/examples/simple-example/test/integration/test_it_process_events.py b/apps/examples/simple-example/test/integration/test_it_process_events.py
index ef1ec53f..481eafd0 100644
--- a/apps/examples/simple-example/test/integration/test_it_process_events.py
+++ b/apps/examples/simple-example/test/integration/test_it_process_events.py
@@ -1,6 +1,7 @@
 import pytest  # type: ignore
 
 from hopeit.testing.apps import execute_event
+from hopeit.fs_storage.partition import get_partition_key
 from simple_example.streams.process_events import SomethingStored
 
 
@@ -12,7 +13,11 @@ async def test_it_process_events(app_config,  # noqa: F811
                                                   payload=something_submitted)
     something_processed.status.ts = result.payload.status.ts
     something_processed.history[-1].ts = result.payload.history[-1].ts
+    partition_key = get_partition_key(something_processed, partition_dateformat="%Y/%m/%d/%H")
     assert result == SomethingStored(
-            path=f"{app_config.env['fs']['data_path']}{result.payload.id}.json",
+            path=(
+                f"{app_config.env['storage']['base_path']}simple_example.0x15.fs_storage.path/"
+                f"{partition_key}{result.payload.id}.json"
+            ),
             payload=something_processed
         )
diff --git a/apps/examples/simple-example/test/integration/test_it_query_concurrently.py b/apps/examples/simple-example/test/integration/test_it_query_concurrently.py
index 70841191..ef2ccd56 100644
--- a/apps/examples/simple-example/test/integration/test_it_query_concurrently.py
+++ b/apps/examples/simple-example/test/integration/test_it_query_concurrently.py
@@ -17,8 +17,8 @@ def sample_file_ids():
     for test_id in ids:
         json_str = '{"id": "' + test_id + '", "user": {"id": "u1", "name": "test_user"}, ' \
                    + '"status": {"ts": "2020-05-01T00:00:00Z", "type": "NEW"}, "history": []}'
-        os.makedirs(f'/tmp/simple_example.{APP_VERSION}.fs.data_path/', exist_ok=True)
-        with open(f'/tmp/simple_example.{APP_VERSION}.fs.data_path/{test_id}.json', 'w') as f:
+        os.makedirs(f'/tmp/hopeit/simple_example.{APP_VERSION}.fs_storage.path/2020/05/01/00/', exist_ok=True)
+        with open(f'/tmp/hopeit/simple_example.{APP_VERSION}.fs_storage.path/2020/05/01/00/{test_id}.json', 'w') as f:
             f.write(json_str)
             f.flush()
     return ids
@@ -27,6 +27,7 @@ def sample_file_ids():
 @pytest.mark.asyncio
 async def test_find_two_items(app_config, sample_file_ids):  # noqa: F811
     payload = ItemsInfo(*sample_file_ids)
+    payload.partition_key = "2020/05/01/00/"
     result = await execute_event(app_config=app_config,
                                  event_name='collector.query_concurrently',
                                  payload=payload)
@@ -39,6 +40,7 @@ async def test_find_two_items(app_config, sample_file_ids):  # noqa: F811
 @pytest.mark.asyncio
 async def test_find_one_item(app_config, sample_file_ids):  # noqa: F811
     payload = ItemsInfo(sample_file_ids[0], str(uuid.uuid4()))
+    payload.partition_key = "2020/05/01/00/"
     result = await execute_event(app_config=app_config,
                                  event_name='collector.query_concurrently',
                                  payload=payload)
@@ -50,6 +52,7 @@ async def test_find_one_item(app_config, sample_file_ids):  # noqa: F811
 @pytest.mark.asyncio
 async def test_find_no_items(app_config, sample_file_ids):  # noqa: F811
     payload = ItemsInfo(str(uuid.uuid4()), str(uuid.uuid4()))
+    payload.partition_key = "2020/05/01/00/"
     result = await execute_event(app_config=app_config,
                                  event_name='collector.query_concurrently',
                                  payload=payload)
diff --git a/apps/examples/simple-example/test/integration/test_it_query_something.py b/apps/examples/simple-example/test/integration/test_it_query_something.py
index 667f3693..bd0f193b 100644
--- a/apps/examples/simple-example/test/integration/test_it_query_something.py
+++ b/apps/examples/simple-example/test/integration/test_it_query_something.py
@@ -16,11 +16,11 @@ def sample_file_id():
     test_id = str(uuid.uuid4())
     json_str = '{"id": "' + test_id + '", "user": {"id": "u1", "name": "test_user"}, ' \
                + '"status": {"ts": "2020-05-01T00:00:00Z", "type": "NEW"}, "history": []}'
-    os.makedirs(f'/tmp/simple_example.{APP_VERSION}.fs.data_path/', exist_ok=True)
-    with open(f'/tmp/simple_example.{APP_VERSION}.fs.data_path/{test_id}.json', 'w') as f:
+    os.makedirs(f'/tmp/hopeit/simple_example.{APP_VERSION}.fs_storage.path/2020/05/01/00/', exist_ok=True)
+    with open(f'/tmp/hopeit/simple_example.{APP_VERSION}.fs_storage.path/2020/05/01/00/{test_id}.json', 'w') as f:
         f.write(json_str)
         f.flush()
-    return test_id
+    return test_id, "2020/05/01/00"
 
 
 @pytest.mark.asyncio
@@ -29,10 +29,11 @@ async def test_query_item(app_config, sample_file_id):  # noqa: F811
                                                  event_name='query_something',
                                                  payload=None,
                                                  postprocess=True,
-                                                 item_id=sample_file_id)
+                                                 item_id=sample_file_id[0],
+                                                 partition_key=sample_file_id[1])
     assert isinstance(result, Something)
     assert result == pp_result
-    assert result.id == sample_file_id
+    assert result.id == sample_file_id[0]
 
 
 @pytest.mark.asyncio
@@ -42,8 +43,9 @@ async def test_query_item_not_found(app_config):  # noqa: F811
                                                  event_name='query_something',
                                                  payload=None,
                                                  postprocess=True,
-                                                 item_id=item_id)
+                                                 item_id=item_id,
+                                                 partition_key="x")
     assert res.status == 404
     assert result == pp_result
     assert result == SomethingNotFound(
-        path=f'/tmp/simple_example.{APP_VERSION}.fs.data_path', id=item_id)
+        path=f'/tmp/hopeit/simple_example.{APP_VERSION}.fs_storage.path/x', id=item_id)
diff --git a/apps/examples/simple-example/test/integration/test_it_query_something_extended.py b/apps/examples/simple-example/test/integration/test_it_query_something_extended.py
index 7eda39c6..56dad37e 100644
--- a/apps/examples/simple-example/test/integration/test_it_query_something_extended.py
+++ b/apps/examples/simple-example/test/integration/test_it_query_something_extended.py
@@ -17,11 +17,11 @@ def sample_file_id():
     test_id = str(uuid.uuid4())
     json_str = '{"id": "' + test_id + '", "user": {"id": "u1", "name": "test_user"}, ' \
                + '"status": {"ts": "2020-05-01T00:00:00Z", "type": "NEW"}, "history": []}'
-    os.makedirs(f'/tmp/simple_example.{APP_VERSION}.fs.data_path/', exist_ok=True)
-    with open(f'/tmp/simple_example.{APP_VERSION}.fs.data_path/{test_id}.json', 'w') as f:
+    os.makedirs(f'/tmp/hopeit/simple_example.{APP_VERSION}.fs_storage.path/2020/05/01/00/', exist_ok=True)
+    with open(f'/tmp/hopeit/simple_example.{APP_VERSION}.fs_storage.path/2020/05/01/00/{test_id}.json', 'w') as f:
         f.write(json_str)
         f.flush()
-    return test_id
+    return test_id, "2020/05/01/00"
 
 
 @pytest.mark.asyncio
@@ -31,10 +31,11 @@ async def test_query_item(app_config, sample_file_id):  # noqa: F811
                                                  event_name='query_something_extended',
                                                  payload=status,
                                                  postprocess=True,
-                                                 item_id=sample_file_id)
+                                                 item_id=sample_file_id[0],
+                                                 partition_key=sample_file_id[1])
     assert isinstance(result, Something)
     assert result == pp_result
-    assert result.id == sample_file_id
+    assert result.id == sample_file_id[0]
     assert result.status == status
 
 
@@ -46,8 +47,9 @@ async def test_query_item_not_found(app_config):  # noqa: F811
                                                  event_name='query_something_extended',
                                                  payload=status,
                                                  postprocess=True,
-                                                 item_id=item_id)
+                                                 item_id=item_id,
+                                                 partition_key="x")
     assert res.status == 404
     assert result == pp_result
     assert result == SomethingNotFound(
-        path=f'/tmp/simple_example.{APP_VERSION}.fs.data_path', id=item_id)
+        path=f'/tmp/hopeit/simple_example.{APP_VERSION}.fs_storage.path/x', id=item_id)
diff --git a/apps/examples/simple-example/test/integration/test_it_save_something.py b/apps/examples/simple-example/test/integration/test_it_save_something.py
index c444a5e9..df84fc13 100644
--- a/apps/examples/simple-example/test/integration/test_it_save_something.py
+++ b/apps/examples/simple-example/test/integration/test_it_save_something.py
@@ -1,8 +1,12 @@
 from hopeit.app.context import PreprocessHeaders, PreprocessHook
+from hopeit.fs_storage.partition import get_partition_key
 import pytest  # type: ignore
 import json
 
 from hopeit.testing.apps import execute_event
+from hopeit.server.version import APPS_API_VERSION
+
+APP_VERSION = APPS_API_VERSION.replace('.', "x")
 
 
 def mock_user_agent_header(module, context, *, preprocess_hook: PreprocessHook):
@@ -16,7 +20,11 @@ async def test_it_save_something(app_config, something_params_example):  # noqa:
                                  payload=something_params_example,
                                  preprocess=True,
                                  mocks=[mock_user_agent_header])
-    assert result == f"{app_config.env['fs']['data_path']}{something_params_example.id}.json"
+    partition_key = get_partition_key(something_params_example, "%Y/%m/%d/%H")
+    assert result == (
+        f"{app_config.env['storage']['base_path']}simple_example.{APP_VERSION}.fs_storage.path/"
+        f"{partition_key}{something_params_example.id}.json"
+    )
     with open(str(result)) as f:
         fields = json.loads(f.read())
         assert fields['id'] == something_params_example.id
diff --git a/apps/examples/simple-example/test/integration/test_it_spawn_event.py b/apps/examples/simple-example/test/integration/test_it_spawn_event.py
index 70a47333..e4ef2463 100644
--- a/apps/examples/simple-example/test/integration/test_it_spawn_event.py
+++ b/apps/examples/simple-example/test/integration/test_it_spawn_event.py
@@ -2,8 +2,14 @@
 
 from hopeit.dataobjects import copy_payload
 from hopeit.testing.apps import execute_event
+from hopeit.server.version import APPS_API_VERSION
+
+from hopeit.fs_storage.partition import get_partition_key
+
 from simple_example.shuffle.spawn_event import SomethingStored
 
+APP_VERSION = APPS_API_VERSION.replace('.', "x")
+
 
 @pytest.mark.asyncio
 async def test_spawn_event(monkeypatch, app_config,  # noqa: F811
@@ -19,8 +25,12 @@ async def test_spawn_event(monkeypatch, app_config,  # noqa: F811
         expected[i].status.ts = result.payload.status.ts
         for j in range(len(expected[i].history)):
             expected[i].history[j].ts = result.payload.history[j].ts
+        partition_key = get_partition_key(result, "%Y/%m/%d/%H")
         assert result == SomethingStored(
-            path=f"{app_config.env['fs']['data_path']}{result.payload.id}.json",
+            path=(
+                f"{app_config.env['storage']['base_path']}simple_example.{APP_VERSION}.fs_storage.path/"
+                f"{partition_key}{result.payload.id}.json"
+            ),
             payload=expected[i]
         )
     assert msg == f"events submitted to stream: {app_config.events['shuffle.spawn_event'].write_stream.name}"
diff --git a/apps/examples/simple-example/test/integration/test_it_upload_something.py b/apps/examples/simple-example/test/integration/test_it_upload_something.py
index 3e0b535b..b1610db9 100644
--- a/apps/examples/simple-example/test/integration/test_it_upload_something.py
+++ b/apps/examples/simple-example/test/integration/test_it_upload_something.py
@@ -30,7 +30,7 @@ async def test_it_save_something(app_config, something_params_example, something
     assert result == [something_upload_example]
 
     with open(
-        f'/tmp/simple_example.{APP_VERSION}.upload_something.save_path/attachment-test_file_name.bytes',
+        f'/tmp/hopeit/simple_example.{APP_VERSION}.upload_something.save_path/attachment-test_file_name.bytes',
         'rb'
     ) as f:
         data = f.read()
diff --git a/engine/src/hopeit/testing/apps.py b/engine/src/hopeit/testing/apps.py
index 9f9aa696..93f6c788 100644
--- a/engine/src/hopeit/testing/apps.py
+++ b/engine/src/hopeit/testing/apps.py
@@ -8,7 +8,7 @@
 
 from hopeit.app.config import AppConfig, parse_app_config_json, EventDescriptor
 from hopeit.app.context import EventContext, PostprocessHook, PreprocessHook
-from hopeit.dataobjects import EventPayload
+from hopeit.dataobjects import DataObject, EventPayload
 from hopeit.server.config import AuthType, ServerConfig, LoggingConfig
 from hopeit.server.events import EventHandler, get_event_settings
 from hopeit.server.steps import split_event_stages, find_datatype_handler
@@ -147,9 +147,9 @@ async def _preprocess(hook: PreprocessHook, payload: EventPayload) -> EventPaylo
 
     if context is None:
         context = create_test_context(app_config, event_name)
-    impl = find_event_handler(app_config=app_config, event_name=event_name)
 
     event_info = app_config.events[event_name]
+    impl = find_event_handler(app_config=app_config, event_name=event_name, event_info=event_info)
     effective_events = {**split_event_stages(app_config.app, event_name, event_info, impl)}
     handler = EventHandler(
         app_config=app_config, plugins=[],
@@ -173,11 +173,11 @@ async def _preprocess(hook: PreprocessHook, payload: EventPayload) -> EventPaylo
         payload = await _preprocess(preprocess_hook, payload)
         if postprocess_hook and preprocess_hook.status is not None:
             postprocess_hook.set_status(preprocess_hook.status)
-    datatype = find_datatype_handler(app_config=app_config, event_name=event_name)
+    datatype = find_datatype_handler(app_config=app_config, event_name=event_name, event_info=event_info)
     if datatype is None:
         if payload is not None:
             return (payload, payload, postprocess_hook) if postprocess else payload
-    elif not isinstance(payload, datatype):
+    elif not (datatype is DataObject or isinstance(payload, datatype)):
         return (payload, payload, postprocess_hook) if postprocess else payload
 
     on_queue, pp_result, pp_called = [payload], None, False
@@ -224,7 +224,8 @@ async def execute_service(app_config: AppConfig,
     :return: List of results of processing available events
     """
     context = create_test_context(app_config, event_name)
-    impl = find_event_handler(app_config=app_config, event_name=event_name)
+    event_info = app_config.events[event_name]
+    impl = find_event_handler(app_config=app_config, event_name=event_name, event_info=event_info)
     handler = getattr(impl, '__service__')
     count = 0
     results = []
diff --git a/engine/test/integration/toolkit/test_it_auth.py b/engine/test/integration/toolkit/test_it_auth.py
index 2ae4518c..78e87bd7 100644
--- a/engine/test/integration/toolkit/test_it_auth.py
+++ b/engine/test/integration/toolkit/test_it_auth.py
@@ -49,8 +49,8 @@ def test_token_lifecycle(mock_app_config):  # noqa: F811
     context = _setup_server_context(mock_app_config)
     payload = {
         'test': 'test_value',
-        'iat': datetime.now().astimezone(timezone.utc),
-        'exp': datetime.now().astimezone(timezone.utc) + timedelta(seconds=2)
+        'iat': datetime.now(tz=timezone.utc),
+        'exp': datetime.now(tz=timezone.utc) + timedelta(seconds=2)
     }
     token = auth.new_token(mock_app_config.app_key(), payload)
     assert token is not None
@@ -72,8 +72,8 @@ def test_token_lifecycle(mock_app_config):  # noqa: F811
 
     token = auth.new_token(mock_app_config.app_key(), payload={
         'test': 'test_value',
-        'iat': datetime.now().astimezone(timezone.utc),
-        'exp': datetime.now().astimezone(timezone.utc) + timedelta(seconds=2)
+        'iat': datetime.now(tz=timezone.utc),
+        'exp': datetime.now(tz=timezone.utc) + timedelta(seconds=2)
     })
     auth.init(mock_app_config.app_key(), AuthConfig(
         secrets_location=f"/tmp/{uuid.uuid4()}",
@@ -97,8 +97,8 @@ def test_client_tokens(mock_app_config, mock_client_app_config):  # noqa: F811
 
     payload = {
         'test': 'test_value',
-        'iat': datetime.now().astimezone(timezone.utc),
-        'exp': datetime.now().astimezone(timezone.utc) + timedelta(seconds=2)
+        'iat': datetime.now(tz=timezone.utc),
+        'exp': datetime.now(tz=timezone.utc) + timedelta(seconds=2)
     }
 
     # Client-generated token validated in server
@@ -123,8 +123,8 @@ def test_client_not_registered(mock_app_config, mock_client_app_config):  # noqa
 
     payload = {
         'test': 'test_value',
-        'iat': datetime.now().astimezone(timezone.utc),
-        'exp': datetime.now().astimezone(timezone.utc) + timedelta(seconds=2)
+        'iat': datetime.now(tz=timezone.utc),
+        'exp': datetime.now(tz=timezone.utc) + timedelta(seconds=2)
     }
 
     # Client-generated token validated in server
@@ -161,7 +161,7 @@ def test_auth_method_bearer(mock_app_config):  # noqa: F811
     context = _setup_server_context(mock_app_config)
     payload = {
         'test': 'test_value',
-        'exp': datetime.now().astimezone(timezone.utc) + timedelta(seconds=2)
+        'exp': datetime.now(tz=timezone.utc) + timedelta(seconds=2)
     }
     token = auth.new_token(mock_app_config.app_key(), payload)
     assert auth.validate_auth_method(
@@ -177,7 +177,7 @@ def test_auth_method_refresh(mock_app_config):  # noqa: F811
     context = _setup_server_context(mock_app_config)
     payload = {
         'test': 'test_value',
-        'exp': datetime.now().astimezone(timezone.utc) + timedelta(seconds=2)
+        'exp': datetime.now(tz=timezone.utc) + timedelta(seconds=2)
     }
     token = auth.new_token(mock_app_config.app_key(), payload)
     assert auth.validate_auth_method(
diff --git a/engine/test/mock_app/__init__.py b/engine/test/mock_app/__init__.py
index fc291f9d..c324de49 100644
--- a/engine/test/mock_app/__init__.py
+++ b/engine/test/mock_app/__init__.py
@@ -212,6 +212,20 @@ def mock_app_config():
                 write_stream=WriteStreamDescriptor(
                     name='mock_read_write_stream.write'
                 )
+            ),
+            "mock_event_custom": EventDescriptor(
+                type=EventType.GET,
+                impl="mock_app.mock_event_custom_impl"
+            ),
+            "mock_event_dataobject_payload": EventDescriptor(
+                type=EventType.STREAM,
+                read_stream=ReadStreamDescriptor(
+                    name='mock_read_write_stream.read',
+                    consumer_group='mock_event_dataobject_payload'
+                ),
+                dataobjects=[
+                    "mock_app.MockData"
+                ]
             )
         },
         server=ServerConfig(
diff --git a/engine/test/mock_app/mock_event_custom_impl.py b/engine/test/mock_app/mock_event_custom_impl.py
new file mode 100644
index 00000000..d2790d2d
--- /dev/null
+++ b/engine/test/mock_app/mock_event_custom_impl.py
@@ -0,0 +1,48 @@
+from typing import Union
+
+from hopeit.app.logger import app_extra_logger
+from hopeit.app.context import EventContext, PostprocessHook
+from . import MockData
+
+__steps__ = ['entry_point', 'handle_ok_case', 'handle_special_case']
+
+logger, extra = app_extra_logger()
+
+initialized = False
+
+
+def __init_event__(context: EventContext):
+    global initialized
+    logger.info(context, "INIT")
+    initialized = True
+
+
+def entry_point(payload: None, context: EventContext, *, query_arg1: str) -> Union[MockData, str]:
+    logger.info(context, "mock_event.entry_point", extra=extra(query_arg1=query_arg1))
+    assert initialized
+    if query_arg1 == 'fail':
+        raise AssertionError("Test for error")
+    if query_arg1 == 'ok':
+        return MockData(value='ok')
+    return "None"
+
+
+async def handle_special_case(payload: str, context: EventContext) -> str:
+    assert initialized
+    return payload
+
+
+async def handle_ok_case(payload: MockData, context: EventContext) -> str:
+    assert initialized
+    return "ok: " + payload.value
+
+
+async def __postprocess__(payload: str, context: EventContext, response: PostprocessHook) -> str:
+    if payload.startswith('ok:'):
+        response.set_status(200)
+        response.set_header('X-Status', 'ok')
+        response.set_cookie('Test-Cookie', 'ok')
+        return payload
+    response.set_status(400)
+    response.del_cookie('Test-Cookie')
+    return "not-ok: " + payload
diff --git a/engine/test/mock_app/mock_event_dataobject_payload.py b/engine/test/mock_app/mock_event_dataobject_payload.py
new file mode 100644
index 00000000..74a3ee48
--- /dev/null
+++ b/engine/test/mock_app/mock_event_dataobject_payload.py
@@ -0,0 +1,13 @@
+from hopeit.app.logger import app_extra_logger
+from hopeit.app.context import EventContext
+from hopeit.dataobjects import DataObject
+from hopeit.dataobjects.payload import Payload
+
+__steps__ = ['entry_point']
+
+logger, extra = app_extra_logger()
+
+
+def entry_point(payload: DataObject, context: EventContext) -> str:
+    logger.info(context, "mock_event_dataobject_payload.entry_point")
+    return Payload.to_json(payload)
diff --git a/engine/test/mock_engine/__init__.py b/engine/test/mock_engine/__init__.py
index ae23d09e..ea00a306 100644
--- a/engine/test/mock_engine/__init__.py
+++ b/engine/test/mock_engine/__init__.py
@@ -43,6 +43,7 @@ class MockEventHandler(EventHandler):
         'track.session_id': 'test_session_id'
     }
     expected_track_ids = test_track_ids
+    call_function = None
 
     def __init__(self, *,
                  app_config: AppConfig,
@@ -59,11 +60,13 @@ async def handle_async_event(self, *,
                                  query_args: Optional[dict],
                                  payload: Optional[EventPayload]) -> AsyncGenerator[Optional[EventPayload], None]:
         assert payload == MockEventHandler.input_payload
-        if isinstance(payload, MockData) and payload.value == 'fail':  # type: ignore
+        if MockEventHandler.call_function is not None:
+            yield MockEventHandler.call_function(payload, context)
+        elif isinstance(payload, MockData) and payload.value == 'fail':  # type: ignore
             raise ValueError("Test for error")
-        if isinstance(payload, MockData) and payload.value == 'cancel':  # type: ignore
+        elif isinstance(payload, MockData) and payload.value == 'cancel':  # type: ignore
             raise asyncio.CancelledError("Test for cancellation")
-        if isinstance(payload, MockData) and payload.value == 'timeout':  # type: ignore
+        elif isinstance(payload, MockData) and payload.value == 'timeout':  # type: ignore
             await asyncio.sleep(5.0)
         yield MockEventHandler.expected_result
 
diff --git a/engine/test/unit/dataobjects/test_dataobjects.py b/engine/test/unit/dataobjects/test_dataobjects.py
index 9f482208..80fbdd57 100644
--- a/engine/test/unit/dataobjects/test_dataobjects.py
+++ b/engine/test/unit/dataobjects/test_dataobjects.py
@@ -1,5 +1,5 @@
 import uuid
-from datetime import datetime
+from datetime import datetime, timezone
 
 from hopeit.dataobjects import StreamEventParams, copy_payload
 
@@ -8,7 +8,7 @@
 
 
 def test_data_event_params_extract_attr():
-    now = datetime.now()
+    now = datetime.now(tz=timezone.utc)
     obj = MockData('id1', 'value1', MockNested(now))
     assert StreamEventParams.extract_attr(obj, 'id') == 'id1'
     assert StreamEventParams.extract_attr(obj, 'value') == 'value1'
@@ -17,7 +17,7 @@ def test_data_event_params_extract_attr():
 
 
 def test_data_event():
-    now = datetime.now()
+    now = datetime.now(tz=timezone.utc)
     obj = MockData('id1', 'value1', MockNested(now))
     assert obj.event_id() == 'id1'
     assert obj.event_ts() == now
@@ -37,7 +37,7 @@ def test_data_event_with_auto_event_id(monkeypatch):
 
 
 def test_copy_mutable_dataobject():
-    now = datetime.now()
+    now = datetime.now(tz=timezone.utc)
     obj = MockData('id1', 'value1', MockNested(now))
     new = copy_payload(obj)
     assert new == obj
@@ -45,7 +45,7 @@ def test_copy_mutable_dataobject():
 
 
 def test_copy_immutable_dataobject_should_return_same():
-    now = datetime.now()
+    now = datetime.now(tz=timezone.utc)
     obj = MockDataImmutable('id1', 'value1', MockNested(now))
     new = copy_payload(obj)
     assert new == obj
@@ -53,7 +53,7 @@ def test_copy_immutable_dataobject_should_return_same():
 
 
 def test_copy_unsafe_dataobject_should_return_same():
-    now = datetime.now()
+    now = datetime.now(tz=timezone.utc)
     obj = MockDataUnsafe('id1', 'value1', MockNested(now))
     new = copy_payload(obj)
     assert new == obj
diff --git a/engine/test/unit/server/test_engine.py b/engine/test/unit/server/test_engine.py
index d9e1981b..06702dcf 100644
--- a/engine/test/unit/server/test_engine.py
+++ b/engine/test/unit/server/test_engine.py
@@ -151,8 +151,9 @@ async def test_read_stream(monkeypatch, mock_app_config, mock_plugin_config):  #
     payload = MockData("ok")
     expected = MockResult("ok: ok")
     setup_mocks(monkeypatch)
+    from mock_app import mock_event_dataobject_payload
     monkeypatch.setattr(MockEventHandler, 'input_payload', payload)
-    monkeypatch.setattr(MockEventHandler, 'expected_result', expected)
+    monkeypatch.setattr(MockEventHandler, 'call_function', mock_event_dataobject_payload.entry_point)
     monkeypatch.setattr(MockStreamManager, 'test_payload', payload)
     engine = await create_engine(app_config=mock_app_config, plugin=mock_plugin_config)
     monkeypatch.setattr(engine, 'stream_manager', MockStreamManager(address='test'))
@@ -161,6 +162,21 @@ async def test_read_stream(monkeypatch, mock_app_config, mock_plugin_config):  #
     await engine.stop()
 
 
+@pytest.mark.asyncio
+async def test_read_stream_dataobject_payload(monkeypatch, mock_app_config, mock_plugin_config):  # noqa: F811
+    payload = MockData("ok")
+    expected = """{"value": "ok"}"""
+    setup_mocks(monkeypatch)
+    monkeypatch.setattr(MockEventHandler, 'input_payload', payload)
+    monkeypatch.setattr(MockEventHandler, 'expected_result', expected)
+    monkeypatch.setattr(MockStreamManager, 'test_payload', payload)
+    engine = await create_engine(app_config=mock_app_config, plugin=mock_plugin_config)
+    monkeypatch.setattr(engine, 'stream_manager', MockStreamManager(address='test'))
+    res = await engine.read_stream(event_name='mock_event_dataobject_payload', test_mode=True)
+    assert res == expected
+    await engine.stop()
+
+
 @pytest.mark.asyncio
 async def test_read_write_stream_auto_queue(
     monkeypatch, mock_app_config, mock_plugin_config  # noqa: F811
diff --git a/engine/test/unit/server/test_imports.py b/engine/test/unit/server/test_imports.py
index 62a49ed0..6fd8a618 100644
--- a/engine/test/unit/server/test_imports.py
+++ b/engine/test/unit/server/test_imports.py
@@ -1,15 +1,29 @@
 import pytest  # type: ignore
 from mock_app import mock_app_config  # type: ignore  # noqa: F401
 import mock_app.mock_event as mock_event  # type: ignore
+import mock_app.mock_event_custom_impl as mock_event_custom_impl  # type: ignore
 
 from hopeit.server.imports import find_event_handler
 
 
 def test_find_event_handler(mock_app_config):  # noqa: F811
-    impl = find_event_handler(app_config=mock_app_config, event_name='mock_event')
+    impl = find_event_handler(
+        app_config=mock_app_config, event_name='mock_event', event_info=mock_app_config.events['mock_event']
+    )
     assert impl is mock_event
 
 
 def test_find_event_handler_not_found(mock_app_config):  # noqa: F811
     with pytest.raises(ImportError):
-        find_event_handler(app_config=mock_app_config, event_name='unknown')
+        find_event_handler(
+            app_config=mock_app_config, event_name='unknown', event_info=mock_app_config.events['mock_event']
+        )
+
+
+def test_find_event_handler_custom_impl(mock_app_config):  # noqa: F811
+    impl = find_event_handler(
+        app_config=mock_app_config,
+        event_name='mock_event_custom',
+        event_info=mock_app_config.events['mock_event_custom']
+    )
+    assert impl is mock_event_custom_impl
diff --git a/engine/test/unit/server/test_metrics.py b/engine/test/unit/server/test_metrics.py
index f9760c81..0035559b 100644
--- a/engine/test/unit/server/test_metrics.py
+++ b/engine/test/unit/server/test_metrics.py
@@ -17,8 +17,8 @@ class MockDatetime(datetime.datetime):
     ts = 0.0
 
     @classmethod
-    def now(cls):
-        return datetime.datetime.fromtimestamp(cls.ts)
+    def now(cls, tz=None):
+        return datetime.datetime.fromtimestamp(cls.ts, tz=tz)
 
 
 def test_metrics(mock_app_config):  # noqa: F811
diff --git a/engine/test/unit/server/test_steps.py b/engine/test/unit/server/test_steps.py
index fe466e31..c880f547 100644
--- a/engine/test/unit/server/test_steps.py
+++ b/engine/test/unit/server/test_steps.py
@@ -255,7 +255,11 @@ async def test_invoke_single_step():
 
 
 def test_split_event_stages(mock_app_config):  # noqa: F811
-    impl = find_event_handler(app_config=mock_app_config, event_name='mock_shuffle_event')
+    impl = find_event_handler(
+        app_config=mock_app_config,
+        event_name='mock_shuffle_event',
+        event_info=mock_app_config.events['mock_shuffle_event']
+    )
     event_info = mock_app_config.events['mock_shuffle_event']
     stages = split_event_stages(mock_app_config.app,
                                 event_name='mock_shuffle_event',
@@ -286,7 +290,11 @@ def test_split_event_stages(mock_app_config):  # noqa: F811
 
 
 def test_split_event_stages_queues(mock_app_config):  # noqa: F811
-    impl = find_event_handler(app_config=mock_app_config, event_name='mock_shuffle_event')
+    impl = find_event_handler(
+        app_config=mock_app_config,
+        event_name='mock_shuffle_event',
+        event_info=mock_app_config.events['mock_shuffle_event']
+    )
     event_info = deepcopy(mock_app_config.events['mock_shuffle_event'])
     event_info.read_stream = ReadStreamDescriptor(
         name="test_read_stream", consumer_group="test_group", queues=["q1", "q2"]
@@ -323,7 +331,11 @@ def test_split_event_stages_queues(mock_app_config):  # noqa: F811
 
 
 def test_collector_steps_descriptor(mock_app_config):  # noqa: F811
-    impl = find_event_handler(app_config=mock_app_config, event_name='mock_collector')
+    impl = find_event_handler(
+        app_config=mock_app_config,
+        event_name='mock_collector',
+        event_info=mock_app_config.events['mock_collector']
+    )
     steps = extract_module_steps(impl)
     collector: CollectorStepsDescriptor = steps[0][0]
     assert isinstance(collector, CollectorStepsDescriptor)
diff --git a/engine/test/unit/streams/test_streams.py b/engine/test/unit/streams/test_streams.py
index 55bc0d08..98f1ba30 100644
--- a/engine/test/unit/streams/test_streams.py
+++ b/engine/test/unit/streams/test_streams.py
@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timezone
 
 import pytest
 
@@ -19,7 +19,7 @@ class MockInvalidDataEvent:
 
 
 def test_as_data_event():  # noqa: F811
-    test_data = MockData("ok", datetime.now())
+    test_data = MockData("ok", datetime.now(tz=timezone.utc))
     assert StreamManager.as_data_event(test_data) == test_data
     with pytest.raises(NotImplementedError):
         StreamManager.as_data_event(MockInvalidDataEvent("ok"))
diff --git a/plugins/auth/basic-auth/test/unit/test_logout.py b/plugins/auth/basic-auth/test/unit/test_logout.py
index c45b64af..78c3bf10 100644
--- a/plugins/auth/basic-auth/test/unit/test_logout.py
+++ b/plugins/auth/basic-auth/test/unit/test_logout.py
@@ -1,5 +1,5 @@
 import pytest  # type: ignore
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 import hopeit.toolkit.auth as auth
 from hopeit.app.context import EventContext, PostprocessHook
@@ -32,7 +32,7 @@ async def execute_flow(context):
 def _event_context(mock_app_config, plugin_config):  # noqa: F811
     settings = get_event_settings(plugin_config.effective_settings, "logout")
     cfg = settings(key='auth', datatype=AuthSettings)
-    iat = datetime.now()
+    iat = datetime.now(tz=timezone.utc)
     timeout = cfg.access_token_expiration
     return EventContext(
         app_config=mock_app_config,
diff --git a/plugins/auth/basic-auth/test/unit/test_refresh.py b/plugins/auth/basic-auth/test/unit/test_refresh.py
index 37ba2773..db5f8f7e 100644
--- a/plugins/auth/basic-auth/test/unit/test_refresh.py
+++ b/plugins/auth/basic-auth/test/unit/test_refresh.py
@@ -1,5 +1,5 @@
 import pytest  # type: ignore
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 import hopeit.toolkit.auth as auth
 from hopeit.app.context import EventContext, PostprocessHook
@@ -69,7 +69,7 @@ async def execute_flow(context):
 def _event_context(mock_app_config, plugin_config):  # noqa: F811
     settings = get_event_settings(plugin_config.effective_settings, "refresh")
     cfg = settings(key='auth', datatype=AuthSettings)
-    iat = datetime.now()
+    iat = datetime.now(tz=timezone.utc)
     timeout = cfg.access_token_expiration
     return EventContext(
         app_config=mock_app_config,
diff --git a/plugins/build/ci-test-plugins.sh b/plugins/build/ci-test-plugins.sh
index 1e3b1bf0..fbf9ba31 100644
--- a/plugins/build/ci-test-plugins.sh
+++ b/plugins/build/ci-test-plugins.sh
@@ -19,7 +19,7 @@ export PYTHONPATH=engine/src/:plugins/storage/redis/src/ && python3 -m pytest -v
 code+=$?
 
 # storage/fs
-export PYTHONPATH=engine/src/:plugins/storage/fs/src/ && python3 -m pytest -v --cov-fail-under=90 --cov-report=term --cov=plugins/storage/fs/src/ plugins/storage/fs/test/unit/
+export PYTHONPATH=engine/src/:plugins/storage/fs/src/ && python3 -m pytest -v --cov-fail-under=90 --cov-report=term --cov=plugins/storage/fs/src/ plugins/storage/fs/test/unit/ plugins/storage/fs/test/integration/
 code+=$?
 
 # ops/apps-visualizer
diff --git a/plugins/ops/apps-visualizer/test/integration/conftest.py b/plugins/ops/apps-visualizer/test/integration/conftest.py
index 2162d02f..003e215d 100644
--- a/plugins/ops/apps-visualizer/test/integration/conftest.py
+++ b/plugins/ops/apps-visualizer/test/integration/conftest.py
@@ -2,7 +2,7 @@
 import json
 import os
 from pathlib import Path
-from datetime import datetime
+from datetime import datetime, timezone
 import socket
 
 import pytest
@@ -73,7 +73,7 @@ async def log_entries() -> LogBatch:
         f"2021-06-02 18:01:44,303 | INFO | simple-example {APPS_API_VERSION} login host 17031 | DONE | response.status=404 | metrics.duration=13.057 | track.operation_id=f2659a30-5ac4-4dd4-b1f7-9a00db0bf7d5 | track.request_id=7ee59fa7-c1e4-4a60-a79b-a25dbbd6cb82 | track.request_ts=2021-06-02T18:01:44.289394+00:00 | track.caller=test | track.session_id=test | event.app=simple_example.{APPS_ROUTE_VERSION} | event.plugin=basic_auth.{APPS_ROUTE_VERSION}",  # noqa: E501
     ])
     batch = await _process_log_entries(raw)
-    ts = datetime.now()
+    ts = datetime.now(tz=timezone.utc)
     for entry in batch.entries:
         entry.ts = ts.strftime("%Y-%m-%d %H:%M:%S")
         entry.host = socket.gethostname()
diff --git a/plugins/ops/apps-visualizer/test/integration/effective_events.json b/plugins/ops/apps-visualizer/test/integration/effective_events.json
index 8939d780..4db61607 100644
--- a/plugins/ops/apps-visualizer/test/integration/effective_events.json
+++ b/plugins/ops/apps-visualizer/test/integration/effective_events.json
@@ -676,6 +676,25 @@
         }
       },
       "auth": []
+    },
+    "simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs": {
+      "type": "STREAM",
+      "plug_mode": "Standalone",
+      "impl": "hopeit.fs_storage.events.stream_batch_storage",
+      "connections": [],
+      "read_stream": {
+        "name": "simple_example.${APPS_ROUTE_VERSION}.streams.something_event",
+        "consumer_group": "simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs",
+        "queues": [
+          "high-prio",
+          "AUTO"
+        ]
+      },
+      "auth": [],
+      "setting_keys": [],
+      "dataobjects": [
+        "model.Something"
+      ]
     }
   },
   "client_example.${APPS_ROUTE_VERSION}": {
diff --git a/plugins/ops/apps-visualizer/test/integration/events_graph_data_expanded.json b/plugins/ops/apps-visualizer/test/integration/events_graph_data_expanded.json
index 705360b4..a5af66d0 100644
--- a/plugins/ops/apps-visualizer/test/integration/events_graph_data_expanded.json
+++ b/plugins/ops/apps-visualizer/test/integration/events_graph_data_expanded.json
@@ -307,6 +307,13 @@
     },
     "classes": "EVENT"
   },
+  "simple_example.${APPS_ROUTE_VERSION}.simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs": {
+    "data": {
+      "id": "simple_example.${APPS_ROUTE_VERSION}.simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs",
+      "content": "simple_example.${APPS_ROUTE_VERSION}\nsimple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs"
+    },
+    "classes": "EVENT"
+  },
   "client_example.${APPS_ROUTE_VERSION}.count_and_save.GET": {
     "data": {
       "id": "client_example.${APPS_ROUTE_VERSION}.count_and_save.GET",
@@ -417,6 +424,14 @@
       "label": ""
     }
   },
+  "edge_simple_example.${APPS_ROUTE_VERSION}.simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.AUTO": {
+    "data": {
+      "id": "edge_simple_example.${APPS_ROUTE_VERSION}.simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.AUTO",
+      "source": ">simple_example.${APPS_ROUTE_VERSION}.streams.something_event.AUTO",
+      "target": "simple_example.${APPS_ROUTE_VERSION}.simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs",
+      "label": ""
+    }
+  },
   "edge_simple_example.${APPS_ROUTE_VERSION}.service.something_generator.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.AUTO": {
     "data": {
       "id": "edge_simple_example.${APPS_ROUTE_VERSION}.service.something_generator.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.AUTO",
@@ -441,6 +456,14 @@
       "label": "high-prio"
     }
   },
+  "edge_simple_example.${APPS_ROUTE_VERSION}.simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.high-prio.high-prio": {
+    "data": {
+      "id": "edge_simple_example.${APPS_ROUTE_VERSION}.simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.high-prio.high-prio",
+      "source": ">simple_example.${APPS_ROUTE_VERSION}.streams.something_event.high-prio",
+      "target": "simple_example.${APPS_ROUTE_VERSION}.simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs",
+      "label": "high-prio"
+    }
+  },
   "edge_simple_example.${APPS_ROUTE_VERSION}.streams.something_event.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.high-prio.high-prio": {
     "data": {
       "id": "edge_simple_example.${APPS_ROUTE_VERSION}.streams.something_event.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.high-prio.high-prio",
diff --git a/plugins/ops/apps-visualizer/test/integration/events_graph_data_standard.json b/plugins/ops/apps-visualizer/test/integration/events_graph_data_standard.json
index ed54e8ab..2593c82d 100644
--- a/plugins/ops/apps-visualizer/test/integration/events_graph_data_standard.json
+++ b/plugins/ops/apps-visualizer/test/integration/events_graph_data_standard.json
@@ -244,6 +244,13 @@
     },
     "classes": "EVENT"
   },
+  "simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs": {
+    "data": {
+      "id": "simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs",
+      "content": "simple_example.${APPS_ROUTE_VERSION}\nstorage.save_events_fs"
+    },
+    "classes": "EVENT"
+  },
   "client_example.${APPS_ROUTE_VERSION}.count_and_save.GET": {
     "data": {
       "id": "client_example.${APPS_ROUTE_VERSION}.count_and_save.GET",
@@ -362,6 +369,22 @@
       "label": ""
     }
   },
+  "edge_simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.high-prio": {
+    "data": {
+      "id": "edge_simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.high-prio",
+      "source": ">simple_example.${APPS_ROUTE_VERSION}.streams.something_event",
+      "target": "simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs",
+      "label": "high-prio"
+    }
+  },
+  "edge_simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.AUTO": {
+    "data": {
+      "id": "edge_simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.AUTO",
+      "source": ">simple_example.${APPS_ROUTE_VERSION}.streams.something_event",
+      "target": "simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs",
+      "label": ""
+    }
+  },
   "edge_simple_example.${APPS_ROUTE_VERSION}.service.something_generator.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.AUTO": {
     "data": {
       "id": "edge_simple_example.${APPS_ROUTE_VERSION}.service.something_generator.simple_example.${APPS_ROUTE_VERSION}.streams.something_event.AUTO",
@@ -474,4 +497,4 @@
       "label": ""
     }
   }
-}
+}
\ No newline at end of file
diff --git a/plugins/ops/apps-visualizer/test/integration/test_events_graph.py b/plugins/ops/apps-visualizer/test/integration/test_events_graph.py
index f350bb49..9f116663 100644
--- a/plugins/ops/apps-visualizer/test/integration/test_events_graph.py
+++ b/plugins/ops/apps-visualizer/test/integration/test_events_graph.py
@@ -12,7 +12,6 @@ async def test_simple_example_events_diagram(
     async with mock_lock:
         mock_runtime(monkeypatch, effective_events)
         result = await execute_event(app_config=plugin_config, event_name="apps.events-graph", payload=None)
-
         assert result.graph.data == events_graph_data_standard
         assert result.options.expanded_view is False
 
diff --git a/plugins/ops/config-manager/test/integration/runtime_simple_example.json b/plugins/ops/config-manager/test/integration/runtime_simple_example.json
index 428b8cfe..d4a77c09 100644
--- a/plugins/ops/config-manager/test/integration/runtime_simple_example.json
+++ b/plugins/ops/config-manager/test/integration/runtime_simple_example.json
@@ -31,12 +31,12 @@
         },
         "app_connections": {},
         "env": {
-          "fs": {
-            "data_path": "/tmp/simple_example.${APPS_ROUTE_VERSION}.fs.data_path/"
+          "storage": {
+            "base_path": "/tmp/hopeit/"
           },
           "upload_something": {
-            "save_path": "/tmp/simple_example.${APPS_ROUTE_VERSION}.upload_something.save_path/",
-            "chunk_size": 16384
+            "save_path": "/tmp/hopeit//simple_example.${APPS_ROUTE_VERSION}.upload_something.save_path/",
+            "chunk_size": "16384"
           }
         },
         "events": {
@@ -45,7 +45,10 @@
             "plug_mode": "Standalone",
             "connections": [],
             "auth": [],
-            "setting_keys": []
+            "setting_keys": [
+              "fs_storage"
+            ],
+            "dataobjects": []
           },
           "query_something": {
             "type": "GET",
@@ -53,7 +56,10 @@
             "route": "simple-example/${APPS_API_VERSION}/query_something",
             "connections": [],
             "auth": [],
-            "setting_keys": []
+            "setting_keys": [
+              "fs_storage"
+            ],
+            "dataobjects": []
           },
           "query_something_extended": {
             "type": "POST",
@@ -61,21 +67,28 @@
             "route": "simple-example/${APPS_API_VERSION}/query_something",
             "connections": [],
             "auth": [],
-            "setting_keys": []
+            "setting_keys": [
+              "fs_storage"
+            ],
+            "dataobjects": []
           },
           "save_something": {
             "type": "POST",
             "plug_mode": "Standalone",
             "connections": [],
             "auth": [],
-            "setting_keys": []
+            "setting_keys": [
+              "fs_storage"
+            ],
+            "dataobjects": []
           },
           "download_something": {
             "type": "GET",
             "plug_mode": "Standalone",
             "connections": [],
             "auth": [],
-            "setting_keys": []
+            "setting_keys": [],
+            "dataobjects": []
           },
           "download_something_streamed": {
             "type": "GET",
@@ -84,14 +97,16 @@
             "auth": [
               "Unsecured"
             ],
-            "setting_keys": []
+            "setting_keys": [],
+            "dataobjects": []
           },
           "upload_something": {
             "type": "MULTIPART",
             "plug_mode": "Standalone",
             "connections": [],
             "auth": [],
-            "setting_keys": []
+            "setting_keys": [],
+            "dataobjects": []
           },
           "service.something_generator": {
             "type": "SERVICE",
@@ -105,7 +120,8 @@
               "queue_strategy": "DROP"
             },
             "auth": [],
-            "setting_keys": []
+            "setting_keys": [],
+            "dataobjects": []
           },
           "streams.something_event": {
             "type": "POST",
@@ -119,7 +135,8 @@
               "queue_strategy": "DROP"
             },
             "auth": [],
-            "setting_keys": []
+            "setting_keys": [],
+            "dataobjects": []
           },
           "streams.process_events": {
             "type": "STREAM",
@@ -134,14 +151,20 @@
               ]
             },
             "auth": [],
-            "setting_keys": []
+            "setting_keys": [
+              "fs_storage"
+            ],
+            "dataobjects": []
           },
           "collector.query_concurrently": {
             "type": "POST",
             "plug_mode": "Standalone",
             "connections": [],
             "auth": [],
-            "setting_keys": []
+            "setting_keys": [
+              "fs_storage"
+            ],
+            "dataobjects": []
           },
           "collector.collect_spawn": {
             "type": "POST",
@@ -155,7 +178,10 @@
               "queue_strategy": "DROP"
             },
             "auth": [],
-            "setting_keys": []
+            "setting_keys": [
+              "fs_storage"
+            ],
+            "dataobjects": []
           },
           "shuffle.spawn_event": {
             "type": "POST",
@@ -169,7 +195,10 @@
               "queue_strategy": "DROP"
             },
             "auth": [],
-            "setting_keys": []
+            "setting_keys": [
+              "fs_storage"
+            ],
+            "dataobjects": []
           },
           "shuffle.parallelize_event": {
             "type": "POST",
@@ -183,7 +212,29 @@
               "queue_strategy": "DROP"
             },
             "auth": [],
-            "setting_keys": []
+            "setting_keys": [
+              "fs_storage"
+            ],
+            "dataobjects": []
+          },
+          "storage.save_events_fs": {
+            "type": "STREAM",
+            "plug_mode": "Standalone",
+            "impl": "hopeit.fs_storage.events.stream_batch_storage",
+            "connections": [],
+            "read_stream": {
+              "name": "simple_example.${APPS_ROUTE_VERSION}.streams.something_event",
+              "consumer_group": "simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs",
+              "queues": [
+                "high-prio",
+                "AUTO"
+              ]
+            },
+            "auth": [],
+            "setting_keys": [],
+            "dataobjects": [
+              "model.Something"
+            ]
           }
         },
         "server": {
@@ -207,7 +258,7 @@
             ]
           },
           "api": {},
-          "engine_version": "${APPS_API_VERSION}.2"
+          "engine_version": "${ENGINE_VERSION}"
         },
         "plugins": [
           {
@@ -260,6 +311,16 @@
               ]
             }
           },
+          "fs_storage": {
+            "path": "/tmp/hopeit//simple_example.${APPS_ROUTE_VERSION}.fs_storage.path",
+            "partition_dateformat": "%Y/%m/%d/%H/"
+          },
+          "storage.save_events_fs": {
+            "path": "/tmp/hopeit//simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs.path",
+            "partition_dateformat": "%Y/%m/%d/%H/",
+            "flush_seconds": 60.0,
+            "flush_max_size": 100
+          },
           "simple_example.${APPS_ROUTE_VERSION}.streams.something_event": {
             "target_max_len": 1000,
             "throttle_ms": 10,
@@ -286,7 +347,12 @@
               "compression": "lz4",
               "serialization": "json+base64"
             },
-            "extras": {}
+            "extras": {
+              "fs_storage": {
+                "path": "/tmp/hopeit//simple_example.${APPS_ROUTE_VERSION}.fs_storage.path",
+                "partition_dateformat": "%Y/%m/%d/%H/"
+              }
+            }
           },
           "query_something": {
             "response_timeout": 60.0,
@@ -307,7 +373,12 @@
               "compression": "lz4",
               "serialization": "json+base64"
             },
-            "extras": {}
+            "extras": {
+              "fs_storage": {
+                "path": "/tmp/hopeit//simple_example.${APPS_ROUTE_VERSION}.fs_storage.path",
+                "partition_dateformat": "%Y/%m/%d/%H/"
+              }
+            }
           },
           "query_something_extended": {
             "response_timeout": 60.0,
@@ -328,7 +399,12 @@
               "compression": "lz4",
               "serialization": "json+base64"
             },
-            "extras": {}
+            "extras": {
+              "fs_storage": {
+                "path": "/tmp/hopeit//simple_example.${APPS_ROUTE_VERSION}.fs_storage.path",
+                "partition_dateformat": "%Y/%m/%d/%H/"
+              }
+            }
           },
           "save_something": {
             "response_timeout": 60.0,
@@ -349,7 +425,12 @@
               "compression": "lz4",
               "serialization": "json+base64"
             },
-            "extras": {}
+            "extras": {
+              "fs_storage": {
+                "path": "/tmp/hopeit//simple_example.${APPS_ROUTE_VERSION}.fs_storage.path",
+                "partition_dateformat": "%Y/%m/%d/%H/"
+              }
+            }
           },
           "download_something": {
             "response_timeout": 60.0,
@@ -507,6 +588,10 @@
                     "something_id"
                   ]
                 }
+              },
+              "fs_storage": {
+                "path": "/tmp/hopeit//simple_example.${APPS_ROUTE_VERSION}.fs_storage.path",
+                "partition_dateformat": "%Y/%m/%d/%H/"
               }
             }
           },
@@ -529,7 +614,12 @@
               "compression": "lz4",
               "serialization": "json+base64"
             },
-            "extras": {}
+            "extras": {
+              "fs_storage": {
+                "path": "/tmp/hopeit//simple_example.${APPS_ROUTE_VERSION}.fs_storage.path",
+                "partition_dateformat": "%Y/%m/%d/%H/"
+              }
+            }
           },
           "collector.collect_spawn": {
             "response_timeout": 60.0,
@@ -559,6 +649,10 @@
                     "something_id"
                   ]
                 }
+              },
+              "fs_storage": {
+                "path": "/tmp/hopeit//simple_example.${APPS_ROUTE_VERSION}.fs_storage.path",
+                "partition_dateformat": "%Y/%m/%d/%H/"
               }
             }
           },
@@ -590,6 +684,10 @@
                     "something_id"
                   ]
                 }
+              },
+              "fs_storage": {
+                "path": "/tmp/hopeit//simple_example.${APPS_ROUTE_VERSION}.fs_storage.path",
+                "partition_dateformat": "%Y/%m/%d/%H/"
               }
             }
           },
@@ -621,6 +719,38 @@
                     "something_id"
                   ]
                 }
+              },
+              "fs_storage": {
+                "path": "/tmp/hopeit//simple_example.${APPS_ROUTE_VERSION}.fs_storage.path",
+                "partition_dateformat": "%Y/%m/%d/%H/"
+              }
+            }
+          },
+          "storage.save_events_fs": {
+            "response_timeout": 60.0,
+            "logging": {
+              "extra_fields": [],
+              "stream_fields": [
+                "stream.name",
+                "stream.msg_id",
+                "stream.consumer_group"
+              ]
+            },
+            "stream": {
+              "timeout": 60.0,
+              "target_max_len": 1000,
+              "throttle_ms": 10,
+              "step_delay": 0,
+              "batch_size": 5,
+              "compression": "lz4",
+              "serialization": "json+base64"
+            },
+            "extras": {
+              "_": {
+                "path": "/tmp/hopeit//simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs.path",
+                "partition_dateformat": "%Y/%m/%d/%H/",
+                "flush_seconds": 60.0,
+                "flush_max_size": 100
               }
             }
           }
@@ -636,7 +766,8 @@
           ],
           "setting_keys": [
             "auth"
-          ]
+          ],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.basic_auth.${APPS_ROUTE_VERSION}.refresh": {
           "type": "GET",
@@ -647,7 +778,8 @@
           ],
           "setting_keys": [
             "auth"
-          ]
+          ],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.basic_auth.${APPS_ROUTE_VERSION}.logout": {
           "type": "GET",
@@ -658,14 +790,18 @@
           ],
           "setting_keys": [
             "auth"
-          ]
+          ],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.list_somethings": {
           "type": "GET",
           "plug_mode": "Standalone",
           "connections": [],
           "auth": [],
-          "setting_keys": []
+          "setting_keys": [
+            "fs_storage"
+          ],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.query_something": {
           "type": "GET",
@@ -673,7 +809,10 @@
           "route": "simple-example/${APPS_API_VERSION}/query_something",
           "connections": [],
           "auth": [],
-          "setting_keys": []
+          "setting_keys": [
+            "fs_storage"
+          ],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.query_something_extended": {
           "type": "POST",
@@ -681,21 +820,28 @@
           "route": "simple-example/${APPS_API_VERSION}/query_something",
           "connections": [],
           "auth": [],
-          "setting_keys": []
+          "setting_keys": [
+            "fs_storage"
+          ],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.save_something": {
           "type": "POST",
           "plug_mode": "Standalone",
           "connections": [],
           "auth": [],
-          "setting_keys": []
+          "setting_keys": [
+            "fs_storage"
+          ],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.download_something": {
           "type": "GET",
           "plug_mode": "Standalone",
           "connections": [],
           "auth": [],
-          "setting_keys": []
+          "setting_keys": [],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.download_something_streamed": {
           "type": "GET",
@@ -704,14 +850,16 @@
           "auth": [
             "Unsecured"
           ],
-          "setting_keys": []
+          "setting_keys": [],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.upload_something": {
           "type": "MULTIPART",
           "plug_mode": "Standalone",
           "connections": [],
           "auth": [],
-          "setting_keys": []
+          "setting_keys": [],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.service.something_generator": {
           "type": "SERVICE",
@@ -725,7 +873,8 @@
             "queue_strategy": "DROP"
           },
           "auth": [],
-          "setting_keys": []
+          "setting_keys": [],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.streams.something_event": {
           "type": "POST",
@@ -739,7 +888,8 @@
             "queue_strategy": "DROP"
           },
           "auth": [],
-          "setting_keys": []
+          "setting_keys": [],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.streams.process_events": {
           "type": "STREAM",
@@ -754,14 +904,20 @@
             ]
           },
           "auth": [],
-          "setting_keys": []
+          "setting_keys": [
+            "fs_storage"
+          ],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.collector.query_concurrently": {
           "type": "POST",
           "plug_mode": "Standalone",
           "connections": [],
           "auth": [],
-          "setting_keys": []
+          "setting_keys": [
+            "fs_storage"
+          ],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.collector.collect_spawn": {
           "type": "POST",
@@ -775,7 +931,10 @@
             "queue_strategy": "DROP"
           },
           "auth": [],
-          "setting_keys": []
+          "setting_keys": [
+            "fs_storage"
+          ],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.shuffle.spawn_event": {
           "type": "POST",
@@ -789,7 +948,10 @@
             "queue_strategy": "DROP"
           },
           "auth": [],
-          "setting_keys": []
+          "setting_keys": [
+            "fs_storage"
+          ],
+          "dataobjects": []
         },
         "simple_example.${APPS_ROUTE_VERSION}.shuffle.parallelize_event": {
           "type": "POST",
@@ -803,7 +965,29 @@
             "queue_strategy": "DROP"
           },
           "auth": [],
-          "setting_keys": []
+          "setting_keys": [
+            "fs_storage"
+          ],
+          "dataobjects": []
+        },
+        "simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs": {
+          "type": "STREAM",
+          "plug_mode": "Standalone",
+          "impl": "hopeit.fs_storage.events.stream_batch_storage",
+          "connections": [],
+          "read_stream": {
+            "name": "simple_example.${APPS_ROUTE_VERSION}.streams.something_event",
+            "consumer_group": "simple_example.${APPS_ROUTE_VERSION}.storage.save_events_fs",
+            "queues": [
+              "high-prio",
+              "AUTO"
+            ]
+          },
+          "auth": [],
+          "setting_keys": [],
+          "dataobjects": [
+            "model.Something"
+          ]
         }
       }
     },
@@ -846,7 +1030,8 @@
             ],
             "setting_keys": [
               "auth"
-            ]
+            ],
+            "dataobjects": []
           },
           "refresh": {
             "type": "GET",
@@ -857,7 +1042,8 @@
             ],
             "setting_keys": [
               "auth"
-            ]
+            ],
+            "dataobjects": []
           },
           "logout": {
             "type": "GET",
@@ -868,7 +1054,8 @@
             ],
             "setting_keys": [
               "auth"
-            ]
+            ],
+            "dataobjects": []
           },
           "decode": {
             "type": "GET",
@@ -877,7 +1064,8 @@
             "auth": [
               "Bearer"
             ],
-            "setting_keys": []
+            "setting_keys": [],
+            "dataobjects": []
           }
         },
         "server": {
@@ -901,7 +1089,7 @@
             ]
           },
           "api": {},
-          "engine_version": "${APPS_API_VERSION}.2"
+          "engine_version": "${ENGINE_VERSION}"
         },
         "plugins": [],
         "settings": {
@@ -1024,7 +1212,8 @@
           "auth": [
             "Bearer"
           ],
-          "setting_keys": []
+          "setting_keys": [],
+          "dataobjects": []
         }
       }
     }
diff --git a/plugins/ops/config-manager/test/integration/test_runtime_apps_config.py b/plugins/ops/config-manager/test/integration/test_runtime_apps_config.py
index cbd52307..3b0ed9b0 100644
--- a/plugins/ops/config-manager/test/integration/test_runtime_apps_config.py
+++ b/plugins/ops/config-manager/test/integration/test_runtime_apps_config.py
@@ -19,6 +19,9 @@ async def test_runtime_apps_config(monkeypatch, runtime_apps_response):
 
     plugin_config = config('plugins/ops/config-manager/config/plugin-config.json')
     result = await execute_event(app_config=plugin_config, event_name="runtime_apps_config", payload=None)
+
+    import json
+    print(json.dumps(result.to_dict()))
     assert result == runtime_apps_response
 
 
diff --git a/plugins/storage/fs/test/integration/__init__.py b/plugins/storage/fs/test/integration/__init__.py
new file mode 100644
index 00000000..3614512c
--- /dev/null
+++ b/plugins/storage/fs/test/integration/__init__.py
@@ -0,0 +1,11 @@
+from dataclasses import dataclass
+from datetime import datetime
+from hopeit.dataobjects import dataobject
+
+
+@dataobject(event_id="object_id", event_ts="object_ts")
+@dataclass
+class MyObject:
+    object_id: str
+    object_ts: datetime
+    value: int
diff --git a/plugins/storage/fs/test/integration/app_config.json b/plugins/storage/fs/test/integration/app_config.json
new file mode 100644
index 00000000..b68ec0e1
--- /dev/null
+++ b/plugins/storage/fs/test/integration/app_config.json
@@ -0,0 +1,32 @@
+{
+    "app": {
+        "name": "test",
+        "version": "test"
+    },
+    "events": {
+        "test_stream_batch_storage": {
+            "type": "STREAM",
+            "plug_mode": "Standalone",
+            "impl": "hopeit.fs_storage.events.stream_batch_storage",
+            "connections": [],
+            "read_stream": {
+                "name": "test_stream",
+                "consumer_group": "test",
+                "queues": [
+                    "AUTO"
+                ]
+            },
+            "dataobjects": [
+                "TestObject"
+            ]
+        }
+    },
+    "settings": {
+        "test_stream_batch_storage": {
+            "path": "/tmp/hopeit/test_stream_batch_storage",
+            "partition_dateformat": "%Y/%m/%d/%H/",
+            "flush_seconds": 60,
+            "flush_max_size": 2
+        }
+    }
+}
\ No newline at end of file
diff --git a/plugins/storage/fs/test/integration/conftest.py b/plugins/storage/fs/test/integration/conftest.py
new file mode 100644
index 00000000..5ce93b48
--- /dev/null
+++ b/plugins/storage/fs/test/integration/conftest.py
@@ -0,0 +1,29 @@
+from datetime import datetime, timezone
+import uuid
+import pytest  # type: ignore
+
+from hopeit.testing.apps import config
+
+from . import MyObject
+
+
+@pytest.fixture
+def test_objs():
+    # Creates 10 elements, in 5 partitions
+    return [
+        MyObject(
+            object_id=f"obj{i}",
+            object_ts=datetime.strptime(
+                f"2020-05-01T0{i//2}:00:00+00:00", "%Y-%m-%dT%H:%M:%S%z"
+            ).astimezone(tz=timezone.utc),
+            value=i
+        ) for i in range(10)
+    ]
+
+
+@pytest.fixture
+def app_config():
+    cfg = config("plugins/storage/fs/test/integration/app_config.json")
+    # Set unique folder for each test run
+    cfg.settings["test_stream_batch_storage"]["path"] += f"/{str(uuid.uuid4())}"
+    return cfg
diff --git a/plugins/storage/fs/test/integration/test_it_stream_batch_storage.py b/plugins/storage/fs/test/integration/test_it_stream_batch_storage.py
new file mode 100644
index 00000000..fba524f8
--- /dev/null
+++ b/plugins/storage/fs/test/integration/test_it_stream_batch_storage.py
@@ -0,0 +1,73 @@
+from glob import glob
+import asyncio
+from hopeit.fs_storage.events.stream_batch_storage import FlushSignal
+
+import pytest  # type: ignore
+
+from hopeit.testing.apps import execute_event
+from hopeit.dataobjects.payload import Payload
+
+from . import MyObject
+
+
+@pytest.mark.asyncio
+async def test_buffer_objects_and_flush_partitions(app_config, test_objs):  # noqa: F811
+    test_save_path = app_config.settings["test_stream_batch_storage"]["path"]
+
+    # Buffer 10 objects that should create 5 partitions
+    for test_obj in test_objs:
+        result = await execute_event(
+            app_config=app_config,
+            event_name='test_stream_batch_storage',
+            payload=test_obj
+        )
+        assert result is None
+
+    await asyncio.sleep(1)  # Allow aiofiles to save
+
+    # Load saved data from disk and compare to input
+    saved_objects = {}
+    for file_name in glob(f'{test_save_path}/2020/05/01/**/*.jsonlines'):
+        with open(file_name) as f:
+            for line in f:
+                obj = Payload.from_json(line, datatype=MyObject)
+                saved_objects[obj.object_id] = obj
+    assert len(saved_objects) == len(test_objs)
+    for obj in test_objs:
+        saved = saved_objects[obj.object_id]
+        assert obj == saved
+
+
+@pytest.mark.asyncio
+async def test_buffer_object_and_flush_signal(app_config, test_objs):  # noqa: F811
+    test_save_path = app_config.settings["test_stream_batch_storage"]["path"]
+
+    # Buffer single object
+    test_obj = test_objs[0]
+    result = await execute_event(
+        app_config=app_config,
+        event_name='test_stream_batch_storage',
+        payload=test_obj
+    )
+    assert result is None
+
+    # Send flush partition signal to force flush single object
+    partition_key = test_obj.object_ts.strftime("%Y/%m/%d/%H") + '/'
+    signal = FlushSignal(partition_key)
+    result = await execute_event(
+        app_config=app_config,
+        event_name='test_stream_batch_storage',
+        payload=signal
+    )
+    assert result is None
+
+    # Load saved object and check is correct
+    await asyncio.sleep(1)  # Allow aiofiles to save
+    saved_objects = {}
+    for file_name in glob(f'{test_save_path}/{partition_key}/*.jsonlines'):
+        with open(file_name) as f:
+            for line in f:
+                obj = Payload.from_json(line, datatype=MyObject)
+                saved_objects[obj.object_id] = obj
+    assert len(saved_objects) == 1
+    assert test_obj == saved_objects[test_obj.object_id]
diff --git a/plugins/storage/fs/test/unit/test_fs.py b/plugins/storage/fs/test/unit/test_fs.py
index d0d481df..73ac2bb5 100644
--- a/plugins/storage/fs/test/unit/test_fs.py
+++ b/plugins/storage/fs/test/unit/test_fs.py
@@ -1,4 +1,5 @@
-from typing import List
+from datetime import datetime, timezone
+from typing import List, Optional
 
 import os
 from dataclasses import dataclass
@@ -7,21 +8,30 @@
 
 import hopeit.fs_storage as fs_module
 from hopeit.dataobjects import dataobject
-from hopeit.fs_storage import FileStorage
+from hopeit.fs_storage import FileStorage, ItemLocator
 import pytest  # type: ignore
 
 
-@dataobject
+@dataobject(event_ts="ts")
 @dataclass
 class FsMockData:
     test: str
+    ts: Optional[datetime] = None
 
 
 payload_str = """{"test":"test_fs"}"""
-test_fs = FsMockData(test='test_fs')
+test_fs = FsMockData(
+    test='test_fs'
+)
 
+payload_str_with_ts = """{"test":"test_fs", "ts":"2022-03-01T00:00:00+00:00"}"""
+test_fs_with_ts = FsMockData(
+    test='test_fs',
+    ts=datetime(2022, 3, 1, tzinfo=timezone.utc)
+)
 
-async def save_and_load_file(payload):
+
+async def save_and_load_file():
     key = "VALIDFILE"
     fs = FileStorage(path=f"/tmp/{key}/")
     path = await fs.store(key, test_fs)
@@ -31,6 +41,27 @@ async def save_and_load_file(payload):
     assert type(loaded) is FsMockData
 
 
+async def save_and_load_file_in_partition():
+    key = "VALIDFILEWITHTS"
+    fs = FileStorage(path=f"/tmp/{key}/", partition_dateformat="%Y/%m/%d")
+    path = await fs.store(key, test_fs_with_ts)
+    assert path == f'/tmp/{key}/2022/03/01/{key}.json'
+    loaded = await fs.get(key, datatype=FsMockData, partition_key="2022/03/01")
+    assert loaded == test_fs_with_ts
+    assert type(loaded) is FsMockData
+
+
+async def save_and_load_file_in_partition_default_ts():
+    key = "VALIDFILE"
+    fs = FileStorage(path=f"/tmp/{key}/", partition_dateformat="%Y/%m/%d")
+    path = await fs.store(key, test_fs)
+    partition_key = datetime.now(tz=timezone.utc).strftime("%Y/%m/%d")
+    assert path == f'/tmp/{key}/{partition_key}/{key}.json'
+    loaded = await fs.get(key, datatype=FsMockData, partition_key=partition_key)
+    assert loaded == test_fs
+    assert type(loaded) is FsMockData
+
+
 async def load_missing_file():
     key = "FILENOTFOUND"
     fs = FileStorage(path=f"/tmp/{key}/")
@@ -75,6 +106,8 @@ def open(path, mode='r'):
         assert mode
         if "FILENOTFOUND" in str(path):
             raise FileNotFoundError(str(path))
+        elif "VALIDFILEWITHTS" in str(path):
+            return MockFile(payload_str_with_ts)
         elif "VALIDFILE" in str(path):
             return MockFile(payload_str)
         else:
@@ -106,13 +139,45 @@ def mock_glob(wc: str) -> List[str]:
     raise ValueError(f"glob received unexpected wildcard: {wc}")
 
 
+def mock_glob_partitions(wc: str) -> List[str]:
+    if wc == '/path/*.json':
+        return []
+    elif wc == '/path/2022/03/01/*.json':
+        return [
+            "/path/2022/03/01/1.json",
+            "/path/2022/03/01/2.json",
+            "/path/2022/03/01/3.json"
+        ]
+    elif wc == '/path/**/**/**/1*.json':
+        return ["/path/2022/03/01/1.json"]
+    raise ValueError(f"glob received unexpected wildcard: {wc}")
+
+
+@pytest.mark.asyncio
+async def test_save_load_file(monkeypatch):
+    monkeypatch.setattr(aiofiles, 'open', MockFile.open)
+    monkeypatch.setattr(os, 'makedirs', MockOs.makedirs)
+    monkeypatch.setattr(os.path, 'exists', MockPath.exists)
+    monkeypatch.setattr(os, 'rename', MockOs.rename)
+    await save_and_load_file()
+
+
 @pytest.mark.asyncio
-async def test_load_file(monkeypatch):
+async def test_save_load_file_in_partition(monkeypatch):
     monkeypatch.setattr(aiofiles, 'open', MockFile.open)
     monkeypatch.setattr(os, 'makedirs', MockOs.makedirs)
     monkeypatch.setattr(os.path, 'exists', MockPath.exists)
     monkeypatch.setattr(os, 'rename', MockOs.rename)
-    await save_and_load_file(payload_str)
+    await save_and_load_file_in_partition()
+
+
+@pytest.mark.asyncio
+async def test_save_load_file_in_partition_default_ts(monkeypatch):
+    monkeypatch.setattr(aiofiles, 'open', MockFile.open)
+    monkeypatch.setattr(os, 'makedirs', MockOs.makedirs)
+    monkeypatch.setattr(os.path, 'exists', MockPath.exists)
+    monkeypatch.setattr(os, 'rename', MockOs.rename)
+    await save_and_load_file_in_partition_default_ts()
 
 
 @pytest.mark.asyncio
@@ -128,7 +193,35 @@ async def test_list_objects(monkeypatch):
     fs = FileStorage(path='/path')
 
     files = await fs.list_objects()
-    assert files == ['1', '2', '3']
+    assert files == [
+        ItemLocator('1', None),
+        ItemLocator('2', None),
+        ItemLocator('3', None),
+    ]
 
     files = await fs.list_objects(wildcard="1*")
-    assert files == ['1']
+    assert files == [
+        ItemLocator('1', None),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_list_objects_within_partitions(monkeypatch):
+    monkeypatch.setattr(os, 'makedirs', MockOs.makedirs)
+    monkeypatch.setattr(fs_module, 'glob', mock_glob_partitions)
+    fs = FileStorage(path='/path', partition_dateformat="%Y/%m/%d/")
+
+    files = await fs.list_objects()
+    assert files == []
+
+    files = await fs.list_objects(wildcard="2022/03/01/*")
+    assert files == [
+        ItemLocator('1', "2022/03/01"),
+        ItemLocator('2', "2022/03/01"),
+        ItemLocator('3', "2022/03/01"),
+    ]
+
+    files = await fs.list_objects(wildcard="**/**/**/1*")
+    assert files == [
+        ItemLocator('1', "2022/03/01")
+    ]
diff --git a/plugins/storage/fs/test/unit/test_partition.py b/plugins/storage/fs/test/unit/test_partition.py
new file mode 100644
index 00000000..598bb16f
--- /dev/null
+++ b/plugins/storage/fs/test/unit/test_partition.py
@@ -0,0 +1,49 @@
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from hopeit.dataobjects import dataobject
+import hopeit.fs_storage.partition as partition
+
+
+@dataobject(event_id="object_id", event_ts="object_ts")
+@dataclass
+class TimedObject:
+    object_id: str
+    object_ts: datetime
+    value: int
+
+
+@dataobject
+@dataclass
+class UntimedObject:
+    object_id: str
+    value: int
+
+
+def test_get_partition_timed_object():
+    obj = TimedObject(
+        "obj1",
+        datetime.strptime(
+            "2020-05-01T00:00:00+00:00", "%Y-%m-%dT%H:%M:%S%z"
+        ).astimezone(tz=timezone.utc),
+        value=1
+    )
+    partition_key = partition.get_partition_key(obj, "%Y/%m/%d/%H")
+    assert partition_key == "2020/05/01/00/"
+
+
+class MockDatetime:
+    @staticmethod
+    def now(*args, **kwargs):
+        return datetime.strptime(
+            "2021-06-02T01:00:00+00:00", "%Y-%m-%dT%H:%M:%S%z"
+        ).astimezone(tz=timezone.utc)
+
+
+def test_get_partition_current_time(monkeypatch):
+    monkeypatch.setattr(partition, 'datetime', MockDatetime)
+    obj = UntimedObject(
+        object_id="obj2",
+        value=2
+    )
+    partition_key = partition.get_partition_key(obj, "%Y/%m/%d/%H")
+    assert partition_key == "2021/06/02/01/"
diff --git a/plugins/streams/redis/test/unit/test_redis_streams.py b/plugins/streams/redis/test/unit/test_redis_streams.py
index 9eec1f37..ebac5873 100644
--- a/plugins/streams/redis/test/unit/test_redis_streams.py
+++ b/plugins/streams/redis/test/unit/test_redis_streams.py
@@ -50,7 +50,7 @@ async def write_stream():
     written_fields = mgr._write_pool.xadd_fields
     assert written_fields == {
         'id': 'test_value',
-        'type': 'MockData',
+        'type': 'unit.test_redis_streams.MockData',
         'submit_ts': written_fields['submit_ts'],
         'event_ts': '1970-01-01T00:00:00+00:00',
         'track.operation_id': 'test_operation_id',
@@ -97,7 +97,7 @@ async def read_stream():
     for stream_event in (await mgr.read_stream(
             stream_name='test_stream',
             consumer_group='test_group',
-            datatypes={'MockData': MockData},
+            datatypes={'unit.test_redis_streams.MockData': MockData},
             track_headers=MockEventHandler.test_track_ids.keys(),
             offset='>',
             batch_size=10,
@@ -202,7 +202,7 @@ class MockRedisPool():
     test_url: str = 'redis://test_url'
     message_count = 10
     test_msg = [b'0000000000-0', {
-        b'type': b'MockData',
+        b'type': b'unit.test_redis_streams.MockData',
         b'track.request_id': b'test_request_id',
         b'track.request_ts': b'2020-02-05T17:07:37.771396+00:00',
         b'track.session_id': b'test_session_id',