Patch 0.12.2

pocketbase/client.py

@@ -24,7 +24,6 @@ class Client:
     auth_store: BaseAuthStore
     settings: SettingsService
     admins: AdminService
-    records: Record
     collections: CollectionService
     records: RecordService
     logs: LogService
@@ -59,13 +58,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]) -> httpx.Response:
+    def _send(self, path: str, req_config: dict[str, Any]) -> httpx.Response:
         """Sends an api http request returning response object."""
-        config = {"method": "GET"}
+        config: dict[str, Any] = {"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})
@@ -99,7 +98,7 @@ def _send(self, path: str, req_config: dict[str:Any]) -> httpx.Response:
                 headers=headers,
                 json=body,
                 data=data,
-                files=files,
+                files=files,  # type: ignore
                 timeout=self.timeout,
             )
         except Exception as e:
@@ -109,12 +108,12 @@ def _send(self, path: str, req_config: dict[str:Any]) -> httpx.Response:
             )
         return response
 
-    def send_raw(self, path: str, req_config: dict[str:Any]) -> bytes:
+    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:
+    def send(self, path: str, req_config: dict[str, Any]) -> Any:
         """Sends an api http request."""
         response = self._send(path, req_config)
         try:

pocketbase/models/__init__.py

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

pocketbase/models/file_upload.py

@@ -1,6 +1,7 @@
-from httpx._types import FileTypes
 from typing import Sequence, Union
 
+from httpx._types import FileTypes
+
 FileUploadTypes = Union[FileTypes, Sequence[FileTypes]]
 
 
@@ -9,6 +10,8 @@ def __init__(self, *args):
         self.files: FileUploadTypes = args
 
     def get(self, key: str):
-        if isinstance(self.files[0], Sequence) and not isinstance(self.files[0], str):
+        if isinstance(self.files[0], Sequence) and not isinstance(
+            self.files[0], str
+        ):
             return tuple((key, i) for i in self.files)
         return ((key, self.files),)

pocketbase/models/utils/base_model.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-from abc import ABC
 import datetime
+from abc import ABC
 
 from pocketbase.utils import to_datetime
 

pocketbase/services/__init__.py

@@ -1,6 +1,6 @@
-from .admin_service import AdminService, AdminAuthResponse
+from .admin_service import AdminAuthResponse, AdminService
 from .collection_service import CollectionService
-from .log_service import LogService, HourlyStats
+from .log_service import HourlyStats, LogService
 from .realtime_service import RealtimeService
 from .record_service import RecordService
 from .settings_service import SettingsService

pocketbase/services/admin_service.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
+from pocketbase.models.admin import Admin
 from pocketbase.models.utils.base_model import BaseModel
 from pocketbase.services.utils.crud_service import CrudService
-from pocketbase.models.admin import Admin
 from pocketbase.utils import validate_token
 
 
@@ -17,7 +17,7 @@ def __init__(self, token: str, admin: Admin, **kwargs) -> None:
             setattr(self, key, value)
 
     @property
-    def is_valid(self)->bool:
+    def is_valid(self) -> bool:
         return validate_token(self.token)
 
 
@@ -33,14 +33,16 @@ def update(self, id: str, body_params: dict, query_params={}) -> BaseModel:
         If the current `client.auth_store.model` matches with the updated id,
         then on success the `client.auth_store.model` will be updated with the result.
         """
-        item = super().update(id, body_params=body_params, query_params=query_params)
+        item = super().update(
+            id, body_params=body_params, query_params=query_params
+        )
         try:
             if (
                 self.client.auth_store.model.collection_id is not None
                 and item.id == self.client.auth_store.model.id
             ):
                 self.client.auth_store.save(self.client.auth_store.token, item)
-        except:
+        except Exception:
             pass
         return item
 
@@ -56,7 +58,7 @@ def delete(self, id: str, query_params={}) -> BaseModel:
                 and item.id == self.client.auth_store.model.id
             ):
                 self.client.auth_store.save(self.client.auth_store.token, item)
-        except:
+        except Exception:
             pass
         return item
 
@@ -69,7 +71,11 @@ def auth_response(self, response_data: dict) -> AdminAuthResponse:
         return AdminAuthResponse(token=token, admin=admin, **response_data)
 
     def auth_with_password(
-        self, email: str, password: str, body_params: dict = {}, query_params: dict = {}
+        self,
+        email: str,
+        password: str,
+        body_params: dict = {},
+        query_params: dict = {},
     ) -> AdminAuthResponse:
         """
         Authenticate an admin account with its email and password

