Merge branch 'tickets/DM-51696'

Author: kfindeisen

python/activator/activator.py

@@ -170,16 +170,21 @@ def _get_storage_client():
 
 
 @functools.cache
-def _get_central_butler(uri):
+def _get_write_butler():
     """Lazy initialization of central Butler.
+    """
+    return get_central_butler(write_repo, instrument_name, writeable=True)
 
-    Parameters
-    ----------
-    uri : `str`
-        The URI of the repository to load. Only one Butler object is
-        constructed for any repository.
+
+@functools.cache
+def _get_read_butler():
+    """Lazy initialization of central Butler.
     """
-    return get_central_butler(uri, instrument_name)
+    if read_repo != write_repo:
+        return get_central_butler(read_repo, instrument_name, writeable=False)
+    else:
+        # Don't initialize an extra Butler from scratch
+        return _get_write_butler()
 
 
 @functools.cache
@@ -192,7 +197,7 @@ def _get_local_repo():
         The directory containing the repo, to be removed when the
         process exits.
     """
-    repo = make_local_repo(local_repos, _get_central_butler(read_repo), instrument_name)
+    repo = make_local_repo(local_repos, _get_read_butler(), instrument_name)
     tracker = LocalRepoTracker.get()
     tracker.register(os.getpid(), repo.name)
     return repo
@@ -452,8 +457,8 @@ def create_app():
         # Check initialization and abort early
         _get_consumer()
         _get_storage_client()
-        _get_central_butler(read_repo)
-        _get_central_butler(write_repo)
+        _get_read_butler()
+        _get_write_butler()
         _get_local_repo()
 
         app = flask.Flask(__name__)
@@ -500,8 +505,8 @@ def keda_start():
         # Check initialization and abort early
         _get_consumer()
         _get_storage_client()
-        _get_central_butler(read_repo)
-        _get_central_butler(write_repo)
+        _get_read_butler()
+        _get_write_butler()
         _get_local_repo()
 
         redis_session = RedisStreamSession(
@@ -923,8 +928,8 @@ def process_visit(expected_visit: FannedOutVisit):
 
                 # Create a fresh MiddlewareInterface object to avoid accidental
                 # "cross-talk" between different visits.
-                mwi = MiddlewareInterface(_get_central_butler(read_repo),
-                                          _get_central_butler(write_repo),
+                mwi = MiddlewareInterface(_get_read_butler(),
+                                          _get_write_butler(),
                                           image_bucket,
                                           expected_visit,
                                           pre_pipelines,

python/activator/middleware_interface.py

@@ -91,7 +91,7 @@
 
 
 @connect.retry(2, SQL_EXCEPTIONS, wait=repo_retry)
-def get_central_butler(central_repo: str, instrument_class: str):
+def get_central_butler(central_repo: str, instrument_class: str, writeable: bool):
     """Provide a Butler that can access the given repository and read and write
     data for the given instrument.
 
@@ -102,6 +102,8 @@ def get_central_butler(central_repo: str, instrument_class: str):
     instrument_class : `str`
         The name of the instrument whose data will be retrieved or written. May
         be either the fully qualified class name or the short name.
+    writeable : `bool`
+        Whether or not it's safe to attempt writes to this Butler.
 
     Returns
     -------
@@ -110,7 +112,7 @@ def get_central_butler(central_repo: str, instrument_class: str):
         ``instrument_name`` data.
     """
     return Butler(central_repo,
-                  writeable=True,
+                  writeable=writeable,
                   inferDefaults=False,
                   )
 

tests/test_middleware_interface.py

@@ -291,14 +291,21 @@ def setUp(self):
                                              prefix="file://")
 
     def test_get_butler(self):
-        for butler in [get_central_butler(self.central_repo, "lsst.obs.lsst.LsstComCamSim"),
-                       get_central_butler(self.central_repo, instname),
+        for butler in [get_central_butler(self.central_repo, "lsst.obs.lsst.LsstComCamSim", writeable=True),
+                       get_central_butler(self.central_repo, instname, writeable=True),
                        ]:
             # TODO: better way to test repo location?
             self.assertTrue(
                 butler.getURI("skyMap", skymap=skymap_name, collections=f"{instname}/defaults").ospath
                 .startswith(self.central_repo))
             self.assertTrue(butler.isWriteable())
+        for butler in [get_central_butler(self.central_repo, "lsst.obs.lsst.LsstComCamSim", writeable=False),
+                       get_central_butler(self.central_repo, instname, writeable=False),
+                       ]:
+            self.assertTrue(
+                butler.getURI("skyMap", skymap=skymap_name, collections=f"{instname}/defaults").ospath
+                .startswith(self.central_repo))
+            self.assertFalse(butler.isWriteable())
 
     def test_make_local_repo(self):
         for inst in [instname, "lsst.obs.lsst.LsstComCamSim"]: