Use pydantic

Author: lsbardel

metablock/__init__.py

@@ -1,8 +1,8 @@
 from .client import Metablock
 from .components import MetablockEntity, MetablockError, MetablockResponseError
-from .extensions import Extension, Plugin
+from .extensions import Extension
 from .orgs import Org
-from .spaces import Block, Service, Space, SpaceExtension
+from .spaces import Block, Space, SpaceExtension
 from .user import User
 
 __version__ = "1.1.1"
@@ -13,10 +13,8 @@
     "MetablockResponseError",
     "MetablockEntity",
     "Space",
-    "Service",
     "Block",
     "Extension",
-    "Plugin",
     "SpaceExtension",
     "Org",
     "User",

metablock/cli.py

@@ -125,14 +125,14 @@ async def _apply(path: str, space_name: str, token: str, dry_run: bool) -> None:
         raise click.Abort()
     async with Metablock(auth_key=token) as mb:
         space = await mb.spaces.get(space_name)
-        svc = await space.blocks.get_list()
-        click.echo(f"space {space.name} has {len(svc)} blocks")
-        by_name = {s["name"]: s for s in svc}
+        blocks = await space.blocks.get_list()
+        click.echo(f"space {space.name} has {len(blocks)} blocks")
+        by_name = {s.name: s for s in blocks}
         for name, config in blocks:
             block = by_name.get(name)
             if block:
                 # update
-                await mb.blocks.update(block.id, **config)
+                await mb.blocks.patch(block.id, **config)
                 click.echo(f"updated block {name}")
             else:
                 # create

metablock/client.py

@@ -9,9 +9,9 @@
 from httpx import Response as ClientResponse
 
 from .components import Callback, HttpComponent, MetablockResponseError
-from .extensions import Extension, Extensions, Plugin, Plugins
-from .orgs import Org, Orgs
-from .spaces import Block, Blocks, Domains, Space, Spaces
+from .extensions import Extensions
+from .orgs import Orgs
+from .spaces import Blocks, Domains, Spaces
 from .user import User
 
 DEFAULT_USER_AGENT = f"Python/{'.'.join(map(str, sys.version_info[:2]))} metablock"
@@ -42,12 +42,11 @@ def __init__(
             "user-agent": user_agent,
             "accept": "application/json",
         }
-        self.orgs: Orgs = Orgs(self, Org)
-        self.spaces: Spaces = Spaces(self, Space)
-        self.blocks: Blocks = Blocks(self, Block)
-        self.plugins: Plugins = Plugins(self, Plugin)
-        self.extensions: Extensions = Extensions(self, Extension)
-        self.domains = Domains(self)
+        self.orgs: Orgs = Orgs(root=self, root_path="orgs")
+        self.spaces: Spaces = Spaces(root=self, root_path="spaces")
+        self.blocks: Blocks = Blocks(root=self, root_path="blocks")
+        self.extensions: Extensions = Extensions(root=self, root_path="extensions")
+        self.domains = Domains(root=self, root_path="domains")
 
     @property
     def cli(self) -> Self:
@@ -65,25 +64,31 @@ async def __aexit__(self, exc_type: type, exc_val: Any, exc_tb: Any) -> None:
         await self.close()
 
     async def spec(self) -> dict:
+        """Get the OpenAPI specification of the API"""
         return await self.request(f"{self.url}/openapi.json")
 
     async def get(self, url: str, **kwargs: Any) -> Any:
+        """Make a GET request to the API"""
         kwargs["method"] = "GET"
         return await self.request(url, **kwargs)
 
     async def patch(self, url: str, **kwargs: Any) -> Any:
+        """Make a PATCH request to the API"""
         kwargs["method"] = "PATCH"
         return await self.request(url, **kwargs)
 
     async def post(self, url: str, **kwargs: Any) -> Any:
+        """Make a POST request to the API"""
         kwargs["method"] = "POST"
         return await self.request(url, **kwargs)
 
     async def put(self, url: str, **kwargs: Any) -> Any:
+        """Make a PUT request to the API"""
         kwargs["method"] = "PUT"
         return await self.request(url, **kwargs)
 
     async def delete(self, url: str, **kwargs: Any) -> Any:
+        """Make a DELETE request to the API"""
         kwargs["method"] = "DELETE"
         return await self.request(url, **kwargs)
 
