OP#10322: Read the prefix from the project, not the environment

The providers now list under the prefix the loaded project declares, and
S3_KEY_PREFIX is gone from the environment. An instance is no longer pinned to
one prefix for its lifetime: opening another project moves where it reads,
within the one bucket it may authenticate to.

With no project loaded there is no prefix and nothing to list, so the time-range
dialog is refused and says to load a project first. That is a behaviour change
to slice 6, which offered the dialog at startup, and matches how the worker
runs: a lease starts when an otconfig is delivered.

The acceptance test preloads a project otconfig instead of setting
S3_KEY_PREFIX, which is the same change seen from the outside.

ADR 0003's variable list is marked as the list at the time of that ADR rather
than rewritten: the decision to configure S3 through the environment stands,
only the prefix left it.

Author: randy-seng

OTAnalytics/plugin_s3/config/env_vars.py

@@ -11,7 +11,6 @@
 ENV_S3_SECRET_KEY = "OTA_S3_SECRET_KEY"  # nosec B105 - variable name, not a secret
 ENV_S3_BUCKET = "OTA_S3_BUCKET"
 ENV_S3_REGION = "OTA_S3_REGION"
-ENV_S3_KEY_PREFIX = "OTA_S3_KEY_PREFIX"  # gitleaks:allow - variable name, not a secret
 ENV_S3_USER_SOURCE = "OTA_S3_USER_SOURCE"
 ENV_S3_MAX_LOAD_DURATION = "OTA_S3_MAX_LOAD_DURATION"
 ENV_S3_DOWNLOAD_CONCURRENCY = "OTA_S3_DOWNLOAD_CONCURRENCY"
@@ -41,7 +40,6 @@ class S3Env:
     secret_key: str | None = field(default_factory=_make_env_reader(ENV_S3_SECRET_KEY))
     bucket: str | None = field(default_factory=_make_env_reader(ENV_S3_BUCKET))
     region: str | None = field(default_factory=_make_env_reader(ENV_S3_REGION))
-    key_prefix: str | None = field(default_factory=_make_env_reader(ENV_S3_KEY_PREFIX))
     user_source: str | None = field(
         default_factory=_make_env_reader(ENV_S3_USER_SOURCE)
     )

OTAnalytics/plugin_s3/config/parsing.py

@@ -8,7 +8,6 @@
     ENV_S3_ACCESS_KEY,
     ENV_S3_BUCKET,
     ENV_S3_ENDPOINT_URL,
-    ENV_S3_KEY_PREFIX,
     ENV_S3_SECRET_KEY,
     ENV_S3_USER_SOURCE,
     S3Env,
@@ -91,7 +90,6 @@ def parse_s3_config(env: S3Env) -> S3Config:
         (ENV_S3_ACCESS_KEY, env.access_key),
         (ENV_S3_SECRET_KEY, env.secret_key),
         (ENV_S3_BUCKET, env.bucket),
-        (ENV_S3_KEY_PREFIX, env.key_prefix),
         (ENV_S3_USER_SOURCE, env.user_source),
     )
     if missing := [name for name, value in required if value is None]:
@@ -103,7 +101,6 @@ def parse_s3_config(env: S3Env) -> S3Config:
         secret_key=_required(env.secret_key),
         bucket=_required(env.bucket),
         region=env.region,
-        key_prefix=_required(env.key_prefix),
         user_source=_required(env.user_source),
         max_load_duration=_parse_max_load_duration(env),
         download_concurrency=_parse_download_concurrency(env),

OTAnalytics/plugin_s3/config/s3.py

@@ -15,8 +15,6 @@ class S3Config:
         secret_key (str): S3 secret key for authentication.
         bucket (str): S3 bucket name.
         region (str | None): AWS region (None if not applicable).
-        key_prefix (str): prefix under which this instance's tracks and videos
-            live. Fixed at startup; the user selects a time range, not a prefix.
         user_source (str): base path on the local filesystem that downloaded
             objects are written to, mirroring their S3 keys. Named after
             OTCloud's `S3_USER_SOURCE` for cross-repo consistency.
@@ -30,7 +28,6 @@ class S3Config:
     secret_key: str
     bucket: str
     region: str | None
-    key_prefix: str
     user_source: str
     max_load_duration: timedelta = DEFAULT_MAX_LOAD_DURATION
     download_concurrency: int = DEFAULT_DOWNLOAD_CONCURRENCY

OTAnalytics/plugin_s3/s3_file_providers.py

@@ -10,7 +10,9 @@
 
 import ijson
 
+from OTAnalytics.application.key_prefix import S3KeyPrefix
 from OTAnalytics.application.logger import logger
