fix: review state management

Author: lc-spxl

docker-compose.yaml

@@ -6,8 +6,8 @@ services:
       MINIO_ROOT_USER: minioadmin
       MINIO_ROOT_PASSWORD: minioadmin
     ports:
-      - "19000:9000"
-      - "19001:9001"
+      - "9000:9000"
+      - "9001:9001"
     volumes:
       - minio_data:/data
     healthcheck:

meltano.yml

@@ -31,23 +31,23 @@ plugins:
           #     - centre
           #     - edition
 
-          - path: ./data/icon-d2*.grib2
-            table_name: icon_d2
-            bboxes:
-              - [46, 11, 45, 12]
-            skip_past: true
-            # ignore_fields:
-            #   - ensemble
-            #   - grid_type
-            #   - centre
-            #   - edition
+          # - path: ./data/icon-d2*.grib2
+          #   table_name: icon_d2
+          #   bboxes:
+          #     - [46, 11, 45, 12]
+          #   skip_past: true
+          #   # ignore_fields:
+          #   #   - ensemble
+          #   #   - grid_type
+          #   #   - centre
+          #   #   - edition
 
-          # - path: s3://local-data/test.grib
-          #   ignore_fields:
-          #     - ensemble
-          #     - grid_type
-          #     - centre
-          #     - edition
+          - path: s3://local-data/*.grib
+            ignore_fields:
+              - ensemble
+              - grid_type
+              - centre
+              - edition
 
   loaders:
     - name: target-jsonl

tap_grib/client.py

@@ -112,8 +112,7 @@ def __init__(
         self,
         tap,
         name: str,
-        *,
-        file_path: str | None,
+        file_path: str | None = None,
         primary_keys: list[str] | None = None,
         skip_past_reference: str | None = None,
         skip_past: bool | None = False,
@@ -122,6 +121,9 @@ def __init__(
         bboxes: list[tuple[float, float, float, float]] | None = None,
         **kwargs,
     ):
+
+        super().__init__(tap=tap, name=name, **kwargs)
+
         self.file_path = file_path
         self.extra_files = extra_files or ([file_path] if file_path else [])
         self.primary_keys = primary_keys or [
@@ -157,10 +159,9 @@ def __init__(
             raise ValueError(f"Cannot ignore core fields: {', '.join(sorted(invalid))}")
         self.ignore_fields = ignore_fields
 
-        # now call parent init with only tap/name/kwargs
-        super().__init__(tap=tap, name=name, **kwargs)
+        # super().__init__(tap=tap, name=name, **kwargs)
 
-        self.state_partitioning_keys = [SDC_FILENAME]
+        # self.state_partitioning_keys = [SDC_FILENAME]
         self.replication_key = SDC_INCREMENTAL_KEY
         self.forced_replication_method = "INCREMENTAL"
 
@@ -210,12 +211,27 @@ def get_records(
         dict[str, t.Any] | tuple[dict[t.Any, t.Any], dict[t.Any, t.Any] | None]
     ]:
 
+        start_mtime: datetime | None = self.get_starting_timestamp(context)
+        if start_mtime and start_mtime.tzinfo is None:
+            start_mtime = start_mtime.replace(tzinfo=timezone.utc)
+        elif start_mtime:
+            start_mtime = start_mtime.astimezone(timezone.utc)
+
         for path in self.extra_files:
             self.logger.info(f"[{self.name}] Streaming records from {path}")
             storage = Storage(path)
             info = storage.describe(path)
             mtime = info.mtime
 
+            if start_mtime is not None and mtime is not None and mtime <= start_mtime:
+                self.logger.info(
+                    "Skipping %s (mtime=%s <= bookmark=%s)",
+                    path,
+                    mtime,
+                    start_mtime,
+                )
+                continue
+
             filename = info.path
             partition_context = {SDC_FILENAME: filename}
             last_bookmark = self.get_starting_replication_key_value(partition_context)
@@ -273,15 +289,6 @@ def get_records(
                         if lats.size == 0:
                             continue
 
-                        for msg in grbs:
-                            try:
-                                lats, lons, vals = _extract_grid(msg)
-                            except Exception as e:
-                                self.logger.warning(f"Skipping message: {e}")
-                                continue
-                            if lats.size == 0:
-                                continue
-
                         # safe datetime extraction
                         valid_dt = getattr(msg, "validDate", None)
                         if valid_dt is None:

tap_grib/tap.py

@@ -111,8 +111,15 @@ def _parse_bboxes(
 
     def default_stream_name(self, pattern: str) -> str:
         base = os.path.splitext(os.path.basename(pattern))[0]
-        safe = re.sub(r"[^0-9a-zA-Z]+", "_", base)
-        return safe.strip("_").lower()
+
+        # sanitize
+        safe = re.sub(r"[^0-9a-zA-Z]+", "_", base).strip("_").lower()
+
+        # Fallback if empty
+        if not safe:
+            safe = "grib_stream"
+
+        return safe
 
     def discover_streams(self) -> list[Stream]:
         """Discover a single stream per path pattern (merging all matching files)."""

taskfile.yaml

@@ -1,9 +1,21 @@
 version: "3"
 
 tasks:
+  run:
+    env:
+      S3_ACCESS_KEY_ID: minioadmin
+      S3_SECRET_ACCESS_KEY: minioadmin
+      S3_BUCKET: local-data
+      S3_ENDPOINT_URL: http://localhost:9000
+    cmds:
+      - uv run meltano run tap-grib target-jsonl {{.CLI_ARGS}}
+
+  run:refresh:
+    cmds:
+      - task run -- --full-refresh
+
   release:
     cmds:
-      # - uv lock --upgrade-package mqtt-ingestor
       - uv run semantic-release -v version --no-vcs-release
       - git push
       - git push --tags