update to pocketbase version 0.8.0-rc2 (#4)

update to pocketbase version 0.8.0-rc1

fixed flake8 errors,
but there are still remaining errors:
F401 ... imported but unused
E501 line too long (... > 79 characters)
E722 do not use bare 'except'

Author: perfalle

pocketbase/client.py

@@ -1,46 +1,33 @@
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Dict
+from urllib.parse import quote, urlencode
 
 import httpx
 
-from pocketbase.services.admins import Admins
-from pocketbase.services.collections import Collections
-from pocketbase.services.logs import Logs
-from pocketbase.services.realtime import Realtime
-from pocketbase.services.records import Records
-from pocketbase.services.users import Users
-from pocketbase.services.settings import Settings
+from pocketbase.utils import ClientResponseError
+from pocketbase.models.record import Record
+from pocketbase.services.admin_service import AdminService
+from pocketbase.services.collection_service import CollectionService
+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
 
 
-class ClientResponseError(Exception):
-    url: str = ""
-    status: int = 0
-    data: dict = {}
-    is_abort: bool = False
-    original_error: Any | None = None
-
-    def __init__(self, *args, **kwargs) -> None:
-        super().__init__(*args)
-        self.url = kwargs.get("url", "")
-        self.status = kwargs.get("status", 0)
-        self.data = kwargs.get("data", {})
-        self.is_abort = kwargs.get("is_abort", False)
-        self.original_error = kwargs.get("original_error", None)
-
-
 class Client:
     base_url: str
     lang: str
     auth_store: BaseAuthStore
-    settings: Settings
-    admins: Admins
-    users: Users
-    collections: Collections
-    records: Records
-    logs: Logs
-    realtime: Realtime
+    settings: SettingsService
+    admins: AdminService
+    records: Record
+    collections: CollectionService
+    records: RecordService
+    logs: LogService
+    realtime: RealtimeService
+    record_service: Dict[str, RecordService]
 
     def __init__(
         self,
@@ -52,13 +39,18 @@ def __init__(
         self.lang = lang
         self.auth_store = auth_store or BaseAuthStore()  # LocalAuthStore()
         # services
-        self.admins = Admins(self)
-        self.users = Users(self)
-        self.records = Records(self)
-        self.collections = Collections(self)
-        self.logs = Logs(self)
-        self.settings = Settings(self)
-        self.realtime = Realtime(self)
+        self.admins = AdminService(self)
+        self.collections = CollectionService(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]
 
     def send(self, path: str, req_config: dict[str:Any]) -> Any:
         """Sends an api http request."""
@@ -68,12 +60,9 @@ def send(self, path: str, req_config: dict[str:Any]) -> Any:
         if self.auth_store.token and (
             "headers" not in config or "Authorization" not in config["headers"]
         ):
-            auth_type = "Admin"
-            if hasattr(self.auth_store.model, "verified"):
-                auth_type = "User"
             config["headers"] = config.get("headers", {})
             config["headers"].update(
-                {"Authorization": f"{auth_type} {self.auth_store.token}"}
+                {"Authorization": self.auth_store.token}
             )
         # build url + path
         url = self.build_url(path)
@@ -109,6 +98,21 @@ def send(self, path: str, req_config: dict[str:Any]) -> Any:
             )
         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 build_url(self, path: str) -> str:
         url = self.base_url
         if not self.base_url.endswith("/"):

pocketbase/models/__init__.py

@@ -3,4 +3,3 @@
 from .external_auth import ExternalAuth
 from .log_request import LogRequest
 from .record import Record
-from .user import User

pocketbase/models/admin.py

@@ -9,10 +9,8 @@
 class Admin(BaseModel):
     avatar: int
     email: str
-    last_reset_sent_at: str | datetime.datetime
 
     def load(self, data: dict) -> None:
         super().load(data)
         self.avatar = data.get("avatar", 0)
         self.email = data.get("email", "")
-        self.last_reset_sent_at = to_datetime(data.get("lastResetSentAt", ""))

pocketbase/models/collection.py

@@ -6,24 +6,41 @@
 
 class Collection(BaseModel):
     name: str
+    type: str
     schema: list[SchemaField]
     system: bool
     list_rule: str | None
     view_rule: str | None
     create_rule: str | None
     update_rule: str | None
     delete_rule: str | None
+    options: dict
 
     def load(self, data: dict) -> None:
         super().load(data)
         self.name = data.get("name", "")
         self.system = data.get("system", False)
+        self.type = data.get("type", "base")
+        self.options = data.get("options", {})
+
+        # rules
         self.list_rule = data.get("listRule", None)
         self.view_rule = data.get("viewRule", None)
         self.create_rule = data.get("createRule", None)
         self.update_rule = data.get("updateRule", None)
         self.delete_rule = data.get("deleteRule", "")
+
+        # schema
         schema = data.get("schema", [])
         self.schema = []
         for field in schema:
             self.schema.append(SchemaField(**field))
+
+    def is_base(self):
+        return self.type == 'base'
+
+    def is_auth(self):
+        return self.type == 'auth'
+
+    def is_single(self):
+        return self.type == 'single'

pocketbase/models/external_auth.py

@@ -4,12 +4,14 @@
 
 
 class ExternalAuth(BaseModel):
-    user_id: str
+    record_id: str
+    collection_id: str
     provider: str
     provider_id: str
 
     def load(self, data: dict) -> None:
         super().load(data)
-        self.user_id = data.get("userId", "")
+        self.record_id = data.get("recordId", "")
+        self.collection_id = data.get("collectionId", "")
         self.provider = data.get("provider", "")
         self.provider_id = data.get("providerId", "")

pocketbase/models/user.py

@@ -30,4 +30,4 @@ def load(self, data: dict) -> None:
     @property
     def is_new(self) -> bool:
         """Returns whether the current loaded data represent a stored db record."""
-        return not self.id or self.id == "00000000-0000-0000-0000-000000000000"
+        return not self.id

pocketbase/models/utils/base_model.py

@@ -1,7 +1,6 @@
-from .admins import Admins, AdminAuthResponse
-from .collections import Collections
-from .logs import Logs, HourlyStats
-from .realtime import Realtime
-from .records import Records
-from .settings import Settings
-from .users import Users, UserAuthResponse, AuthMethodsList, AuthProviderInfo
+from .admin_service import AdminService, AdminAuthResponse
+from .collection_service import CollectionService
+from .log_service import LogService, HourlyStats
+from .realtime_service import RealtimeService
+from .record_service import RecordService
+from .settings_service import SettingsService

pocketbase/services/__init__.py

@@ -16,13 +16,39 @@ def __init__(self, token: str, admin: Admin, **kwargs) -> None:
             setattr(self, key, value)
 
 
-class Admins(CrudService):
+class AdminService(CrudService):
     def decode(self, data: dict) -> BaseModel:
         return Admin(data)
 
     def base_crud_path(self) -> str:
         return "/api/admins"
 
+    def update(self, id: str, body_params: dict, query_params: dict) -> 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(AdminService).update(id, body_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:
+            pass
+        return item
+
+    def delete(self, id: str, body_params: dict, query_params: dict) -> BaseModel:
+        """
+        If the current `client.auth_store.model` matches with the deleted id,
+        then on success the `client.auth_store` will be cleared.
+        """
+        item = super(AdminService).delete(id, body_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:
+            pass
+        return item
+
     def auth_response(self, response_data: dict) -> AdminAuthResponse:
         """Prepare successful authorize response."""
         admin = self.decode(response_data.pop("admin", {}))
@@ -31,18 +57,18 @@ def auth_response(self, response_data: dict) -> AdminAuthResponse:
             self.client.auth_store.save(token, admin)
         return AdminAuthResponse(token=token, admin=admin, **response_data)
 
-    def auth_via_email(
+    def auth_with_password(
         self, email: str, password: str, body_params: dict = {}, query_params: dict = {}
     ) -> AdminAuthResponse:
         """
-        Authenticate an admin account by its email and password
+        Authenticate an admin account with its email and password
         and returns a new admin token and data.
 
         On success this method automatically updates the client's AuthStore data.
         """
-        body_params.update({"email": email, "password": password})
+        body_params.update({"identity": email, "password": password})
         response_data = self.client.send(
-            self.base_crud_path() + "/auth-via-email",
+            self.base_crud_path() + "/auth-with-password",
             {
                 "method": "POST",
                 "params": query_params,
@@ -52,7 +78,7 @@ def auth_via_email(
         )
         return self.auth_response(response_data)
 
-    def refresh(
+    def authRefresh(
         self, body_params: dict = {}, query_params: dict = {}
     ) -> AdminAuthResponse:
         """
@@ -63,7 +89,7 @@ def refresh(
         """
         return self.auth_response(
             self.client.send(
-                self.base_crud_path() + "/refresh",
+                self.base_crud_path() + "/auth-refresh",
                 {"method": "POST", "params": query_params, "body": body_params},
             )
         )

pocketbase/services/admins.py pocketbase/services/admin_service.pypocketbase/services/admins.py renamed to pocketbase/services/admin_service.py

@@ -5,7 +5,7 @@
 from pocketbase.models.collection import Collection
 
 
-class Collections(CrudService):
+class CollectionService(CrudService):
     def decode(self, data: dict) -> BaseModel:
         return Collection(data)