Introduce backups service. (#87)

* Support create(), get_full_list(), download() and delete()

* Refactored Client.send() to support binary responses.

* Added Client.send_raw() which returns raw bytes from HTTP response.

pocketbase/client.py

@@ -8,6 +8,7 @@
 from pocketbase.models import FileUpload
 from pocketbase.models.record import Record
 from pocketbase.services.admin_service import AdminService
+from pocketbase.services.backups_service import BackupsService
 from pocketbase.services.collection_service import CollectionService
 from pocketbase.services.log_service import LogService
 from pocketbase.services.realtime_service import RealtimeService
@@ -45,6 +46,7 @@ def __init__(
         self.http_client = http_client or httpx.Client()
         # services
         self.admins = AdminService(self)
+        self.backups = BackupsService(self)
         self.collections = CollectionService(self)
         self.logs = LogService(self)
         self.settings = SettingsService(self)
@@ -57,13 +59,13 @@ def collection(self, id_or_name: str) -> RecordService:
             self.record_service[id_or_name] = RecordService(self, id_or_name)
         return self.record_service[id_or_name]
 
-    def send(self, path: str, req_config: dict[str:Any]) -> Any:
-        """Sends an api http request."""
+    def _send(self, path: str, req_config: dict[str:Any]) -> httpx.Response:
+        """Sends an api http request returning response object."""
         config = {"method": "GET"}
         config.update(req_config)
         # check if Authorization header can be added
         if self.auth_store.token and (
-            "headers" not in config or "Authorization" not in config["headers"]
+                "headers" not in config or "Authorization" not in config["headers"]
         ):
             config["headers"] = config.get("headers", {})
             config["headers"].update({"Authorization": self.auth_store.token})
@@ -105,6 +107,16 @@ def send(self, path: str, req_config: dict[str:Any]) -> Any:
                 f"General request error. Original error: {e}",
                 original_error=e,
             )
+        return response
+
+    def send_raw(self, path: str, req_config: dict[str:Any]) -> bytes:
+        """Sends an api http request returning raw bytes response."""
+        response = self._send(path, req_config)
+        return response.content
+
+    def send(self, path: str, req_config: dict[str:Any]) -> Any:
+        """Sends an api http request."""
+        response = self._send(path, req_config)
         try:
             data = response.json()
         except Exception:

pocketbase/models/__init__.py

@@ -1,8 +1,9 @@
 from .admin import Admin
+from .backups import Backup
 from .collection import Collection
 from .external_auth import ExternalAuth
 from .log_request import LogRequest
 from .record import Record
 from .file_upload import FileUpload
 
-__all__ = ["Admin", "Collection", "ExternalAuth", "LogRequest", "Record", "FileUpload"]
+__all__ = ["Admin", "Backup", "Collection", "ExternalAuth", "LogRequest", "Record", "FileUpload"]

pocketbase/models/backups.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+import datetime
+
+from pocketbase.models.utils import BaseModel
+from pocketbase.utils import to_datetime
+
+
+class Backup(BaseModel):
+    key: str
+    modified: str | datetime.datetime
+    size: int
+
+    def load(self, data: dict) -> None:
+        super().load(data)
+        self.key = data.get("key", "")
+        self.modified = to_datetime(data.pop("modified", ""))
+        self.size = data.get("size", 0)

pocketbase/services/backups_service.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from pocketbase.models import Backup
+from pocketbase.services.utils import BaseService
+
+
+class BackupsService(BaseService):
+    def decode(self, data: dict) -> Backup:
+        return Backup(data)
+
+    def base_path(self) -> str:
+        return "/api/backups"
+
+    def create(self, name: str):
+        # The backups service create method does not return an object.
+        self.client.send(
+            self.base_path(),
+            {"method": "POST", "body": {"name": name}},
+        )
+
+    def get_full_list(self, query_params: dict = {}) -> list[Backup]:
+        response_data = self.client.send(self.base_path(), {"method": "GET", "params": query_params})
+        return [self.decode(item) for item in response_data]
+
+    def download(self, key: str, file_token: str = None) -> bytes:
+        if file_token is None:
+            file_token = self.client.get_file_token()
+        return self.client.send_raw("%s/%s" % (self.base_path(), key),
+                                    {"method": "GET", "params": {"token": file_token}})
+
+    def delete(self, key: str):
+        self.client.send("%s/%s" % (self.base_path(), key), {"method": "DELETE"})

tests/integration/conftest.py

@@ -1,3 +1,5 @@
+import logging
+
 from pocketbase import PocketBase
 from pocketbase.utils import ClientResponseError
 import pytest

tests/integration/test_backups.py

@@ -0,0 +1,36 @@
+import datetime
+from uuid import uuid4
+
+from pocketbase import PocketBase
+
+
+class TestBackupsService:
+    def test_create_list_download_and_delete(self, client: PocketBase, state):
+        state.backup_name = "%s.zip" % (uuid4().hex[:16],)
+        client.backups.create(state.backup_name)
+
+        try:
+            # Find new backup in list of all backups
+            for backup in client.backups.get_full_list():
+                if backup.key == state.backup_name:
+                    state.backup = backup
+                    assert isinstance(backup.modified, datetime.datetime)
+                    assert backup.size > 0
+                    break
+            else:
+                self.fail("Backup %s not found in list of all backups" % (state.backup_name,))
+
+            # Download the backup
+            data = client.backups.download(state.backup_name)
+            assert isinstance(data, bytes)
+            assert len(data) == state.backup.size
+        finally:
+            # Cleanup
+            client.backups.delete(state.backup_name)
+
+            # Check that it was deleted
+            for backup in client.backups.get_full_list():
+                if backup.key == state.backup_name:
+                    self.fail("Backup %s still found in list of all backups" % (state.backup_name,))
+                    break
+