@@ -96,6 +101,7 @@ async def request(
         wrap: Any = None,
         **kw: Any,
     ) -> Any:
+        """Make a request to the API with the given method, url, headers and body."""
         if not self.session:
             self.session = AsyncClient()
         method = method or "GET"
@@ -122,22 +128,19 @@ async def handle_response(self, response: ClientResponse, wrap: Any = None) -> A
         data = response.json()
         return wrap(data) if wrap else data
 
-    async def get_user(self, **kw: Any) -> User:
-        kw.setdefault("wrap", self._user)
-        return await self.get(f"{self.url}/user", **kw)
+    async def get_user(self) -> User:
+        data = await self.get(f"{self.url}/user")
+        return User(root=self, root_path="user", **data)
 
-    async def update_user(self, **kw: Any) -> User:
-        kw.setdefault("wrap", self._user)
-        return await self.patch(f"{self.url}/user", **kw)
+    async def update_user(self, **params: Any) -> User:
+        data = await self.patch(f"{self.url}/user", **params)
+        return User(root=self, root_path="user", **data)
 
-    async def delete_user(self, **kw: Any) -> None:
-        return await self.delete(f"{self.url}/user", **kw)
+    async def delete_user(self) -> None:
+        return await self.delete(f"{self.url}/user")
 
     def get_default_headers(self) -> dict[str, str]:
         headers = self.default_headers.copy()
         if self.auth_key:
             headers[self.auth_key_name] = self.auth_key
         return headers
-
-    def _user(self, data: dict) -> User:
-        return User(self, data)

metablock/components.py

@@ -1,22 +1,12 @@
 from __future__ import annotations
 
 import json
-from abc import ABC, abstractmethod
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    AsyncIterator,
-    Awaitable,
-    Callable,
-    Generic,
-    Mapping,
-    TypeVar,
-    cast,
-)
+from typing import TYPE_CHECKING, Any, Awaitable, Callable, Protocol, runtime_checkable
 
 from httpx import Response as ClientResponse
+from pydantic import BaseModel, ConfigDict, Field
 
-from .utils import as_dict, as_params
+from .utils import as_dict
 
 if TYPE_CHECKING:  # pragma: no cover
     from .client import Metablock
@@ -45,212 +35,33 @@ def __str__(self) -> str:
         return json.dumps(self.message, indent=4)
 
 
-class HttpComponent(ABC):
+@runtime_checkable
+class HttpComponent(Protocol):
     @property
-    @abstractmethod
     def cli(self) -> Metablock:  # pragma: no cover
         ...
 
     @property
-    @abstractmethod
     def url(self) -> str:  # pragma: no cover
         ...
 
-    def __repr__(self) -> str:
-        return self.url
-
-    def __str__(self) -> str:
-        return self.__repr__()
 
