Use check_readonly decorator instead of explicit checks in each fun…

…ction

endpoints/decorators.py

@@ -185,13 +185,17 @@ def wrapper(*args, **kwargs):
             return func(*args, **kwargs)
 
         # Skip if not in read only mode.
-        if app.config.get("REGISTRY_STATE", "normal") != "readonly":
+        if (
+            app.config.get("REGISTRY_STATE", "normal") != "readonly"
+            or app.config.get("DISABLE_PUSHES", False) == False
+        ):
             return func(*args, **kwargs)
 
         # Skip if readonly access is allowed.
         if hasattr(func, "__readonly_call_allowed"):
             return func(*args, **kwargs)
 
+        # Raise if complete registry is in read-only mode
         raise ReadOnlyModeException()
 
     return wrapper

endpoints/v2/__init__.py

@@ -28,6 +28,7 @@
     Unauthorized,
     Unsupported,
     V2RegistryException,
+    PushesDisabled,
 )
 from proxy import UpstreamRegistryError
 from util.http import abort
@@ -52,7 +53,10 @@ def handle_registry_v2_exception(error):
 
 @v2_bp.app_errorhandler(ReadOnlyModeException)
 def handle_readonly(ex):
-    return _format_error_response(ReadOnlyMode())
+    if app.config.get("REGISTRY_STATE") == "readonly":
+        return _format_error_response(ReadOnlyMode())
+    if app.config.get("PUSHES_DISABLED", False):
+        return _format_error_response(PushesDisabled())
 
 
 @v2_bp.app_errorhandler(UpstreamRegistryError)

workers/storagereplication.py

@@ -214,6 +214,11 @@ def create_gunicorn_worker():
         while True:
             time.sleep(100000)
 
+    if app.config.get("DISABLE_PUSHES", False):
+        logger.debug("Pushes to the registry are disabled; skipping")
+        while True:
+            time.sleep(100000)
+
     if features.STORAGE_REPLICATION:
         for storage_type, _ in list(app.config.get("DISTRIBUTED_STORAGE_CONFIG", {}).values()):
             if storage_type == "LocalStorage":