CHANGELOG.md

@@ -1,4 +1,4 @@
-# 5.x.x (2026-01-28)
+# 5.4.0 (2026-01-28)
 
 ## New features
 
@@ -6,6 +6,9 @@
   supported by the collections, i.e., classes for which storage- and
   validation-endpoints exist.
 
+- Add `/maintenance`-endpoint to temporarilly lock collections for non-curator
+  access.
+
 
 # 5.3.6 (2026-01-13)
 

dump_things_service/__about__.py

@@ -1 +1 @@
-__version__ = '5.3.6'
+__version__ = '5.4.0'

dump_things_service/__init__.py

@@ -12,6 +12,7 @@ from starlette.status import (
     HTTP_403_FORBIDDEN,
     HTTP_404_NOT_FOUND,
     HTTP_500_INTERNAL_SERVER_ERROR,
+    HTTP_503_SERVICE_UNAVAILABLE,
 )
 from starlette.status import (
     HTTP_413_REQUEST_ENTITY_TOO_LARGE as HTTP_413_CONTENT_TOO_LARGE,
@@ -31,6 +32,7 @@ __all__ = [
     'HTTP_413_CONTENT_TOO_LARGE',
     'HTTP_422_UNPROCESSABLE_CONTENT',
     'HTTP_500_INTERNAL_SERVER_ERROR',
+    'HTTP_503_SERVICE_UNAVAILABLE',
     'JSON',
     'YAML',
     'config_file_name',

dump_things_service/config.py

@@ -172,7 +172,7 @@ class InstanceConfig:
     hashed_tokens: dict = dataclasses.field(default_factory=dict)
     validators: dict = dataclasses.field(default_factory=dict)
     use_classes: dict = dataclasses.field(default_factory=dict)
-
+    maintenance_mode: set = dataclasses.field(default_factory=set)
 
 mode_mapping = {
     TokenModes.READ_CURATED: TokenPermission(curated_read=True),

dump_things_service/main.py

@@ -80,6 +80,7 @@ from dump_things_service.model import (
     get_subclasses,
 )
 from dump_things_service.utils import (
+    authenticate_token,
     check_bounds,
     check_collection,
     combine_ttl,
@@ -94,8 +95,9 @@ if TYPE_CHECKING:
     from dump_things_service.lazy_list import LazyList
 
 
-class TokenCapabilityRequest(BaseModel):
+class MaintenanceRequest(BaseModel):
-    token: str | None
+    collection: str
+    active: bool
 
 
 class ServerCollectionResponse(BaseModel):
@@ -166,8 +168,8 @@ of the project.
 
 tag_info = [
     {
-        'name': 'Server info',
+        'name': 'Server management',
-        'description': 'Get general information about the server',
+        'description': 'General server operations',
     },
     {
         'name': 'Read records',
@@ -407,7 +409,7 @@ async def root() -> RedirectResponse:
 
 @app.get(
     '/server',
-    tags=['Server info'],
+    tags=['Server management'],
     name='get server information'
 )
 async def server() -> ServerResponse:
@@ -424,6 +426,48 @@ async def server() -> ServerResponse:
     )
 
 
+@app.post(
+    '/maintenance',
+    tags=['Server management'],
+    name='put a collection in maintenance mode'
+)
+async def maintenance(
+    body: MaintenanceRequest,
+    api_key: str | None = Depends(api_key_header_scheme),
+):
+
+    if api_key is None:
+        raise HTTPException(
+            status_code=HTTP_400_BAD_REQUEST,
+            detail=f'Token required for this operation',
+        )
+
+    collection = body.collection
+    active = body.active
+
+    # Try to authenticate the token with the authentication providers that
+    # are associated with the collection.
+    check_collection(g_instance_config, collection)
+    auth_info = authenticate_token(g_instance_config, collection, api_key)
+    permissions = auth_info.token_permission
+
+    if not (
+            permissions.curated_write
+            and permissions.curated_read
+            and permissions.zones_access
+    ):
+        raise HTTPException(
+            status_code=HTTP_400_BAD_REQUEST,
+            detail=f'Curator permissions required for this operation',
+        )
+
+    if active:
+        g_instance_config.maintenance_mode.add(collection)
+    else:
+        g_instance_config.maintenance_mode.remove(collection)
+    return
+
+
 @app.get(
     '/{collection}/record',
     tags=['Read records'],

dump_things_service/tests/test_basic.py

@@ -4,9 +4,11 @@ import pytest  # F401
 
 from .. import (
     HTTP_200_OK,
+    HTTP_400_BAD_REQUEST,
     HTTP_401_UNAUTHORIZED,
     HTTP_403_FORBIDDEN,
     HTTP_404_NOT_FOUND,
+    HTTP_503_SERVICE_UNAVAILABLE,
 )
 from ..__about__ import __version__
 from ..utils import cleaned_json
@@ -446,3 +448,57 @@ def test_ignore_classes(fastapi_client_simple):
             json={'pid': f'dlflatsocial:c_{class_name}'},
         )
         assert response.status_code == HTTP_404_NOT_FOUND
+
+
+def test_maintenance(fastapi_client_simple):
+    test_client, _ = fastapi_client_simple
+
+    # Ensure that only curators can put a collection in maintenance mode
+    response = test_client.post(
+        '/maintenance',
+        headers={'x-dumpthings-token': 'token-1'},
+        json={'collection': 'collection_1', 'active': True},
+    )
+    assert response.status_code == HTTP_400_BAD_REQUEST
+
+    # Ensure unknown collections are caught in maintenance mode
+    response = test_client.post(
+        '/maintenance',
+        headers={'x-dumpthings-token': 'token_admin'},
+        json={'collection': 'collection_x', 'active': True},
+    )
+    assert response.status_code == HTTP_404_NOT_FOUND
+
+    response = test_client.post(
+        '/maintenance',
+        headers={'x-dumpthings-token': 'token_admin'},
+        json={'collection': 'collection_1', 'active': True},
+    )
+
+    response = test_client.get(
+        '/collection_1/record?pid=abc:something/',
+        headers={'x-dumpthings-token': 'token-1'},
+    )
+    assert response.status_code == HTTP_503_SERVICE_UNAVAILABLE
+
+    # Ensure unknown collections are caught in maintenance mode
+    response = test_client.get(
+        '/collection_x/record?pid=abc:something/',
+        headers={'x-dumpthings-token': 'token-1'},
+    )
+    assert response.status_code == HTTP_404_NOT_FOUND
+
+    # Deactivate maintenance mode
+    response = test_client.post(
+        '/maintenance',
+        headers={'x-dumpthings-token': 'token_admin'},
+        json={'collection': 'collection_1', 'active': False},
+    )
+    assert response.status_code == HTTP_200_OK
+
+    # Ensure that
+    response = test_client.get(
+        '/collection_1/record?pid=abc:something/',
+        headers={'x-dumpthings-token': 'token-1'},
+    )
+    assert response.status_code == HTTP_200_OK

dump_things_service/utils.py

@@ -20,6 +20,7 @@ from dump_things_service import (
     HTTP_403_FORBIDDEN,
     HTTP_404_NOT_FOUND,
     HTTP_413_CONTENT_TOO_LARGE,
+    HTTP_503_SERVICE_UNAVAILABLE,
 )
 from dump_things_service.auth import (
     AuthenticationError,
@@ -213,6 +214,19 @@ async def process_token(
     final_permissions = join_default_token_permissions(
         instance_config, token_permissions, collection
     )
+
+    # Check for maintenance mode
+    if collection in instance_config.maintenance_mode:
+        if not (
+                final_permissions.curated_read
+                and final_permissions.curated_write
+                and final_permissions.zones_access
+        ):
+            raise HTTPException(
+                status_code=HTTP_503_SERVICE_UNAVAILABLE,
+                detail=f"Collection '{collection}' is in maintenance mode",
+            )
+
     if not final_permissions.incoming_read and not final_permissions.curated_read:
         raise HTTPException(
             status_code=HTTP_403_FORBIDDEN,
@@ -281,12 +295,6 @@ def get_token_store(
 ) -> tuple[ModelStore, str, TokenPermission, str] | tuple[None, None, None, None]:
     check_collection(instance_config, collection_name)
 
-    # Check whether a store for this collection and token does already exist.
-    # If the token is a hashed token, we have to
-    store_info = instance_config.token_stores[collection_name].get(plain_token)
-    if store_info:
-        return store_info
-
     # Try to authenticate the token with the authentication providers that
     # are associated with the collection.
     auth_info = authenticate_token(instance_config, collection_name, plain_token)
@@ -319,6 +327,11 @@ def get_token_store(
             detail='No incoming area for collection ' +  collection_name
         )
 
+    # Check whether a store for this collection and token does already exist.
+    store_info = instance_config.token_stores[collection_name].get(plain_token)
+    if store_info:
+        return store_info
+
     store_dir = instance_config.store_path / incoming / auth_info.incoming_label
     token_store = create_token_store(
         instance_config=instance_config,