From 832fa13ed8977e3ab998dd6cb42a78115c6c9871 Mon Sep 17 00:00:00 2001
From: Christian Chwala <christian.chwala@kit.edu>
Date: Mon, 13 Jul 2026 20:33:01 +0200
Subject: [PATCH 1/2] feat(parser): add entrypoint infrastructure for multiple
 stream types
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Split parser into pluggable entrypoints to support different ingestion loops:

- parser/entrypoints/sftp_push — current logic (watch local directory)
- parser/entrypoints/sftp_pull — stub for future SFTP polling
- parser/entrypoints/api_pull — stub for future HTTP/API polling

Changes:
- parser/main.py becomes thin shim delegating to sftp_push
- generate_config.py emits 'command:' based on new 'entrypoint' field in users.yml
- users.yml schema extended with 'entrypoint' documentation
- Tests updated to import from parser.entrypoints.sftp_push

Backward compatible: existing services default to 'sftp_push' entrypoint.
---
 docker-compose.override.yml     |   4 +
 parser/entrypoints/__init__.py  |   0
 parser/entrypoints/api_pull.py  |   6 ++
 parser/entrypoints/sftp_pull.py |   6 ++
 parser/entrypoints/sftp_push.py | 182 ++++++++++++++++++++++++++++++++
 parser/main.py                  | 181 +------------------------------
 parser/tests/test_main.py       |  80 +++++++-------
 scripts/generate_config.py      |  10 ++
 users.yml                       |   4 +
 9 files changed, 257 insertions(+), 216 deletions(-)
 create mode 100644 parser/entrypoints/__init__.py
 create mode 100644 parser/entrypoints/api_pull.py
 create mode 100644 parser/entrypoints/sftp_pull.py
 create mode 100644 parser/entrypoints/sftp_push.py

diff --git a/docker-compose.override.yml b/docker-compose.override.yml
index dfec13e..108a983 100644
--- a/docker-compose.override.yml
+++ b/docker-compose.override.yml
@@ -22,6 +22,7 @@ services:
     environment:
       - DATABASE_URL=postgresql://demo_openmrg:demo_openmrg_password@database:5432/mydatabase
       - USER_ID=demo_openmrg
+      - PARSER_TYPE=demo_csv_data
       - PARSER_INCOMING_DIR=/app/data/incoming
       - PARSER_ARCHIVED_DIR=/app/data/archived
       - PARSER_QUARANTINE_DIR=/app/data/quarantine
@@ -33,6 +34,7 @@ services:
       - parser_demo_openmrg_main_archived:/app/data/archived
       - parser_demo_openmrg_main_quarantine:/app/data/quarantine
     restart: unless-stopped
+    command: python -m parser.entrypoints.sftp_push
 
   parser_demo_orange_cameroun_douala:
     build: ./parser
@@ -42,6 +44,7 @@ services:
     environment:
       - DATABASE_URL=postgresql://demo_orange_cameroun:demo_orange_cameroun_password@database:5432/mydatabase
       - USER_ID=demo_orange_cameroun
+      - PARSER_TYPE=demo_csv_data
       - PARSER_INCOMING_DIR=/app/data/incoming
       - PARSER_ARCHIVED_DIR=/app/data/archived
       - PARSER_QUARANTINE_DIR=/app/data/quarantine
@@ -53,6 +56,7 @@ services:
       - parser_demo_orange_cameroun_douala_archived:/app/data/archived
       - parser_demo_orange_cameroun_douala_quarantine:/app/data/quarantine
     restart: unless-stopped
+    command: python -m parser.entrypoints.sftp_push
 
 volumes:
   sftp_demo_openmrg_main:
diff --git a/parser/entrypoints/__init__.py b/parser/entrypoints/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/parser/entrypoints/api_pull.py b/parser/entrypoints/api_pull.py
new file mode 100644
index 0000000..ccc1620
--- /dev/null
+++ b/parser/entrypoints/api_pull.py
@@ -0,0 +1,6 @@
+"""API pull entrypoint — polls an HTTP endpoint and writes JSON data.
+
+Not yet implemented. Placeholder so the module exists for import checking.
+"""
+
+raise NotImplementedError("api_pull entrypoint is not yet implemented")
diff --git a/parser/entrypoints/sftp_pull.py b/parser/entrypoints/sftp_pull.py
new file mode 100644
index 0000000..47a3c5f
--- /dev/null
+++ b/parser/entrypoints/sftp_pull.py
@@ -0,0 +1,6 @@
+"""SFTP pull entrypoint — polls a remote SFTP server for new files.
+
+Not yet implemented. Placeholder so the module exists for import checking.
+"""
+
+raise NotImplementedError("sftp_pull entrypoint is not yet implemented")
diff --git a/parser/entrypoints/sftp_push.py b/parser/entrypoints/sftp_push.py
new file mode 100644
index 0000000..ebf9bbf
--- /dev/null
+++ b/parser/entrypoints/sftp_push.py
@@ -0,0 +1,182 @@
+"""Parser service entrypoint and orchestration.
+
+This module wires together the FileWatcher, DBWriter and FileManager to implement the parser service. It is intentionally lightweight and delegates parsing logic to function-based parsers in `parsers/demo_csv_data/`.
+"""
+
+import json
+import os
+import time
+import logging
+import threading
+from pathlib import Path
+
+from ..file_watcher import FileWatcher
+from ..file_manager import FileManager
+from ..db_writer import DBWriter
+from ..service_logic import (
+    load_parser,
+    load_api_json_bundle,
+    process_cml_file,
+    process_rawdata_files_batch,
+    _make_default_bundle,
+)
+
+
+class Config:
+    DATABASE_URL = os.getenv(
+        "DATABASE_URL", "postgresql://myuser:mypassword@database:5432/mydatabase"
+    )
+    USER_ID = os.getenv("USER_ID", "demo_openmrg")
+    PARSER_TYPE = os.getenv("PARSER_TYPE", "demo_csv_data")
+    PARSER_CSV_CONFIG: dict = json.loads(os.getenv("PARSER_CSV_CONFIG", "{}"))
+    INCOMING_DIR = Path(os.getenv("PARSER_INCOMING_DIR", "data/incoming"))
+    ARCHIVED_DIR = Path(os.getenv("PARSER_ARCHIVED_DIR", "data/archived"))
+    QUARANTINE_DIR = Path(os.getenv("PARSER_QUARANTINE_DIR", "data/quarantine"))
+    PARSER_ENABLED = os.getenv("PARSER_ENABLED", "True").lower() in ("1", "true", "yes")
+    PROCESS_EXISTING_ON_STARTUP = os.getenv(
+        "PROCESS_EXISTING_ON_STARTUP", "True"
+    ).lower() in ("1", "true", "yes")
+    LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
+    # How often (seconds) to recalculate aggregate CML stats in the background
+    STATS_REFRESH_INTERVAL = int(os.getenv("STATS_REFRESH_INTERVAL", "60"))
+
+
+def setup_logging():
+    logging.basicConfig(
+        level=getattr(logging, Config.LOG_LEVEL),
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    )
+
+
+def process_existing_files(db_writer, file_manager, logger, parser=None):
+    incoming = sorted(f for f in Config.INCOMING_DIR.glob("*.csv") if f.is_file())
+
+    _parser = parser if parser is not None else _make_default_bundle()
+
+    metadata_files = [f for f in incoming if _parser.is_metadata_file(f.name.lower())]
+    data_files = [f for f in incoming if not _parser.is_metadata_file(f.name.lower())]
+
+    # Metadata files: process individually (typically just one)
+    for f in metadata_files:
+        try:
+            process_cml_file(f, db_writer, file_manager, logger, parser=_parser)
+        except Exception:
+            pass
+
+    # Data files: batch-process for efficiency
+    if data_files:
+        logger.info("Found %d data file(s) to process", len(data_files))
+        process_rawdata_files_batch(
+            data_files, db_writer, file_manager, logger, parser=_parser
+        )
+
+    # JSON files (from api_fetcher): process individually
+    json_files = sorted(f for f in Config.INCOMING_DIR.glob("*.json") if f.is_file())
+    for f in json_files:
+        try:
+            process_cml_file(f, db_writer, file_manager, logger, parser=_parser)
+        except Exception:
+            pass
+
+
+def main():
+    setup_logging()
+    logger = logging.getLogger("parser.service")
+    file_manager = FileManager(
+        str(Config.INCOMING_DIR),
+        str(Config.ARCHIVED_DIR),
+        str(Config.QUARANTINE_DIR),
+    )
+    db_writer = DBWriter(Config.DATABASE_URL, user_id=Config.USER_ID)
+
+    # Select parser bundle based on PARSER_TYPE
+    if Config.PARSER_TYPE == "api_json":
+        from .service_logic import load_api_json_bundle
+
+        parser_bundle = load_api_json_bundle()
+    else:
+        parser_bundle = load_parser(Config.PARSER_TYPE, Config.PARSER_CSV_CONFIG)
+
+    logger.info("Starting parser service (parser_type=%s)", Config.PARSER_TYPE)
+    Config.INCOMING_DIR.mkdir(parents=True, exist_ok=True)
+    Config.ARCHIVED_DIR.mkdir(parents=True, exist_ok=True)
+    Config.QUARANTINE_DIR.mkdir(parents=True, exist_ok=True)
+
+    if not Config.PARSER_ENABLED:
+        logger.warning("Parser is disabled via configuration. Exiting.")
+        return
+
+    try:
+        db_writer.connect()
+    except Exception:
+        logger.exception("Unable to connect to DB at startup")
+
+    if Config.PROCESS_EXISTING_ON_STARTUP:
+        process_existing_files(db_writer, file_manager, logger, parser=parser_bundle)
+
+    def on_new_file(filepath):
+        try:
+            process_cml_file(
+                filepath, db_writer, file_manager, logger, parser=parser_bundle
+            )
+        except Exception:
+            pass
+
+    watcher = FileWatcher(
+        str(Config.INCOMING_DIR),
+        on_new_file,
+        {".csv", ".json"},
+    )
+    watcher.start()
+
+    # Background thread: refresh cml_stats on a slow timer so it never
+    # blocks file processing.
+    stop_event = threading.Event()
+
+    def stats_loop():
+        # Use a separate DBWriter connection so stats queries don't contend
+        # with the insert connection.
+        stats_db = DBWriter(Config.DATABASE_URL, user_id=Config.USER_ID)
+
+        # Keep retrying until the DB is reachable (e.g. if it starts slowly).
+        while not stop_event.is_set():
+            try:
+                stats_db.connect()
+                break
+            except Exception:
+                logger.warning("Stats thread: DB not ready, retrying in 5s...")
+                stop_event.wait(5)
+        if stop_event.is_set():
+            return
+
+        # Run immediately on startup so Grafana has fresh stats without
+        # waiting a full interval after the backlog is processed.
+        try:
+            stats_db.refresh_windowed_stats()
+        except Exception:
+            logger.exception("Stats thread: initial refresh_windowed_stats failed")
+        while not stop_event.wait(Config.STATS_REFRESH_INTERVAL):
+            try:
+                stats_db.refresh_windowed_stats()
+            except Exception:
+                logger.exception("Stats thread: refresh_windowed_stats failed")
+        stats_db.close()
+
+    stats_thread = threading.Thread(
+        target=stats_loop, daemon=True, name="stats-refresh"
+    )
+    stats_thread.start()
+
+    try:
+        while True:
+            time.sleep(1)
+    except KeyboardInterrupt:
+        logger.info("Shutting down parser service")
+    finally:
+        stop_event.set()
+        watcher.stop()
+        db_writer.close()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/parser/main.py b/parser/main.py
index 8cb9418..f504828 100644
--- a/parser/main.py
+++ b/parser/main.py
@@ -1,182 +1,11 @@
-"""Parser service entrypoint and orchestration.
+"""Parser service entrypoint shim.
 
