Better typing annotations and some updates to match TS lib

Author: vaphes

.gitignore

@@ -3,6 +3,7 @@
 
 
 .vscode/
+.ruff_cache/
 
 ### Python ###
 # Byte-compiled / optimized / DLL files

pocketbase/client.py

@@ -1,40 +1,30 @@
 from __future__ import annotations
 
 from typing import Any, Dict
-from urllib.parse import quote, urlencode
 
 import httpx
 
+from pocketbase.errors import ClientResponseError
 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.files_service import FileService
+from pocketbase.services.health_service import HealthService
 from pocketbase.services.log_service import LogService
 from pocketbase.services.realtime_service import RealtimeService
 from pocketbase.services.record_service import RecordService
 from pocketbase.services.settings_service import SettingsService
-from pocketbase.stores.base_auth_store import BaseAuthStore
-from pocketbase.utils import ClientResponseError
+from pocketbase.stores.base_auth_store import AuthStore, BaseAuthStore
 
 
 class Client:
-    base_url: str
-    lang: str
-    auth_store: BaseAuthStore
-    settings: SettingsService
-    admins: AdminService
-    collections: CollectionService
-    records: RecordService
-    logs: LogService
-    realtime: RealtimeService
-    record_service: Dict[str, RecordService]
-
     def __init__(
         self,
         base_url: str = "/",
         lang: str = "en-US",
-        auth_store: BaseAuthStore | None = None,
+        auth_store: AuthStore | None = None,
         timeout: float = 120,
         http_client: httpx.Client | None = None,
     ) -> None:
@@ -47,16 +37,12 @@ def __init__(
         self.admins = AdminService(self)
         self.backups = BackupsService(self)
         self.collections = CollectionService(self)
+        self.files = FileService(self)
+        self.health = HealthService(self)
         self.logs = LogService(self)
         self.settings = SettingsService(self)
         self.realtime = RealtimeService(self)
-        self.record_service = {}
-
-    def collection(self, id_or_name: str) -> RecordService:
-        """Returns the RecordService associated to the specified collection."""
-        if id_or_name not in self.record_service:
-            self.record_service[id_or_name] = RecordService(self, id_or_name)
-        return self.record_service[id_or_name]
+        self.record_service: Dict[str, RecordService] = {}
 
     def _send(self, path: str, req_config: dict[str, Any]) -> httpx.Response:
         """Sends an api http request returning response object."""
@@ -74,9 +60,9 @@ def _send(self, path: str, req_config: dict[str, Any]) -> httpx.Response:
         method = config.get("method", "GET")
         params = config.get("params", None)
         headers = config.get("headers", None)
-        body = config.get("body", None)
+        body: dict[str, Any] | None = config.get("body", None)
         # handle requests including files as multipart:
-        data = {}
+        data: dict[str, Any] | None = {}
         files = ()
         for k, v in (body if isinstance(body, dict) else {}).items():
             if isinstance(v, FileUpload):
@@ -108,6 +94,12 @@ def _send(self, path: str, req_config: dict[str, Any]) -> httpx.Response:
             )
         return response
 
+    def collection(self, id_or_name: str) -> RecordService:
+        """Returns the RecordService associated to the specified collection."""
+        if id_or_name not in self.record_service:
+            self.record_service[id_or_name] = RecordService(self, id_or_name)
+        return self.record_service[id_or_name]
+
     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)
@@ -123,35 +115,29 @@ def send(self, path: str, req_config: dict[str, Any]) -> Any:
         if response.status_code >= 400:
             raise ClientResponseError(
                 f"Response error. Status code:{response.status_code}",
-                url=response.url,
+                url=str(response.url),
                 status=response.status_code,
                 data=data,
             )
         return data
 
-    def get_file_url(self, record: Record, filename: str, query_params: dict):
-        parts = [
-            "api",
-            "files",
-            quote(record.collection_id or record.collection_name),
-            quote(record.id),
-            quote(filename),
-        ]
-        result = self.build_url("/".join(parts))
-        if len(query_params) != 0:
-            params: str = urlencode(query_params)
-            result += "&" if "?" in result else "?"
-            result += params
-        return result
-
-    def get_file_token(self):
-        res = self.send("/api/files/token", req_config={"method": "POST"})
-        return res["token"]
-
     def build_url(self, path: str) -> str:
         url = self.base_url
         if not self.base_url.endswith("/"):
             url += "/"
         if path.startswith("/"):
             path = path[1:]
         return url + path