+class MetablockComponent(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
 
-class Component(HttpComponent):
-    def __init__(
-        self, root: Metablock | CrudComponent | MetablockEntity, name: str = ""
-    ) -> None:
-        self.root = root
-        self.name = name or self.__class__.__name__.lower()
+    root: HttpComponent = Field(exclude=True)
+    root_path: str = Field(exclude=True)
 
     @property
     def cli(self) -> Metablock:
         return self.root.cli
 
     @property
     def url(self) -> str:
-        return f"{self.cli.url}/{self.name}"
-
-    @property
-    def parent_url(self) -> str:
-        return f"{self.root.url}/{self.name}"
-
-    @property
-    def is_entity(self) -> bool:
-        return isinstance(self.root, MetablockEntity)
+        return f"{self.root.url}/{self.root_path}"
 
 
-class MetablockEntity(HttpComponent):
+class MetablockEntity(MetablockComponent):
     """A Metablock entity"""
 
-    __slots__ = ("root", "data")
-
-    def __init__(
-        self,
-        root: Metablock | CrudComponent | MetablockEntity,
-        data: dict,
-    ) -> None:
-        self.root = root
-        self.data = data
-
-    def __repr__(self) -> str:
-        return repr(self.data)
-
-    def __getitem__(self, item: str) -> Any:
-        return self.data[item]
-
-    def __contains__(self, item: str) -> bool:
-        return item in self.data
-
-    def __eq__(self, other: Any) -> bool:
-        return isinstance(other, self.__class__) and self.data == other.data
-
-    def _asdict(self) -> dict:
-        return self.data
-
-    @property
-    def cli(self) -> Metablock:
-        return self.root.cli
-
-    @property
-    def id(self) -> str:
-        return self.data.get("id", "")
-
-    @property
-    def name(self) -> str:
-        return self.data.get("name", "")
-
-    @property
-    def url(self) -> str:
-        return "%s/%s" % (self.root.url, self.id)
-
-    def nice(self) -> str:
-        return json.dumps(self.data, indent=4)
-
-
-E = TypeVar("E", bound="MetablockEntity")
-
-
-class CrudComponent(Component, Generic[E]):
-    def __init__(
-        self,
-        root: Metablock | CrudComponent | MetablockEntity,
-        factory: type[E],
-        name: str = "",
-    ) -> None:
-        super().__init__(root, name)
-        self.factory = factory
-
-    async def paginate(self, **params: Any) -> AsyncIterator[E]:
-        next_ = self.list_create_url()
-        url_params: Any = as_params(**params)
-        while next_:
-            next_, data = await self.cli.request(
-                next_, params=url_params, callback=self._paginated
-            )
-            url_params = None
-            for d in data:
-                yield self.entity(d)
-
-    async def get_list(self, **kwargs: Any) -> list[E]:
-        url = self.list_create_url()
-        kwargs.setdefault("wrap", self.entity_list)
-        return cast(
-            list[E],
-            await self.cli.request(url, **kwargs),
-        )
-
-    async def get_full_list(self, **kwargs: Any) -> list[E]:
-        return [d async for d in self.paginate(**kwargs)]
-
-    async def get(self, id_: str, **kwargs: Any) -> E:
-        url = f"{self.url}/{id_}"
-        kwargs.setdefault("wrap", self.entity)
-        return cast(E, await self.cli.get(url, **kwargs))
-
-    async def has(self, id_: str, **kwargs: Any) -> bool:
-        url = f"{self.url}/{id_}"
-        return cast(bool, await self.cli.get(url, callback=self._head))
-
-    async def create(self, callback: Callback | None = None, **params: Any) -> E:
-        url = self.list_create_url()
-        return cast(
-            E,
-            await self.cli.post(url, json=params, callback=callback, wrap=self.entity),
-        )
-
-    async def update(
-        self, id_name: str, callback: Callback | None = None, **params: Any
-    ) -> E:
-        return cast(
-            E,
-            await self.cli.patch(
-                self.update_url(id_name),
-                json=params,
-                callback=callback,
-                wrap=self.entity,
-            ),
-        )
-
-    async def upsert(self, callback: Callback | None = None, **params: Any) -> E:
-        return cast(
-            E,
-            await self.cli.put(
-                self.url, json=params, callback=callback, wrap=self.entity
-            ),
-        )
-
-    async def delete(  # type: ignore
-        self,
-        id_name: str,
-        **kwargs: Any,
-    ) -> Any:
-        return await self.cli.delete(self.delete_url(id_name), **kwargs)
-
-    async def delete_all(self) -> int:
-        n = 0
-        async for entity in self.paginate():
-            await self.delete(entity.id)
-            n += 1
-        return n
-
-    def entity(self, data: dict) -> E:
-        return self.factory(self, data)
-
-    def entity_list(self, data: list[dict]) -> list[E]:
-        return [self.entity(d) for d in data]
-
-    def list_create_url(self) -> str:
-        return self.parent_url if self.is_entity else self.url
-
-    def update_url(self, id_name: str) -> str:
-        return f"{self.url}/{id_name}"
-
-    def delete_url(self, id_name: str) -> str:
-        return f"{self.url}/{id_name}"
-
-    # callbacks
-
-    async def _head(self, response: ClientResponse) -> bool:
-        if response.status_code == 404:
-            return False
-        elif response.status_code == 200:
-            return True
-        else:  # pragma: no cover
-            raise MetablockResponseError(response)
-
-    async def _paginated(self, response: ClientResponse) -> Any:
-        next_ = response.links.get("next")
-        if isinstance(next_, Mapping):
-            url = next_.get("url")
-        else:
-            url = None
-        data = await self.cli.handle_response(response)
-        return (url, data)
+    id: str = Field(description="The unique identifier of the entity")