-This module wires together the FileWatcher, DBWriter and FileManager to implement the parser service. It is intentionally lightweight and delegates parsing logic to function-based parsers in `parsers/demo_csv_data/`.
+This module delegates to the appropriate entrypoint script based on
+the PARSER_ENTRYPOINT environment variable. For backward compatibility,
+if not set it defaults to running the sftp_push entrypoint.
 """
 
-import json
-import os
-import time
-import logging
-import threading
-from pathlib import Path
-
-from .file_watcher import FileWatcher
-from .file_manager import FileManager
-from .db_writer import DBWriter
-from .service_logic import (
-    load_parser,
-    load_api_json_bundle,
-    process_cml_file,
-    process_rawdata_files_batch,
-    _make_default_bundle,
-)
-
-
-class Config:
-    DATABASE_URL = os.getenv(
-        "DATABASE_URL", "postgresql://myuser:mypassword@database:5432/mydatabase"
-    )
-    USER_ID = os.getenv("USER_ID", "demo_openmrg")
-    PARSER_TYPE = os.getenv("PARSER_TYPE", "demo_csv_data")
-    PARSER_CSV_CONFIG: dict = json.loads(os.getenv("PARSER_CSV_CONFIG", "{}"))
-    INCOMING_DIR = Path(os.getenv("PARSER_INCOMING_DIR", "data/incoming"))
-    ARCHIVED_DIR = Path(os.getenv("PARSER_ARCHIVED_DIR", "data/archived"))
-    QUARANTINE_DIR = Path(os.getenv("PARSER_QUARANTINE_DIR", "data/quarantine"))
-    PARSER_ENABLED = os.getenv("PARSER_ENABLED", "True").lower() in ("1", "true", "yes")
-    PROCESS_EXISTING_ON_STARTUP = os.getenv(
-        "PROCESS_EXISTING_ON_STARTUP", "True"
-    ).lower() in ("1", "true", "yes")
-    LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
-    # How often (seconds) to recalculate aggregate CML stats in the background
-    STATS_REFRESH_INTERVAL = int(os.getenv("STATS_REFRESH_INTERVAL", "60"))
-
-
-def setup_logging():
-    logging.basicConfig(
-        level=getattr(logging, Config.LOG_LEVEL),
-        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
-    )
-
-
-def process_existing_files(db_writer, file_manager, logger, parser=None):
-    incoming = sorted(f for f in Config.INCOMING_DIR.glob("*.csv") if f.is_file())
-
-    _parser = parser if parser is not None else _make_default_bundle()
-
-    metadata_files = [f for f in incoming if _parser.is_metadata_file(f.name.lower())]
-    data_files = [f for f in incoming if not _parser.is_metadata_file(f.name.lower())]
-
-    # Metadata files: process individually (typically just one)
-    for f in metadata_files:
-        try:
-            process_cml_file(f, db_writer, file_manager, logger, parser=_parser)
-        except Exception:
-            pass
-
-    # Data files: batch-process for efficiency
-    if data_files:
-        logger.info("Found %d data file(s) to process", len(data_files))
-        process_rawdata_files_batch(
-            data_files, db_writer, file_manager, logger, parser=_parser
-        )
-
-    # JSON files (from api_fetcher): process individually
-    json_files = sorted(f for f in Config.INCOMING_DIR.glob("*.json") if f.is_file())
-    for f in json_files:
-        try:
-            process_cml_file(f, db_writer, file_manager, logger, parser=_parser)
-        except Exception:
-            pass
-
-
-def main():
-    setup_logging()
-    logger = logging.getLogger("parser.service")
-    file_manager = FileManager(
-        str(Config.INCOMING_DIR),
-        str(Config.ARCHIVED_DIR),
-        str(Config.QUARANTINE_DIR),
-    )
-    db_writer = DBWriter(Config.DATABASE_URL, user_id=Config.USER_ID)
-
-    # Select parser bundle based on PARSER_TYPE
-    if Config.PARSER_TYPE == "api_json":
-        from .service_logic import load_api_json_bundle
-
-        parser_bundle = load_api_json_bundle()
-    else:
-        parser_bundle = load_parser(Config.PARSER_TYPE, Config.PARSER_CSV_CONFIG)
-
-    logger.info("Starting parser service (parser_type=%s)", Config.PARSER_TYPE)
-    Config.INCOMING_DIR.mkdir(parents=True, exist_ok=True)
-    Config.ARCHIVED_DIR.mkdir(parents=True, exist_ok=True)
-    Config.QUARANTINE_DIR.mkdir(parents=True, exist_ok=True)
-
-    if not Config.PARSER_ENABLED:
-        logger.warning("Parser is disabled via configuration. Exiting.")
-        return
-
-    try:
-        db_writer.connect()
-    except Exception:
-        logger.exception("Unable to connect to DB at startup")
-
-    if Config.PROCESS_EXISTING_ON_STARTUP:
-        process_existing_files(db_writer, file_manager, logger, parser=parser_bundle)
-
-    def on_new_file(filepath):
-        try:
-            process_cml_file(
-                filepath, db_writer, file_manager, logger, parser=parser_bundle
-            )
-        except Exception:
-            pass
-
-    watcher = FileWatcher(
-        str(Config.INCOMING_DIR),
-        on_new_file,
-        {".csv", ".json"},
-    )
-    watcher.start()
-
-    # Background thread: refresh cml_stats on a slow timer so it never
-    # blocks file processing.
-    stop_event = threading.Event()
-
-    def stats_loop():
-        # Use a separate DBWriter connection so stats queries don't contend
-        # with the insert connection.
-        stats_db = DBWriter(Config.DATABASE_URL, user_id=Config.USER_ID)
-
-        # Keep retrying until the DB is reachable (e.g. if it starts slowly).
-        while not stop_event.is_set():
-            try:
-                stats_db.connect()
-                break
-            except Exception:
-                logger.warning("Stats thread: DB not ready, retrying in 5s...")
-                stop_event.wait(5)
-        if stop_event.is_set():
-            return
-
-        # Run immediately on startup so Grafana has fresh stats without
-        # waiting a full interval after the backlog is processed.
-        try:
-            stats_db.refresh_windowed_stats()
-        except Exception:
-            logger.exception("Stats thread: initial refresh_windowed_stats failed")
-        while not stop_event.wait(Config.STATS_REFRESH_INTERVAL):
-            try:
-                stats_db.refresh_windowed_stats()
-            except Exception:
-                logger.exception("Stats thread: refresh_windowed_stats failed")
-        stats_db.close()
-
-    stats_thread = threading.Thread(
-        target=stats_loop, daemon=True, name="stats-refresh"
-    )
-    stats_thread.start()
-
-    try:
-        while True:
-            time.sleep(1)
-    except KeyboardInterrupt:
-        logger.info("Shutting down parser service")
-    finally:
-        stop_event.set()
-        watcher.stop()
-        db_writer.close()
-
+from parser.entrypoints.sftp_push import main
 
 if __name__ == "__main__":
     main()
diff --git a/parser/tests/test_main.py b/parser/tests/test_main.py
index b64ab6c..22b5250 100644
--- a/parser/tests/test_main.py
+++ b/parser/tests/test_main.py
@@ -6,7 +6,7 @@
 from unittest.mock import MagicMock, patch, call
 
 
-from ..main import process_existing_files, main
+from ..entrypoints.sftp_push import process_existing_files, main
 
 
 @pytest.fixture
@@ -36,7 +36,7 @@ def _write_csv(directory: Path, name: str) -> Path:
 
 
 def _run(tmp_path, db_writer, file_manager, logger):
-    with patch("parser.main.Config.INCOMING_DIR", tmp_path):
+    with patch("parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path):
         process_existing_files(db_writer, file_manager, logger)
 
 
@@ -59,9 +59,9 @@ def test_data_files_only_use_batch_processing(
     """Data files are forwarded to process_rawdata_files_batch."""
     files = [_write_csv(tmp_path, f"raw_data_{i}.csv") for i in range(3)]
 
-    with patch("parser.main.process_rawdata_files_batch") as mock_batch, patch(
-        "parser.main.process_cml_file"
-    ) as mock_single, patch("parser.main.Config.INCOMING_DIR", tmp_path):
+    with patch("parser.entrypoints.sftp_push.process_rawdata_files_batch") as mock_batch, patch(
+        "parser.entrypoints.sftp_push.process_cml_file"
+    ) as mock_single, patch("parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path):
         process_existing_files(mock_db_writer, mock_file_manager, logger)
 
     mock_single.assert_not_called()
@@ -77,9 +77,9 @@ def test_metadata_files_only_use_individual_processing(
     meta1 = _write_csv(tmp_path, "metadata_links.csv")
     meta2 = _write_csv(tmp_path, "meta_extra.csv")
 
-    with patch("parser.main.process_rawdata_files_batch") as mock_batch, patch(
-        "parser.main.process_cml_file"
-    ) as mock_single, patch("parser.main.Config.INCOMING_DIR", tmp_path):
+    with patch("parser.entrypoints.sftp_push.process_rawdata_files_batch") as mock_batch, patch(
+        "parser.entrypoints.sftp_push.process_cml_file"
+    ) as mock_single, patch("parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path):
         process_existing_files(mock_db_writer, mock_file_manager, logger)
 
     assert mock_single.call_count == 2
@@ -94,9 +94,9 @@ def test_mixed_files_routes_to_correct_handlers(
     data1 = _write_csv(tmp_path, "raw_data_001.csv")
     data2 = _write_csv(tmp_path, "raw_data_002.csv")
 
-    with patch("parser.main.process_rawdata_files_batch") as mock_batch, patch(
-        "parser.main.process_cml_file"
-    ) as mock_single, patch("parser.main.Config.INCOMING_DIR", tmp_path):
+    with patch("parser.entrypoints.sftp_push.process_rawdata_files_batch") as mock_batch, patch(
+        "parser.entrypoints.sftp_push.process_cml_file"
+    ) as mock_single, patch("parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path):
         process_existing_files(mock_db_writer, mock_file_manager, logger)
 
     # Check positional args; the branch now also passes parser= as a kwarg
@@ -115,11 +115,11 @@ def test_metadata_exception_is_swallowed(
     data = _write_csv(tmp_path, "raw_data_001.csv")
 
     with patch(
-        "parser.main.process_cml_file", side_effect=Exception("DB down")
+        "parser.entrypoints.sftp_push.process_cml_file", side_effect=Exception("DB down")
     ) as mock_single, patch(
-        "parser.main.process_rawdata_files_batch"
+        "parser.entrypoints.sftp_push.process_rawdata_files_batch"
     ) as mock_batch, patch(
-        "parser.main.Config.INCOMING_DIR", tmp_path
+        "parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path
     ):
         process_existing_files(
             mock_db_writer, mock_file_manager, logger
@@ -152,20 +152,20 @@ def start(self):
 
     mock_db = MagicMock()
 
-    with patch("parser.main.threading.Thread", CapturingThread), \
-         patch("parser.main.threading.Event", return_value=mock_event), \
-         patch("parser.main.FileManager"), \
-         patch("parser.main.FileWatcher"), \
-         patch("parser.main.DBWriter", return_value=mock_db), \
-         patch("parser.main.Config.PARSER_ENABLED", True), \
-         patch("parser.main.Config.PROCESS_EXISTING_ON_STARTUP", False), \
-         patch("parser.main.Config.DATABASE_URL", "postgresql://test"), \
-         patch("parser.main.Config.USER_ID", "test_user"), \
-         patch("parser.main.Config.STATS_REFRESH_INTERVAL", 60), \
-         patch("parser.main.Config.INCOMING_DIR", tmp_path), \
-         patch("parser.main.Config.ARCHIVED_DIR", tmp_path), \
-         patch("parser.main.Config.QUARANTINE_DIR", tmp_path), \
-         patch("parser.main.time.sleep", side_effect=KeyboardInterrupt):
+    with patch("parser.entrypoints.sftp_push.threading.Thread", CapturingThread), \
+         patch("parser.entrypoints.sftp_push.threading.Event", return_value=mock_event), \
+         patch("parser.entrypoints.sftp_push.FileManager"), \
+         patch("parser.entrypoints.sftp_push.FileWatcher"), \
+         patch("parser.entrypoints.sftp_push.DBWriter", return_value=mock_db), \
+         patch("parser.entrypoints.sftp_push.Config.PARSER_ENABLED", True), \
+         patch("parser.entrypoints.sftp_push.Config.PROCESS_EXISTING_ON_STARTUP", False), \
+         patch("parser.entrypoints.sftp_push.Config.DATABASE_URL", "postgresql://test"), \
+         patch("parser.entrypoints.sftp_push.Config.USER_ID", "test_user"), \
+         patch("parser.entrypoints.sftp_push.Config.STATS_REFRESH_INTERVAL", 60), \
+         patch("parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path), \
+         patch("parser.entrypoints.sftp_push.Config.ARCHIVED_DIR", tmp_path), \
+         patch("parser.entrypoints.sftp_push.Config.QUARANTINE_DIR", tmp_path), \
+         patch("parser.entrypoints.sftp_push.time.sleep", side_effect=KeyboardInterrupt):
         try:
             main()
         except (KeyboardInterrupt, SystemExit):
@@ -234,18 +234,18 @@ def __init__(self, *args, target=None, name=None, **kwargs):
         def start(self):
             pass  # don't spawn real threads in unit tests
 
-    with patch("parser.main.threading.Thread", CapturingThread), \
-         patch("parser.main.FileManager"), \
-         patch("parser.main.FileWatcher"), \
-         patch("parser.main.DBWriter") as MockDBWriter, \
-         patch("parser.main.Config.PARSER_ENABLED", True), \
-         patch("parser.main.Config.PROCESS_EXISTING_ON_STARTUP", False), \
-         patch("parser.main.Config.DATABASE_URL", "postgresql://test"), \
-         patch("parser.main.Config.USER_ID", "ctu_cz_tmobile"), \
-         patch("parser.main.Config.INCOMING_DIR", tmp_path), \
-         patch("parser.main.Config.ARCHIVED_DIR", tmp_path), \
-         patch("parser.main.Config.QUARANTINE_DIR", tmp_path), \
-         patch("parser.main.time.sleep", side_effect=KeyboardInterrupt):
+    with patch("parser.entrypoints.sftp_push.threading.Thread", CapturingThread), \
+         patch("parser.entrypoints.sftp_push.FileManager"), \
+         patch("parser.entrypoints.sftp_push.FileWatcher"), \
+         patch("parser.entrypoints.sftp_push.DBWriter") as MockDBWriter, \
+         patch("parser.entrypoints.sftp_push.Config.PARSER_ENABLED", True), \
+         patch("parser.entrypoints.sftp_push.Config.PROCESS_EXISTING_ON_STARTUP", False), \
+         patch("parser.entrypoints.sftp_push.Config.DATABASE_URL", "postgresql://test"), \
+         patch("parser.entrypoints.sftp_push.Config.USER_ID", "ctu_cz_tmobile"), \
+         patch("parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path), \
+         patch("parser.entrypoints.sftp_push.Config.ARCHIVED_DIR", tmp_path), \
+         patch("parser.entrypoints.sftp_push.Config.QUARANTINE_DIR", tmp_path), \
+         patch("parser.entrypoints.sftp_push.time.sleep", side_effect=KeyboardInterrupt):
         try:
             main()
         except (KeyboardInterrupt, SystemExit):
diff --git a/scripts/generate_config.py b/scripts/generate_config.py
index aa35d7a..df4dce1 100644
--- a/scripts/generate_config.py
+++ b/scripts/generate_config.py
@@ -89,6 +89,14 @@ def _validate(users: list[dict]) -> None:
                     f"Source {src['id']!r} for user {uid!r} uses parser "
                     f"'csv_generic' but has no 'csv_config' block"
                 )
+            # Validate entrypoint if specified
+            entrypoint = src.get("entrypoint", "sftp_push")
+            valid_entrypoints = {"sftp_push", "sftp_pull", "api_pull"}
+            if entrypoint not in valid_entrypoints:
+                raise ValueError(
+                    f"Source {src['id']!r} for user {uid!r} has unknown entrypoint "
+                    f"{entrypoint!r}. Valid values: {sorted(valid_entrypoints)}"
+                )
 
 
 # ---------------------------------------------------------------------------
@@ -126,6 +134,7 @@ def _parser_service(user: dict, src: dict) -> str:
     db_url = (
         f"postgresql://{user['id']}:{user['id']}_password" "@database:5432/mydatabase"
     )
+    entrypoint = src.get("entrypoint", "sftp_push")
     env_lines = [
         f"      - DATABASE_URL={db_url}",
         f"      - USER_ID={user['id']}",
@@ -156,6 +165,7 @@ def _parser_service(user: dict, src: dict) -> str:
             f"      - {arch_vol}:/app/data/archived",
             f"      - {quar_vol}:/app/data/quarantine",
             f"    restart: unless-stopped",
+            f"    command: python -m parser.entrypoints.{entrypoint}",
             "",
         ]
     )
diff --git a/users.yml b/users.yml
index 9173585..fbb1697 100644
--- a/users.yml
+++ b/users.yml
@@ -33,6 +33,10 @@
 #                   parser/parsers/<name>/ and add it to load_parser() in
 #                   parser/service_logic.py. This is the recommended approach for
 #                   non-trivial format differences or non-CSV formats (JSON, etc.).
+#     entrypoint  — Which ingestion loop to run for this source (default: "sftp_push").
+#                   "sftp_push" — watch a local directory fed by SFTP.
+#                   "sftp_pull" — poll a remote SFTP server (not yet implemented).
+#                   "api_pull"  — poll an HTTP/JSON API (not yet implemented).
 #     csv_config  — (only for parser: csv_generic) Format configuration:
 #       read_csv_kwargs          — dict of kwargs forwarded to pd.read_csv
 #                                  (e.g. sep, encoding, decimal, skiprows).

From 2e55780d8d55df351f75be34d75ffa5a43b81762 Mon Sep 17 00:00:00 2001
From: Christian Chwala <christian.chwala@kit.edu>
Date: Mon, 13 Jul 2026 20:47:27 +0200
Subject: [PATCH 2/2] refactor: remove unused import and format test code

- Remove unused load_api_json_bundle import from sftp_push.py (imported locally in main() where needed)
- Apply black formatting to test_main.py for consistency
---
 parser/entrypoints/sftp_push.py |   1 -
 parser/tests/test_main.py       | 112 +++++++++++++++++++++-----------
 2 files changed, 74 insertions(+), 39 deletions(-)

diff --git a/parser/entrypoints/sftp_push.py b/parser/entrypoints/sftp_push.py
index ebf9bbf..0398d46 100644
--- a/parser/entrypoints/sftp_push.py
+++ b/parser/entrypoints/sftp_push.py
@@ -15,7 +15,6 @@
 from ..db_writer import DBWriter
 from ..service_logic import (
     load_parser,
-    load_api_json_bundle,
     process_cml_file,
     process_rawdata_files_batch,
     _make_default_bundle,
diff --git a/parser/tests/test_main.py b/parser/tests/test_main.py
index 22b5250..557ce6a 100644
--- a/parser/tests/test_main.py
+++ b/parser/tests/test_main.py
@@ -59,9 +59,13 @@ def test_data_files_only_use_batch_processing(
     """Data files are forwarded to process_rawdata_files_batch."""
     files = [_write_csv(tmp_path, f"raw_data_{i}.csv") for i in range(3)]
 
-    with patch("parser.entrypoints.sftp_push.process_rawdata_files_batch") as mock_batch, patch(
+    with patch(
+        "parser.entrypoints.sftp_push.process_rawdata_files_batch"
+    ) as mock_batch, patch(
         "parser.entrypoints.sftp_push.process_cml_file"
-    ) as mock_single, patch("parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path):
+    ) as mock_single, patch(
+        "parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path
+    ):
         process_existing_files(mock_db_writer, mock_file_manager, logger)
 
     mock_single.assert_not_called()
@@ -77,9 +81,13 @@ def test_metadata_files_only_use_individual_processing(
     meta1 = _write_csv(tmp_path, "metadata_links.csv")
     meta2 = _write_csv(tmp_path, "meta_extra.csv")
 
-    with patch("parser.entrypoints.sftp_push.process_rawdata_files_batch") as mock_batch, patch(
+    with patch(
+        "parser.entrypoints.sftp_push.process_rawdata_files_batch"
+    ) as mock_batch, patch(
         "parser.entrypoints.sftp_push.process_cml_file"
-    ) as mock_single, patch("parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path):
+    ) as mock_single, patch(
+        "parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path
+    ):
         process_existing_files(mock_db_writer, mock_file_manager, logger)
 
     assert mock_single.call_count == 2
@@ -94,9 +102,13 @@ def test_mixed_files_routes_to_correct_handlers(
     data1 = _write_csv(tmp_path, "raw_data_001.csv")
     data2 = _write_csv(tmp_path, "raw_data_002.csv")
 
-    with patch("parser.entrypoints.sftp_push.process_rawdata_files_batch") as mock_batch, patch(
+    with patch(
+        "parser.entrypoints.sftp_push.process_rawdata_files_batch"
+    ) as mock_batch, patch(
         "parser.entrypoints.sftp_push.process_cml_file"
-    ) as mock_single, patch("parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path):
+    ) as mock_single, patch(
+        "parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path
+    ):
         process_existing_files(mock_db_writer, mock_file_manager, logger)
 
     # Check positional args; the branch now also passes parser= as a kwarg
@@ -115,7 +127,8 @@ def test_metadata_exception_is_swallowed(
     data = _write_csv(tmp_path, "raw_data_001.csv")
 
     with patch(
-        "parser.entrypoints.sftp_push.process_cml_file", side_effect=Exception("DB down")
+        "parser.entrypoints.sftp_push.process_cml_file",
+        side_effect=Exception("DB down"),
     ) as mock_single, patch(
         "parser.entrypoints.sftp_push.process_rawdata_files_batch"
     ) as mock_batch, patch(
@@ -152,20 +165,31 @@ def start(self):
 
     mock_db = MagicMock()
 
-    with patch("parser.entrypoints.sftp_push.threading.Thread", CapturingThread), \
-         patch("parser.entrypoints.sftp_push.threading.Event", return_value=mock_event), \
-         patch("parser.entrypoints.sftp_push.FileManager"), \
-         patch("parser.entrypoints.sftp_push.FileWatcher"), \
-         patch("parser.entrypoints.sftp_push.DBWriter", return_value=mock_db), \
-         patch("parser.entrypoints.sftp_push.Config.PARSER_ENABLED", True), \
-         patch("parser.entrypoints.sftp_push.Config.PROCESS_EXISTING_ON_STARTUP", False), \
-         patch("parser.entrypoints.sftp_push.Config.DATABASE_URL", "postgresql://test"), \
-         patch("parser.entrypoints.sftp_push.Config.USER_ID", "test_user"), \
-         patch("parser.entrypoints.sftp_push.Config.STATS_REFRESH_INTERVAL", 60), \
-         patch("parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path), \
-         patch("parser.entrypoints.sftp_push.Config.ARCHIVED_DIR", tmp_path), \
-         patch("parser.entrypoints.sftp_push.Config.QUARANTINE_DIR", tmp_path), \
-         patch("parser.entrypoints.sftp_push.time.sleep", side_effect=KeyboardInterrupt):
+    with patch("parser.entrypoints.sftp_push.threading.Thread", CapturingThread), patch(
+        "parser.entrypoints.sftp_push.threading.Event", return_value=mock_event
+    ), patch("parser.entrypoints.sftp_push.FileManager"), patch(
+        "parser.entrypoints.sftp_push.FileWatcher"
+    ), patch(
+        "parser.entrypoints.sftp_push.DBWriter", return_value=mock_db
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.PARSER_ENABLED", True
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.PROCESS_EXISTING_ON_STARTUP", False
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.DATABASE_URL", "postgresql://test"
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.USER_ID", "test_user"
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.STATS_REFRESH_INTERVAL", 60
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.ARCHIVED_DIR", tmp_path
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.QUARANTINE_DIR", tmp_path
+    ), patch(
+        "parser.entrypoints.sftp_push.time.sleep", side_effect=KeyboardInterrupt
+    ):
         try:
             main()
         except (KeyboardInterrupt, SystemExit):
@@ -181,8 +205,10 @@ def start(self):
 def test_stats_loop_calls_refresh_windowed_stats_on_startup(tmp_path):
     """stats_loop calls refresh_windowed_stats once before entering the timer loop."""
     mock_event = MagicMock()
-    mock_event.is_set.return_value = False  # connect loop: enter → connect succeeds → break
-    mock_event.wait.return_value = True     # timer wait → loop exits immediately
+    mock_event.is_set.return_value = (
+        False  # connect loop: enter → connect succeeds → break
+    )
+    mock_event.wait.return_value = True  # timer wait → loop exits immediately
 
     mock_db = _run_stats_loop(tmp_path, mock_event)
 
@@ -226,6 +252,7 @@ def test_stats_loop_creates_dbwriter_with_config_user_id(tmp_path):
 
     class CapturingThread(threading.Thread):
         """Intercepts Thread creation to capture the stats-refresh target."""
+
         def __init__(self, *args, target=None, name=None, **kwargs):
             super().__init__(*args, target=target, name=name, **kwargs)
             if name == "stats-refresh":
@@ -234,18 +261,27 @@ def __init__(self, *args, target=None, name=None, **kwargs):
         def start(self):
             pass  # don't spawn real threads in unit tests
 
-    with patch("parser.entrypoints.sftp_push.threading.Thread", CapturingThread), \
-         patch("parser.entrypoints.sftp_push.FileManager"), \
-         patch("parser.entrypoints.sftp_push.FileWatcher"), \
-         patch("parser.entrypoints.sftp_push.DBWriter") as MockDBWriter, \
-         patch("parser.entrypoints.sftp_push.Config.PARSER_ENABLED", True), \
-         patch("parser.entrypoints.sftp_push.Config.PROCESS_EXISTING_ON_STARTUP", False), \
-         patch("parser.entrypoints.sftp_push.Config.DATABASE_URL", "postgresql://test"), \
-         patch("parser.entrypoints.sftp_push.Config.USER_ID", "ctu_cz_tmobile"), \
-         patch("parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path), \
-         patch("parser.entrypoints.sftp_push.Config.ARCHIVED_DIR", tmp_path), \
-         patch("parser.entrypoints.sftp_push.Config.QUARANTINE_DIR", tmp_path), \
-         patch("parser.entrypoints.sftp_push.time.sleep", side_effect=KeyboardInterrupt):
+    with patch("parser.entrypoints.sftp_push.threading.Thread", CapturingThread), patch(
+        "parser.entrypoints.sftp_push.FileManager"
+    ), patch("parser.entrypoints.sftp_push.FileWatcher"), patch(
+        "parser.entrypoints.sftp_push.DBWriter"
+    ) as MockDBWriter, patch(
+        "parser.entrypoints.sftp_push.Config.PARSER_ENABLED", True
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.PROCESS_EXISTING_ON_STARTUP", False
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.DATABASE_URL", "postgresql://test"
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.USER_ID", "ctu_cz_tmobile"
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.INCOMING_DIR", tmp_path
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.ARCHIVED_DIR", tmp_path
+    ), patch(
+        "parser.entrypoints.sftp_push.Config.QUARANTINE_DIR", tmp_path
+    ), patch(
+        "parser.entrypoints.sftp_push.time.sleep", side_effect=KeyboardInterrupt
+    ):
         try:
             main()
         except (KeyboardInterrupt, SystemExit):
@@ -260,6 +296,6 @@ def start(self):
 
     for c in MockDBWriter.call_args_list:
         user_id = c.kwargs.get("user_id") or (c.args[1] if len(c.args) > 1 else None)
-        assert user_id == "ctu_cz_tmobile", (
-            f"DBWriter called without user_id=Config.USER_ID: {c}"
-        )
+        assert (
+            user_id == "ctu_cz_tmobile"
+        ), f"DBWriter called without user_id=Config.USER_ID: {c}"