+
+    # TODO: add deprecated decorator
+    def get_file_url(
+        self,
+        record: Record,
+        filename: str,
+        query_params: dict[str, Any] | None = None,
+    ):
+        return self.files.get_url(record, filename, query_params)
+
+    # TODO: add deprecated decorator
+    def get_file_token(self) -> str:
+        return self.files.get_token()

pocketbase/errors.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+class ClientResponseError(Exception):
+    def __init__(
+        self,
+        *args: Any,
+        url: str = "",
+        status: int = 0,
+        data: dict[str, Any] | None = None,
+        is_abort: bool = False,
+        original_error: Any = None,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(*args)
+        self.url = url
+        self.status = status
+        self.data = data or {}
+        self.is_abort = is_abort
+        self.original_error = original_error

pocketbase/models/admin.py

@@ -1,13 +1,15 @@
 from __future__ import annotations
 
+from typing import Any
+
 from pocketbase.models.utils.base_model import BaseModel
 
 
 class Admin(BaseModel):
     avatar: int
     email: str
 
-    def load(self, data: dict) -> None:
+    def load(self, data: dict[str, Any]) -> None:
         super().load(data)
         self.avatar = data.get("avatar", 0)
         self.email = data.get("email", "")

pocketbase/models/backups.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import datetime
+from typing import Any
 
 from pocketbase.models.utils import BaseModel
 from pocketbase.utils import to_datetime
@@ -11,7 +12,7 @@ class Backup(BaseModel):
     modified: str | datetime.datetime
     size: int
 
-    def load(self, data: dict) -> None:
+    def load(self, data: dict[str, Any]) -> None:
         super().load(data)
         self.key = data.get("key", "")
         self.modified = to_datetime(data.pop("modified", ""))

pocketbase/models/collection.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from typing import Any
+
 from pocketbase.models.utils.base_model import BaseModel
 from pocketbase.models.utils.schema_field import SchemaField
 
@@ -14,9 +16,9 @@ class Collection(BaseModel):
     create_rule: str | None
     update_rule: str | None
     delete_rule: str | None
-    options: dict
+    options: dict[str, Any]
 
-    def load(self, data: dict) -> None:
+    def load(self, data: dict[str, Any]) -> None:
         super().load(data)
         self.name = data.get("name", "")
         self.system = data.get("system", False)

pocketbase/models/external_auth.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from typing import Any
+
 from pocketbase.models.utils.base_model import BaseModel
 
 
@@ -9,7 +11,7 @@ class ExternalAuth(BaseModel):
     provider: str
     provider_id: str
 
-    def load(self, data: dict) -> None:
+    def load(self, data: dict[str, Any]) -> None:
         super().load(data)
         self.record_id = data.get("recordId", "")
         self.collection_id = data.get("collectionId", "")

pocketbase/models/file_upload.py

@@ -6,8 +6,8 @@
 
 
 class FileUpload:
-    def __init__(self, *args):
-        self.files: FileUploadTypes = args
+    def __init__(self, *args: FileUploadTypes):
+        self.files = args
 
     def get(self, key: str):
         if isinstance(self.files[0], Sequence) and not isinstance(

pocketbase/models/log_request.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from typing import Any
+
 from pocketbase.models.utils.base_model import BaseModel
 
 
@@ -12,9 +14,9 @@ class LogRequest(BaseModel):
     user_ip: str
     referer: str
     user_agent: str
-    meta: dict
+    meta: dict[str, Any]
 
-    def load(self, data: dict) -> None:
+    def load(self, data: dict[str, Any]) -> None:
         super().load(data)
         self.url = data.get("url", "")
         self.method = data.get("method", "")

pocketbase/models/record.py

@@ -1,15 +1,17 @@
 from __future__ import annotations
 
+from typing import Any
+
 from pocketbase.models.utils.base_model import BaseModel
 from pocketbase.utils import camel_to_snake
 
 
 class Record(BaseModel):
     collection_id: str
     collection_name: str
-    expand: dict
+    expand: dict[str, Any]
 
-    def load(self, data: dict) -> None:
+    def load(self, data: dict[str, Any]) -> None:
         super().load(data)
         self.expand = {}
         for key, value in data.items():
@@ -18,9 +20,9 @@ def load(self, data: dict) -> None:
         self.load_expanded()
 
     @classmethod
-    def parse_expanded(cls, data: dict):
+    def parse_expanded(cls, data: Any):
         if isinstance(data, list):
-            return [cls(a) for a in data]
+            return [cls(a) for a in data]  # type: ignore
         return cls(data)
 
     def load_expanded(self) -> None: