Listener code

configs/watchdog_listener_config.toml

@@ -0,0 +1,62 @@
+# =======================================================
+# Config for the Watchdog Listener
+# =======================================================
+
+[parser]
+
+[parser.class]
+type = "dynamic"
+name = "LocalToS3BatchListener"
+parts = [
+  ["listeners.watchdog_listener", "WatchdogListenerMixin"],
+  ["uploaders.local_to_s3", "S3UploaderMixin"]
+]
+
+
+[parser.init]
+# The directory where the state file ('upload_state.json') will be stored.
+state_path = "path/to/state/file"
+
+[parser.init.source]
+# The local directory that the listener will monitor for new files.
+path = "path/to/local/dir"
+#File name patterns to include/exclude.
+include_patterns = [".*\\.csv$", ".*\\.txt$"]
+exclude_patterns = [".*\\.tmp$", "^.*/\\~\\$.*"]
+#Batch configuration
+# Trigger the pipeline after 5 files have accumulated...
+batch_max_size = 5
+# ...or after 60 seconds have passed since the first file arrived,
+# whichever comes first.
+batch_max_latency_seconds = 60
+
+[parser.init.target]
+# Settings for the S3 uploader.
+bucket = "bucket-name" # Your S3 bucket name
+prefix = "raw_data/"
+# Set to 'true' to upload files that have been modified.
+allow-overwrite = true
+
+
+
+# Enable logging to a local file using the python's built-in logging.
+[logging.file]
+filepath = "path/to/log/file.txt"
+level = 0             # level of events to include in this logger
+max_size = 21048576   # Max size, in bytes, of a log file before it is rotated out
+max_files = 5         # Max number of files to keep before deleting the oldest
+
+# Enable logging to Sentry using the Sentry SDK API
+# [logging.sentry]
+# dsn = "placeholder"
+# level = 10
+# event_level = 30
+
+[dependencies]
+# Required dependencies for this specific parser configuration.
+# The `prepare.py` script will install these.
+pip = [
+  "boto3",
+  "watchdog"
+]
+

listen.py

@@ -0,0 +1,49 @@
+"""
+Entry-point script to run a parser in continuous listener mode.
+This is equivalent to: python run.py config.toml --mode listen
+"""
+
+import argparse
+
+# Reuse the existing functions from run.py to load the config and build the parser.
+from run import load_config, load_parser
+
+
+if __name__ == '__main__':
+    # 1. Set up argument parsing to accept the path to a config file.
+    arg_parser = argparse.ArgumentParser(
+        description="Run a data-net-source parser in continuous listener mode."
+    )
+    arg_parser.add_argument(
+        'config_file',
+        type=str,
+        help='Path to the .toml config file that specifies the parser to run.'
+    )
+    args = arg_parser.parse_args()
+
+    # 2. Load the specified configuration file using the helper from run.py.
+    config = load_config(args.config_file)
+
+    # 3. Dynamically build the parser object from the mixins defined in the config.
+    parser = load_parser(config['parser'])
+
+    # 4. Set up logging as defined in the config, if present.
+    if 'logging' in config:
+        parser.make_loggers(config['logging'])
+
+    # 5. Call the .listen() method to start the long-running, event-driven process.
+    try:
+        parser.listen()
+    except AttributeError as e:
+        if 'listen' in str(e):
+            # Provide a helpful error if the user tries to run a checker-based config
+            # with this script, as it will not have a .listen() method.
+            parser.error(
+                f"The configured parser '{parser.__class__.__name__}' does not have listener capabilities. "
+                f"This config appears to use checker mixins instead of listener mixins. "
+                f"Try using 'python run.py {args.config_file}' for checker mode."
+            )
+        else:
+            parser.error(f"Parser setup failed: {e}", exc_info=True)
+    except Exception as e:
+        parser.error(f"The listener exited with an unexpected error: {e}", exc_info=True)

run.py

@@ -12,6 +12,7 @@
 
 
 def load_config(config_fp: Union[str, PathLike]) -> dict:
+
     with open(config_fp, 'r') as f:
         toml_config = toml.load(f)
     return toml_config
@@ -106,7 +107,9 @@ def apply_overrides(config: dict, overrides: List[str], allow_overwrite: bool =
 
 
 if __name__ == '__main__':
-    arg_parser = argparse.ArgumentParser()
+    arg_parser = argparse.ArgumentParser(
+        description="Run a data-net-source parser in batch checker mode."
+    )
     arg_parser.add_argument('config_file', type=str,
                             help='Path to the config file that specifies the parser to run')
     arg_parser.add_argument(
@@ -138,4 +141,14 @@ def apply_overrides(config: dict, overrides: List[str], allow_overwrite: bool =
     if 'logging' in config:
         parser.make_loggers(config['logging'])
 
-    parser.process()
+    # Checker mode (batch processing)
+    try:
+        parser.process()
+    except AttributeError as e:
+        parser.error(
+            f"Parser '{parser.__class__.__name__}' does not support checker mode. "
+            f"This config appears to use listener mixins instead of checker mixins. "
+            f"Try using 'python listen.py {args.config_file}' for continuous listener mode."
+        )
+    except Exception as e:
+        parser.error(f"Checker mode failed: {e}", exc_info=True)

source/common.py

@@ -64,6 +64,7 @@ def __init__(self, state_path, source=None, middle=None, target=None):
             self.target_location = target
 
     def process(self):
+
         self.start_notify()
         try:
             to_do = self.check()
@@ -87,6 +88,41 @@ def process(self):
         else:
             self.end_notify(0)
 
+    def listen(self):
+        """
+        Execute the parser in continuous listener mode.
+        This is for event-driven, continuous operation.
+
+        This orchestrates the listener lifecycle.
+        """
+        self.start_notify()
+        try:
+            # Setup phase - prepare batching and observer
+            self.setup_batching()
+            self.setup_observer()
+            self.start_observer()
+
+            # Main listening loop - runs continuously
+            self.info(f"Listener is now monitoring for events...")
+            self.run_observer_loop()
+
+        except KeyboardInterrupt:
+            self.warning("Listener stopped by user. Processing any remaining files...")
+            # Process any remaining files in the batch
+            if hasattr(self, '_process_batch'):
+                self.process_batch()
+            self.end_notify(0)
+        except Exception as e:
+            import sys, traceback
+            self.error(e)
+            self.error(f'Encountered {str(e)}')
+            self.debug(traceback.format_exception(*sys.exc_info()))
+            self.end_notify(1)
+        finally:
+            # Cleanup phase - stop observer
+            if hasattr(self, 'stop_observer'):
+                self.stop_observer()
+
     def check(self):
         """
         Check should look for new data that needs to be uploaded