api: add Snapshot endpoint, with raw/binary transport support

ENDPOINT_STATUS.md

@@ -206,3 +206,38 @@ File: `consul/api/connect.py`
 | `GET /v1/connect/intentions/match` | `Connect.Intentions.match` | ✅ | - |
 | `POST /v1/connect/intentions` (legacy, deprecated 1.9.0) | - | ❌ | Intentionally not implemented — use exact-match endpoints above |
 | `GET/PUT/DELETE /v1/connect/intentions/:uuid` (legacy, deprecated 1.9.0) | - | ❌ | Intentionally not implemented — use exact-match endpoints above |
+
+## 14. Config Entries (`/v1/config`)
+File: `consul/api/config.py`
+
+Generic CRUD covering all config entry kinds (service-defaults, service-router,
+service-splitter, service-resolver, service-intentions, ingress-gateway,
+terminating-gateway, proxy-defaults, mesh, exported-services, api-gateway,
+http-route, tcp-route, inline-certificate, file-system-certificate, and others)
+through a single Kind/Name-based implementation. Kind-specific body fields are
+not individually modeled — callers pass the kind-specific body as a dict.
+
+| Endpoint | Python Method | Status | Missing Parameters |
+| :--- | :--- | :--- | :--- |
+| `PUT /v1/config` | `Config.set` | ✅ | - |
+| `GET /v1/config/:kind/:name` | `Config.get` | ✅ | - |
+| `GET /v1/config/:kind` | `Config.list` | ✅ | - |
+| `DELETE /v1/config/:kind/:name` | `Config.delete` | ✅ | - |
+
+## 15. Discovery Chain (`/v1/discovery-chain`)
+File: `consul/api/discovery_chain.py`
+
+| Endpoint | Python Method | Status | Missing Parameters |
+| :--- | :--- | :--- | :--- |
+| `GET /v1/discovery-chain/:service` | `DiscoveryChain.get` | ✅ | - |
+| `POST /v1/discovery-chain/:service` | `DiscoveryChain.get` | ✅ | Uses POST automatically when any override is passed |
+
+## 16. Snapshot (`/v1/snapshot`)
+File: `consul/api/snapshot.py`
+
+Both endpoints require a management-level ACL token (not a granular ACL rule).
+
+| Endpoint | Python Method | Status | Missing Parameters |
+| :--- | :--- | :--- | :--- |
+| `GET /v1/snapshot` | `Snapshot.save` | ✅ | - |
+| `PUT /v1/snapshot` | `Snapshot.restore` | ✅ | - |

consul/aio.py