+from OTAnalytics.application.state import CurrentKeyPrefix
 from OTAnalytics.application.use_cases.ask_for_load_window import AskForLoadWindow
 from OTAnalytics.application.use_cases.provide_input_files import (
     ProvideTrackFiles,
@@ -35,6 +37,10 @@
 DOWNLOADING_VIDEOS = "Downloading videos"
 
 CANCELLED = "Loading from S3 cancelled. Nothing was loaded."
+NO_PROJECT = (
+    "This installation reads from S3, where a project says which data is its"
+    " own. Load a project first, then select a time range."
+)
 
 
 class MissingVideoForTrackFile(Exception):
@@ -104,6 +110,9 @@ class _S3Provider:
         list_objects (S3ListObjects): lists the bucket under the key prefix.
         download_objects (DownloadObjects): downloads the selected objects.
         config (S3Config): the S3 settings fixed at startup.
+        current_key_prefix (CurrentKeyPrefix): where the loaded project's data
+            lives. Read per load rather than at startup, because opening another
+            project moves it.
     """
 
     def __init__(
@@ -112,11 +121,13 @@ def __init__(
         list_objects: S3ListObjects,
         download_objects: DownloadObjects,
         config: S3Config,
+        current_key_prefix: CurrentKeyPrefix,
     ) -> None:
         self._dialog = dialog
         self._list_objects = list_objects
         self._download_objects = download_objects
         self._config = config
+        self._current_key_prefix = current_key_prefix
 
     async def _provide(
         self, title: str, suffixes: set[str], description: str
@@ -131,13 +142,19 @@ async def _provide(
         Returns:
             list[Path]: the local paths, empty if nothing was loaded.
         """
-        selected = await self._dialog.ask(title, self._source())
+        key_prefix = self._current_key_prefix.get()
+        if key_prefix is None:
+            # The prefix belongs to the project, so before one is loaded there
+            # is nothing to list and no window worth asking for.
+            self._dialog.report_error(NO_PROJECT)
+            return []
+        selected = await self._dialog.ask(title, self._source(key_prefix))
         if selected is None:
             return []
         window = selected.clamp(self._config.max_load_duration)
         if window.was_clamped:
             self._dialog.report_clamped(window)
-        keys = await self._list_objects.list_keys(self._config.key_prefix)
+        keys = await self._list_objects.list_keys(key_prefix.value)
         wanted = select_in_window(keys, window, suffixes)
         if not wanted:
             logger().info(f"Nothing to load between {window.start} and {window.end}")
@@ -157,7 +174,7 @@ async def _provide(
             # browser. The traceback still goes to the log.
             logger().exception(cause, exc_info=True)
             self._dialog.report_error(
-                f"Could not load from '{self._source()}': {cause}."
+                f"Could not load from '{self._source(key_prefix)}': {cause}."
                 " Nothing was loaded."
             )
             return []
@@ -168,9 +185,9 @@ async def _load(
         """Download what was selected. Overridden to fetch companions too."""
         return await self._download_objects.download_all(wanted, description)
 
-    def _source(self) -> str:
+    def _source(self, key_prefix: S3KeyPrefix) -> str:
         """Where files come from, for the user to see but not to change."""
-        return f"{self._config.bucket}/{self._config.key_prefix}"
+        return f"{self._config.bucket}/{key_prefix}"
 
 
 class S3TrackFileProvider(_S3Provider, ProvideTrackFiles):
@@ -181,6 +198,7 @@ class S3TrackFileProvider(_S3Provider, ProvideTrackFiles):
         list_objects (S3ListObjects): lists the bucket under the key prefix.
         download_objects (DownloadObjects): downloads the selected objects.
         config (S3Config): the S3 settings fixed at startup.
+        current_key_prefix (CurrentKeyPrefix): where the loaded project's data lives.
         read_video_name (Callable[[Path], str]): reads a track file's video name.
     """
 
@@ -190,9 +208,12 @@ def __init__(
         list_objects: S3ListObjects,
         download_objects: DownloadObjects,
         config: S3Config,
+        current_key_prefix: CurrentKeyPrefix,
         read_video_name: Callable[[Path], str] = read_video_name,
     ) -> None:
-        super().__init__(dialog, list_objects, download_objects, config)
+        super().__init__(
+            dialog, list_objects, download_objects, config, current_key_prefix
+        )
         self._read_video_name = read_video_name
 
     async def provide(self) -> list[Path]:

OTAnalytics/plugin_ui/nicegui_application.py

@@ -306,6 +306,7 @@ def provide_track_files(self) -> ProvideTrackFiles:
                 list_objects=self.s3_list_objects,
                 download_objects=self.download_objects,
                 config=config,
+                current_key_prefix=self.current_key_prefix,
             )
         return LocalTrackFileProvider(self.ui_factory)
 
@@ -317,6 +318,7 @@ def provide_video_files(self) -> ProvideVideoFiles:
                 list_objects=self.s3_list_objects,
                 download_objects=self.download_objects,
                 config=config,
+                current_key_prefix=self.current_key_prefix,
             )
         return LocalVideoFileProvider(self.ui_factory)
 

tests/acceptance/test_load_from_s3.py

@@ -1,6 +1,7 @@
 """Acceptance: in s3 mode, adding tracks asks for a time range, not a file."""
 
 import io
+import json
 import os
 import subprocess
 import sys
@@ -22,7 +23,6 @@
     ENV_S3_ACCESS_KEY,
     ENV_S3_BUCKET,
     ENV_S3_ENDPOINT_URL,
-    ENV_S3_KEY_PREFIX,
     ENV_S3_MAX_LOAD_DURATION,
     ENV_S3_SECRET_KEY,
     ENV_S3_USER_SOURCE,
@@ -76,12 +76,49 @@ def rustfs() -> Iterator[dict]:
         }
 
 
+def _project_otconfig(directory: Path) -> Path:
+    """An otconfig declaring where this project's data lives.
+
+    In s3 mode the prefix comes from the project, not the environment (ADR
+    0004), so the application has to be given a project before it can list
+    anything at all.
+    """
+    directory.mkdir(parents=True, exist_ok=True)
+    file = directory / "project.otconfig"
+    file.write_text(
+        json.dumps(
+            {
+                "project": {"name": "Acceptance", "start_date": 1684886400},
+                "s3_key_prefix": PREFIX,
+                "videos": [],
+                "analysis": {
+                    "do_events": True,
+                    "do_counting": True,
+                    "tracks": [],
+                    "export": {
+                        "save_name": "acceptance",
+                        "save_suffix": "",
+                        "event_formats": ["csv"],
+                        "count_intervals": [15],
+                    },
+                    "num_processes": 1,
+                    "logfile": "logs",
+                },
+                "sections": [],
+                "flows": [],
+            }
+        )
+    )
+    return file
+
+
 @pytest.fixture
 def s3_app(rustfs: dict, tmp_path: Path) -> Iterator[Any]:
     """The real application, started in s3 mode against RustFS.
 
     Configuration is environment only, so pointing the documented variables at a
-    throwaway RustFS exercises the production path with no test-only hooks.
+    throwaway RustFS exercises the production path with no test-only hooks. The
+    project is preloaded from a file, which is where the key prefix comes from.
     """
     environment = dict(os.environ)
     environment.update(
@@ -91,7 +128,6 @@ def s3_app(rustfs: dict, tmp_path: Path) -> Iterator[Any]:
             ENV_S3_ACCESS_KEY: rustfs["access_key"],
             ENV_S3_SECRET_KEY: rustfs["secret_key"],
             ENV_S3_BUCKET: BUCKET,
-            ENV_S3_KEY_PREFIX: PREFIX,
             ENV_S3_USER_SOURCE: str(tmp_path / "user-source"),
             ENV_S3_MAX_LOAD_DURATION: "20m",
         }
@@ -102,6 +138,8 @@ def s3_app(rustfs: dict, tmp_path: Path) -> Iterator[Any]:
             "-m",
             "OTAnalytics",
             "--webui",
+            "--config",
+            str(_project_otconfig(tmp_path / "project")),
             "--file-picker-directory",
             file_picker_directory(),
         ],

tests/integration/OTAnalytics/plugin_s3/test_load_from_rustfs.py

@@ -17,6 +17,8 @@
 
 import pytest
 
+from OTAnalytics.application.key_prefix import S3KeyPrefix
+from OTAnalytics.application.state import CurrentKeyPrefix
 from OTAnalytics.domain.load_window import LoadWindow
 from OTAnalytics.plugin_s3.config.s3 import S3Config
 from OTAnalytics.plugin_s3.download import S3Download
@@ -101,6 +103,7 @@ class Given:
     config: S3Config
     dialog: Mock
     user_source: Path
+    current_key_prefix: CurrentKeyPrefix
 
 
 def create_given(rustfs: dict, tmp_path: Path, hours: float = 1) -> Given:
@@ -111,7 +114,6 @@ def create_given(rustfs: dict, tmp_path: Path, hours: float = 1) -> Given:
         secret_key=rustfs["secret_key"],
         bucket=BUCKET,
         region=None,
-        key_prefix=PREFIX,
         user_source=str(user_source),
         max_load_duration=MAX_LOAD_DURATION,
         download_concurrency=4,
@@ -120,7 +122,14 @@ def create_given(rustfs: dict, tmp_path: Path, hours: float = 1) -> Given:
     dialog.ask = AsyncMock(
         return_value=LoadWindow(start=START, end=START + timedelta(hours=hours))
     )
-    return Given(config=config, dialog=dialog, user_source=user_source)
+    current_key_prefix = CurrentKeyPrefix()
+    current_key_prefix.set(S3KeyPrefix(PREFIX))
+    return Given(
+        config=config,
+        dialog=dialog,
+        user_source=user_source,
+        current_key_prefix=current_key_prefix,
+    )
 
 
 def _download_objects(given: Given) -> DownloadObjects:
@@ -138,6 +147,7 @@ def create_track_target(given: Given) -> S3TrackFileProvider:
         list_objects=S3ListObjects(S3Store(given.config)),
         download_objects=_download_objects(given),
         config=given.config,
+        current_key_prefix=given.current_key_prefix,
     )
 
 
@@ -147,6 +157,7 @@ def create_video_target(given: Given) -> S3VideoFileProvider:
         list_objects=S3ListObjects(S3Store(given.config)),
         download_objects=_download_objects(given),
         config=given.config,
+        current_key_prefix=given.current_key_prefix,
     )
 
 

tests/unit/OTAnalytics/plugin_s3/config/test_parsing.py

@@ -9,7 +9,6 @@
     ENV_S3_BUCKET,
     ENV_S3_DOWNLOAD_CONCURRENCY,
     ENV_S3_ENDPOINT_URL,
-    ENV_S3_KEY_PREFIX,
     ENV_S3_MAX_LOAD_DURATION,
     ENV_S3_REGION,
     ENV_S3_SECRET_KEY,
@@ -32,7 +31,6 @@ def env(**overrides: str | None) -> S3Env:
         "secret_key": None,
         "bucket": None,
         "region": None,
-        "key_prefix": None,
         "user_source": None,
         "max_load_duration": None,
         "download_concurrency": None,
@@ -48,7 +46,6 @@ def complete_env(**overrides: str | None) -> S3Env:
         "access_key": "minioadmin",
         "secret_key": "minioadmin",  # gitleaks:allow
         "bucket": "otcloud",
-        "key_prefix": "6-1145/site/OTCamera19/",
         "user_source": "/data/otanalytics-source",
     }
     required.update(overrides)
@@ -93,7 +90,6 @@ def test_parse_every_variable(self) -> None:
         assert actual.access_key == "minioadmin"
         assert actual.bucket == "otcloud"
         assert actual.region == "us-east-1"
-        assert actual.key_prefix == "6-1145/site/OTCamera19/"
         assert actual.user_source == "/data/otanalytics-source"
         assert actual.max_load_duration == timedelta(hours=2)
         assert actual.download_concurrency == 16
@@ -109,15 +105,6 @@ def test_optional_variables_default(self) -> None:
         assert actual.max_load_duration == timedelta(hours=10)
         assert actual.download_concurrency == 8
 
-    def test_empty_key_prefix_is_kept(self) -> None:
-        """`S3_KEY_PREFIX=""` means the bucket root, not "unset".
-
-        # Requirement OP#10256
-        """
-        actual = parse_s3_config(complete_env(key_prefix=""))
-
-        assert actual.key_prefix == ""
-
     def test_malformed_duration_is_rejected(self) -> None:
         with pytest.raises(InvalidDurationError):
             parse_s3_config(complete_env(max_load_duration="10"))
@@ -138,7 +125,6 @@ def test_report_every_missing_variable_at_once(self) -> None:
             ENV_S3_ENDPOINT_URL,
             ENV_S3_ACCESS_KEY,
             ENV_S3_SECRET_KEY,
-            ENV_S3_KEY_PREFIX,
             ENV_S3_USER_SOURCE,
         ]
 
@@ -162,7 +148,6 @@ def test_only_absent_variables_are_reported(self) -> None:
         assert excinfo.value.missing == [
             ENV_S3_ENDPOINT_URL,
             ENV_S3_SECRET_KEY,
-            ENV_S3_KEY_PREFIX,
         ]
 
     def test_message_names_the_environment_variables(self) -> None:
@@ -189,7 +174,6 @@ def test_unset_variables_are_none(self, monkeypatch: pytest.MonkeyPatch) -> None
             ENV_S3_SECRET_KEY,
             ENV_S3_BUCKET,
             ENV_S3_REGION,
-            ENV_S3_KEY_PREFIX,
             ENV_S3_USER_SOURCE,
             ENV_S3_MAX_LOAD_DURATION,
             ENV_S3_DOWNLOAD_CONCURRENCY,