CHANGELOG.md

@@ -2,7 +2,8 @@
 
 ## New features
 
-- Add a "delete-record" subcommand to `dtc`.
+- Add "delete-record" subcommand to `dtc`.
+- Add "export" subcommand to `dtc`.
 
 ## Bug fixes
 

dump_things_pyclient/commands/dtc_plugins/maintenance.py

@@ -0,0 +1,73 @@
+import logging
+
+import rich_click as click
+
+from ...communicate import (
+    HTTPError,
+    maintenance as communicate_maintenance,
+)
+
+
+logger = logging.getLogger('maintenance')
+
+subcommand_name = 'maintenance'
+
+
+@click.command(short_help='Activate or deactivate maintenance mode on a collection')
+@click.pass_obj
+@click.argument(
+    'service_url',
+    metavar='SERVICE_URL',
+)
+@click.argument(
+    'collection',
+    metavar='COLLECTION',
+)
+@click.argument(
+    'active',
+    metavar='ACTIVE',
+    type=click.Choice(['On', 'Off'], case_sensitive=False),
+)
+def cli(
+        obj,
+        service_url: str,
+        collection: str,
+        active: bool,
+):
+    """Activate or deactivate maintenance mode on collection COLLECTION on the
+    service SERVICE_URL. The argument ACTIVE should be either `On` or `Off`
+    (case-insensitive).
+
+    A token with curator rights is required.
+
+    This command expects a server version >= 5.4.0"""
+    try:
+        return maintenance(
+            obj,
+            service_url,
+            collection,
+            active,
+        )
+    except HTTPError as e:
+        click.echo(f'ERROR: {e}: {e.response.text}', err=True)
+    return 1
+
+
+def maintenance(
+        obj: str,
+        service_url: str,
+        collection: str,
+        active: bool,
+):
+    token = obj
+    if token is None:
+        click.echo('ERROR: no token provided', err=True)
+        return 1
+
+    communicate_maintenance(
+        service_url=service_url,
+        collection=collection,
+        active=active,
+        token=token,
+    )
+    return 0

dump_things_pyclient/communicate.py

@@ -37,6 +37,8 @@ __all__ = [
     'incoming_read_records_of_class',
     'incoming_read_record_with_pid',
     'incoming_write_record',
+    'maintenance',
+    'server',
 ]
 
 
@@ -694,6 +696,33 @@ def server(
     return _do_request(requests.get, url=url, token=None, params=None)
 
 
+def maintenance(
+        service_url: str,
+        collection: str,
+        active: bool,
+        token: str,
+) -> None:
+    """Activate or deactivate maintenance mode of a collection
+
+    :param service_url: the base URL of the service, i.e., the URL up to
+           `/<collection>/...`, `/maintenance`, or `/server`
+    :param collection: the name of the collection
+    :param active: whether maintenance mode should be active (`True`) or
+           non-active (`False`).
+    :param token: a token to authenticate against the endpoint, the token
+           must have curator-rights for the collection
+    """
+    url = (
+            (f'{service_url[:-1]}' if service_url.endswith('/') else service_url)
+            + '/maintenance'
+    )
+    _post_to_url(
+        url=url,
+        token=token,
+        json={'collection': collection, 'active': active}
+    )
+
+
 def _get_from_url(url: str,
                   token: str | None,
                   params: dict[str, str] | None = None,