pocketbase/services/backups_service.py

@@ -19,20 +19,31 @@ def create(self, name: str):
         )
 
     def get_full_list(self, query_params: dict = {}) -> list[Backup]:
-        response_data = self.client.send(self.base_path(), {"method": "GET", "params": query_params})
+        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}})
+        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"})
+        self.client.send(
+            "%s/%s" % (self.base_path(), key), {"method": "DELETE"}
+        )
 
-    def restore(self,  key: str):
-        self.client.send("%s/%s/restore" % (self.base_path(), key), {"method": "POST"})
+    def restore(self, key: str):
+        self.client.send(
+            "%s/%s/restore" % (self.base_path(), key), {"method": "POST"}
+        )
 
     def upload(self, file_upload: FileUpload):
-        self.client.send(self.base_path() + "/upload", {"method": "POST", "body": {"file": file_upload}})
+        self.client.send(
+            self.base_path() + "/upload",
+            {"method": "POST", "body": {"file": file_upload}},
+        )

pocketbase/services/collection_service.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
-from pocketbase.services.utils.crud_service import CrudService
-from pocketbase.models.utils.base_model import BaseModel
 from pocketbase.models.collection import Collection
+from pocketbase.models.utils.base_model import BaseModel
+from pocketbase.services.utils.crud_service import CrudService
 
 
 class CollectionService(CrudService):
@@ -30,7 +30,10 @@ def import_collections(
             {
                 "method": "PUT",
                 "params": query_params,
-                "body": {"collections": collections, "deleteMissing": delete_missing},
+                "body": {
+                    "collections": collections,
+                    "deleteMissing": delete_missing,
+                },
             },
         )
         return True

pocketbase/services/log_service.py

@@ -1,13 +1,13 @@
 from __future__ import annotations
 
+import datetime
 from dataclasses import dataclass
 from typing import Union
 from urllib.parse import quote
-import datetime
 
-from pocketbase.services.utils.base_service import BaseService
-from pocketbase.models.utils.list_result import ListResult
 from pocketbase.models.log_request import LogRequest
+from pocketbase.models.utils.list_result import ListResult
+from pocketbase.services.utils.base_service import BaseService
 from pocketbase.utils import to_datetime
 
 

pocketbase/services/realtime_service.py

@@ -1,12 +1,12 @@
 from __future__ import annotations
 
-from typing import Callable, List
 import dataclasses
 import json
+from typing import Callable, List
 
+from pocketbase.models.record import Record
 from pocketbase.services.utils.base_service import BaseService
 from pocketbase.services.utils.sse import Event, SSEClient
-from pocketbase.models.record import Record
 
 
 @dataclasses.dataclass
@@ -142,13 +142,17 @@ def _connect_handler(self, event: Event) -> None:
     def _connect(self) -> None:
         self._disconnect()
         self.event_source = SSEClient(self.client.build_url("/api/realtime"))
-        self.event_source.add_event_listener("PB_CONNECT", self._connect_handler)
+        self.event_source.add_event_listener(
+            "PB_CONNECT", self._connect_handler
+        )
 
     def _disconnect(self) -> None:
         self._remove_subscription_listeners()
         self.client_id = ""
         if not self.event_source:
             return
-        self.event_source.remove_event_listener("PB_CONNECT", self._connect_handler)
+        self.event_source.remove_event_listener(
+            "PB_CONNECT", self._connect_handler
+        )
         self.event_source.close()
         self.event_source = None