@@ -34,29 +34,46 @@ def __init__(self, *args, loop=None, connections_limit=None, connections_timeout
         self._session = aiohttp.ClientSession(connector=connector, **session_kwargs)  # type: ignore
 
     async def _request(
-        self, callback, method, uri, headers: dict[str, str] | None, data=None, connections_timeout=None
+        self,
+        callback,
+        method,
+        uri,
+        headers: dict[str, str] | None,
+        data=None,
+        connections_timeout=None,
+        raw: bool = False,
     ):
         session_kwargs = {}
         if connections_timeout:
             timeout = aiohttp.ClientTimeout(total=connections_timeout)
             session_kwargs["timeout"] = timeout
         resp = await self._session.request(method, uri, headers=headers, data=data, **session_kwargs)  # type: ignore
-        body = await resp.text(encoding="utf-8")
+        # raw=True keeps the response as bytes (e.g. the gzip archive returned by
+        # GET /v1/snapshot) instead of decoding it as UTF-8 text, which would corrupt it.
+        body = await resp.read() if raw else await resp.text(encoding="utf-8")
         if resp.status == 599:
             raise Timeout
         r = base.Response(resp.status, resp.headers, body)
         return callback(r)
 
-    def get(self, callback, path, params=None, headers: dict[str, str] | None = None, connections_timeout=None):
+    def get(
+        self,
+        callback,
+        path,
+        params=None,
+        headers: dict[str, str] | None = None,
+        raw: bool = False,
+        connections_timeout=None,
+    ):
         uri = self.uri(path, params)
-        return self._request(callback, "GET", uri, headers=headers, connections_timeout=connections_timeout)
+        return self._request(callback, "GET", uri, headers=headers, connections_timeout=connections_timeout, raw=raw)
 
     def put(
         self,
         callback,
         path,
         params=None,
-        data: str = "",
+        data: str | bytes = "",
         headers: dict[str, str] | None = None,
         connections_timeout=None,
     ):

consul/api/snapshot.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from typing import Any
+
+from consul.callback import CB
+
+
+class Snapshot:
+    """
+    Saves/restores a full point-in-time snapshot of the Consul server state.
+    Both endpoints require a management-level ACL token -- not a granular ACL
+    rule -- and are primarily intended for disaster recovery.
+    """
+
+    def __init__(self, agent) -> None:
+        self.agent = agent
+
+    def save(self, token: str | None = None, dc: str | None = None, stale: bool = False) -> bytes:
+        """
+        Saves a snapshot of the current Consul server state as a gzip archive.
+        :param token: a management-level token
+        :param dc: Optional datacenter to target; defaults to the client's own dc.
+        :param stale: If True, allows any server (not just the leader) to service the request.
+        :return: the raw gzip archive bytes
+        """
+        params: list[tuple[str, Any]] = []
+        dc = dc or self.agent.dc
+        if dc:
+            params.append(("dc", dc))
+        if stale:
+            params.append(("stale", "true"))
+        headers = self.agent.prepare_headers(token)
+        return self.agent.http.get(CB.binary(), "/v1/snapshot", params=params, headers=headers, raw=True)
+
+    def restore(self, snapshot: bytes, token: str | None = None, dc: str | None = None) -> bool:
+        """
+        Restores a previously-saved snapshot. This is a disaster-recovery operation;
+        the target cluster must run the same Consul version as the source cluster.
+        :param snapshot: raw gzip archive bytes, as returned by :meth:`save`
+        :param token: a management-level token
+        :param dc: Optional datacenter to target; defaults to the client's own dc.
+        :return: True if the restore succeeded
+        """
+        params: list[tuple[str, Any]] = []
+        dc = dc or self.agent.dc
+        if dc:
+            params.append(("dc", dc))
+        headers = self.agent.prepare_headers(token)
+        return self.agent.http.put(CB.boolean(), "/v1/snapshot", params=params, headers=headers, data=snapshot)

consul/base.py

@@ -21,6 +21,7 @@
 from consul.api.operator import Operator
 from consul.api.query import Query
 from consul.api.session import Session
+from consul.api.snapshot import Snapshot
 from consul.api.status import Status
 from consul.api.txn import Txn
 from consul.exceptions import ConsulException
@@ -56,11 +57,11 @@ def uri(self, path: str, params: list[tuple[str, Any]] | None = None):
         return uri
 
     @abc.abstractmethod
-    def get(self, callback, path, params=None, headers: dict[str, str] | None = None):
+    def get(self, callback, path, params=None, headers: dict[str, str] | None = None, raw: bool = False):
         raise NotImplementedError
 
     @abc.abstractmethod
-    def put(self, callback, path, params=None, data: str = "", headers: dict[str, str] | None = None):
+    def put(self, callback, path, params=None, data: str | bytes = "", headers: dict[str, str] | None = None):
         raise NotImplementedError
 
     @abc.abstractmethod
@@ -167,6 +168,7 @@ def __init__(
         self.connect = Connect(self)
         self.config = Config(self)
         self.discovery_chain = DiscoveryChain(self)
+        self.snapshot = Snapshot(self)
 
     def __enter__(self):
         return self

consul/callback.py

@@ -43,6 +43,16 @@ def cb(response):
 
         return cb
 
+    @classmethod
+    def binary(cls) -> Callable[[Response], bytes]:
+        # returns the raw response body, for binary payloads (e.g. GET /v1/snapshot)
+        # that a JSON/text-oriented callback would corrupt
+        def cb(response):
+            CB._status(response, allow_404=False)
+            return response.body
+
+        return cb
+
     @classmethod
     def json(
         cls,

consul/std.py

@@ -13,15 +13,21 @@ def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
         self.session = requests.session()
 
-    def response(self, response: Response):
+    def response(self, response: Response, raw: bool = False):
+        if raw:
+            # e.g. the gzip archive returned by GET /v1/snapshot -- decoding it as
+            # UTF-8 text would corrupt it, so the raw bytes are kept as-is.
+            return base.Response(response.status_code, response.headers, response.content)
         response.encoding = "utf-8"
         return base.Response(response.status_code, response.headers, response.text)
 
-    def get(self, callback, path, params=None, headers: dict[str, str] | None = None):
+    def get(self, callback, path, params=None, headers: dict[str, str] | None = None, raw: bool = False):
         uri = self.uri(path, params)
-        return callback(self.response(self.session.get(uri, headers=headers, verify=self.verify, cert=self.cert)))
+        return callback(
+            self.response(self.session.get(uri, headers=headers, verify=self.verify, cert=self.cert), raw=raw)
+        )
 
-    def put(self, callback, path, params=None, data: str = "", headers: dict[str, str] | None = None):
+    def put(self, callback, path, params=None, data: str | bytes = "", headers: dict[str, str] | None = None):
         uri = self.uri(path, params)
         return callback(
             self.response(self.session.put(uri, headers=headers, data=data, verify=self.verify, cert=self.cert))

tests/api/test_snapshot.py

@@ -0,0 +1,26 @@
+import pytest
+
+import consul
+
+
+class TestSnapshot:
+    def test_snapshot_save_and_restore(self, acl_consul) -> None:
+        c, master_token, _consul_version = acl_consul
+
+        c.kv.put("snapshot-test-key", "before-snapshot", token=master_token)
+
+        snapshot = c.snapshot.save(token=master_token)
+        assert isinstance(snapshot, bytes)
+        assert snapshot[:2] == b"\x1f\x8b"  # gzip magic number
+
+        c.kv.put("snapshot-test-key", "after-snapshot", token=master_token)
+
+        assert c.snapshot.restore(snapshot, token=master_token) is True
+
+        _index, value = c.kv.get("snapshot-test-key", token=master_token)
+        assert value["Value"] == b"before-snapshot"
+
+    def test_snapshot_requires_management_token(self, acl_consul) -> None:
+        c, _master_token, _consul_version = acl_consul
+
+        pytest.raises(consul.ACLPermissionDenied, c.snapshot.save, token="anonymous")