Use pydantic

Author: lsbardel

LICENSE

@@ -1,4 +1,4 @@
-Copyright (c) 2025 Quantmind
+Copyright (c) 2026 Quantmind
 
 Redistribution and use in source and binary forms, with or without modification,
 are permitted provided that the following conditions are met:

metablock/cli.py

@@ -125,9 +125,9 @@ 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)
-        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}
+        space_blocks = await space.blocks.get_list()
+        click.echo(f"space {space.name} has {len(space_blocks)} blocks")
+        by_name = {s.name: s for s in space_blocks}
         for name, config in blocks:
             block = by_name.get(name)
             if block:

metablock/client.py

@@ -128,8 +128,8 @@ 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) -> User:
-        data = await self.get(f"{self.url}/user")
+    async def get_user(self, **kwargs: Any) -> User:
+        data = await self.get(f"{self.url}/user", **kwargs)
         return User(root=self, root_path="user", **data)
 
     async def update_user(self, **params: Any) -> User:

metablock/orgs.py

@@ -3,22 +3,32 @@
 from typing import Any
 
 from pydantic import Field
+from typing_extensions import Annotated, Doc
 
 from .components import MetablockComponent, MetablockEntity
 from .extensions import Extension, Extensions
 from .spaces import Space
+from .utils import compact_dict
 
 
 class OrgSpaces(MetablockComponent):
-    async def get_list(self) -> list[Space]:
-        data = await self.cli.get(self.url)
-        return [
-            Space(root=self.cli, root_path=f"spaces/{s['name']}", **s) for s in data
-        ]
+    async def get_list(
+        self,
+        *,
+        limit: Annotated[int | None, Doc("Maximum number of spaces to return")] = None,
+        cursor: Annotated[str | None, Doc("Cursor for pagination")] = None,
+        **kwargs: Any,
+    ) -> list[Space]:
+        data = await self.cli.get(
+            self.url,
+            params=compact_dict(limit=limit, cursor=cursor),
+            **kwargs,
+        )
+        return [Space(root=self.cli.spaces, root_path=s["name"], **s) for s in data]
 
     async def create(self, **data: Any) -> Space:
         data = await self.cli.post(f"{self.url}", json=data)
-        return Space(root=self.cli, root_path=f"spaces/{data['name']}", **data)
+        return Space(root=self.cli.spaces, root_path=data["name"], **data)
 
 
 class OrgExtensions(Extensions):

metablock/spaces.py

@@ -32,7 +32,7 @@ class Spaces(MetablockComponent):
 
     async def get(self, space_id_or_name: str) -> Space:
         data = await self.cli.get(f"{self.url}/{space_id_or_name}")
-        return Space(root=self.cli, root_path=data["id"], **data)
+        return Space(root=self, root_path=data["id"], **data)
 
 
 class Route(BaseModel):
@@ -82,9 +82,6 @@ class Block(MetablockEntity):
     def deployments(self) -> Deployments:
         return Deployments(root=self, root_path="deployments")
 
-    async def config(self, *, callback: Any = None) -> dict:
-        return await self.cli.get(f"{self.url}/config", callback=callback)
-
     async def certificate(self) -> Certificate:
         data = await self.cli.get(f"{self.url}/certificate")
         return Certificate(**data)
@@ -124,7 +121,7 @@ async def get(self, block_id: str) -> Block:
         """Get a block by id"""
         data = await self.cli.get(f"{self.url}/{block_id}")
         return Block(
-            root=self.cli,
+            root=self,
             root_path=data["id"],
             is_root=data.pop("root", False),
             **data,
@@ -134,7 +131,7 @@ async def patch(self, block_id: str, **kwargs: Any) -> Block:
         """Update a block by id"""
         data = await self.cli.patch(f"{self.url}/{block_id}", json=kwargs)
         return Block(
-            root=self.cli,
+            root=self,
             root_path=data["id"],
             is_root=data.pop("root", False),
             **data,
@@ -146,15 +143,20 @@ async def get_list(self) -> list[Block]:
         """Get a list of blocks in the space"""
         data = await self.cli.get(self.url)
         return [
-            Block(root=self.cli, root_path=s["id"], is_root=s.pop("root", False), **s)
+            Block(
+                root=self.cli.blocks,
+                root_path=s["id"],
+                is_root=s.pop("root", False),
+                **s,
+            )
             for s in data
         ]
 
     async def create(self, name: str, **kwargs: Any) -> Block:
         """Create a new block in the space"""
         data = await self.cli.post(self.url, json=dict(name=name, **kwargs))
         return Block(
-            root=self.cli,
+            root=self.cli.cli,
             root_path=data["id"],
             is_root=data.pop("root", False),
             **data,

metablock/user.py

@@ -19,8 +19,8 @@ class OrgMember(BaseModel):
     roles: list[str] = Field(description="The user's roles in the organization")
 
     @classmethod
-    def create(cls, cli: Metablock, **data: Any) -> Self:
-        name = data["org"]["name"]
+    def from_cli(cls, cli: Metablock, **data: Any) -> Self:
+        name = data["org_name"]
         org = Org(
             root=cli,
             root_path=f"orgs/{name}",
@@ -48,7 +48,7 @@ class User(MetablockEntity):
     async def orgs(self, **kwargs: Any) -> list[OrgMember]:
         """List user organizations"""
         orgs = await self.cli.get(f"{self.url}/orgs", **kwargs)
-        return [OrgMember.create(self.cli, **data) for data in orgs]
+        return [OrgMember.from_cli(self.cli, **data) for data in orgs]
 
     async def get_permissions(self, **kwargs: Any) -> list[dict]:
         """List user permissions"""