Infer field specification and mapping from model definition

src/config.py

@@ -22,7 +22,7 @@ team = [
     "nle",
     "abz",
     "hael",
-    #"roen",
+    "roen",
     "sbel",
     "wasa",
     "huvw",

src/main.py

@@ -2,13 +2,12 @@ import click
 import random
 import odoo_client
 import json
-from tasks import fetch_tasks
 from rich import print, console
 from datetime import datetime
 from collections import defaultdict
 from typing import Dict, List, Optional
 from config import vfyd_tags, team
-
+from tasks import Task
 
 console = console.Console()
 
@@ -42,9 +41,8 @@ def dispatch():
         ("project_id", "=", 49),
         ("tag_ids", "not in", vfyd_tags),
     ]
-
     client = odoo_client.OdooClient()
-    records = fetch_tasks(client, domain)
+    records = client.web_search_read(Task, domain)
 
     random.shuffle(records)
     dispatch = defaultdict(list)
@@ -63,16 +61,15 @@ def dispatch():
 @cli.command()
 def status():
     """Show all unassigned tasks in the backlog."""
-    client = odoo_client.OdooClient()
 
+    client = odoo_client.OdooClient()
     domain = [
         ("stage_id", "=", 194),
         ("user_ids", "=", False),
         ("project_id", "=", 49),
         ("tag_ids", "not in", vfyd_tags),
     ]
-
+    tasks = client.web_search_read(Task, domain)
-    tasks = fetch_tasks(client, domain)
     for task in tasks:
         print(task)
 
@@ -80,20 +77,18 @@ def status():
 @cli.command()
 def check():
     """Check which previously dispatched tasks are still not completed."""
+
     dispatch = load_dispatch()
-
     reverse_dispatch = {v: k for k, vs in dispatch.items() for v in vs}
-
     client = odoo_client.OdooClient()
-
     domain = [
         ("id", "in", list(reverse_dispatch.keys())),
         ("tag_ids", "not in", vfyd_tags),
         ("stage_id", "=", 194),
         ("project_id", "=", 49),
     ]
+    tasks = client.web_search_read(Task, domain)
 
-    tasks = fetch_tasks(client, domain)
     not_done = defaultdict(list)
     for task in tasks:
         not_done[reverse_dispatch[task.id]].append(task)

src/model.py

@@ -0,0 +1,51 @@
+from typing import Dict, Self, Any, get_origin
+from dataclasses import fields, Field
+
+
+class OdooModel:
+    @classmethod
+    def specification(cls) -> Dict[str, Any]:
+        def field_spec(f: Field) -> Dict[str, Any]:
+            field_type = f.type
+            if f.name.endswith("_id") or f.name.endswith("_ids"):
+                return {
+                    f.name: {
+                        "fields": {
+                            "display_name": {},
+                        },
+                    },
+                }
+            elif field_type in {str, int, bool}:
+                return {f.name: {}}
+            else:
+                raise NotImplementedError(f"Unsupported field type: {field_type}")
+
+        spec = dict()
+        for f in fields(cls):
+            if f.init:
+                spec.update(field_spec(f))
+
+        return spec
+
+    @classmethod
+    def from_record(cls, record: Dict) -> Self:
+        init_args = {}
+        for f in fields(cls):
+            if f.init and f.name in record:
+                value = record[f.name]
+
+                if f.name.endswith("_id"):
+                    if f.type is str:
+                        value = value["display_name"]
+                    else:
+                        raise NotImplementedError()
+
+                if f.name.endswith("_ids"):
+                    if get_origin(f.type) is list and f.type.__args__[0] is str:
+                        value = [v["display_name"] for v in value]
+                    else:
+                        raise NotImplementedError()
+
+                init_args[f.name] = value
+
+        return cls(**init_args)

src/odoo_client.py

@@ -1,7 +1,11 @@
 import os
 import requests
 import random
-from typing import Any, Dict, List
+from typing import Any, Dict, List, TypeVar, Type
+from model import OdooModel
+
+
+T = TypeVar("T", bound=OdooModel)
 
 
 class OdooClient:
@@ -37,16 +41,16 @@ class OdooClient:
 
     def web_search_read(
         self,
-        model: str,
+        model: Type[T],
         domain: List[tuple],
-        spec: Dict[str, Any],
         *args: Any,
         **kwargs: Any,
-    ) -> List[Dict[str, Any]]:
+    ) -> List[T]:
-        args = [domain, spec, *args]
+        args = [domain, model.specification(), *args]
-        res = self.rpc(model, "web_search_read", *args, **kwargs)
+        res = self.rpc(model._name, "web_search_read", *args, **kwargs)
         res.raise_for_status()
         json = res.json()
         if json.get("error"):
             raise Exception(json.get("error"))
-        return json.get("result", {}).get("records", [])
+        records = json.get("result", {}).get("records", [])
+        return [model.from_record(record) for record in records]

src/tasks.py

@@ -0,0 +1,18 @@
+from dataclasses import dataclass, field
+from typing import List
+from model import OdooModel
+
+
+@dataclass
+class Task(OdooModel):
+    _name = "project.task"
+
+    id: int
+    name: str
+    tag_ids: List[str]
+    stage_id: str
+    priority: bool
+    url: str = field(init=False)
+
+    def __post_init__(self) -> None:
+        self.url = f"https://www.odoo.com/odoo/49/tasks/{self.id}"

tasks.py

@@ -1,60 +0,0 @@
-from dataclasses import dataclass, field
-from typing import Dict, List
-from odoo_client import OdooClient
-from rich.table import Table
-
-@dataclass
-class Task:
-    id: int
-    name: str
-    tags: List[str]
-    stage: str
-    priority: bool
-    url: str = field(init=False)
-
-    def __post_init__(self) -> None:
-        self.url = f"https://www.odoo.com/odoo/49/tasks/{self.id}"
-
-    def __rich__(self):
-        table = Table(show_header=True, header_style="bold magenta", box=None)
-        table.add_column("Field", style="cyan", justify="right")
-        table.add_column("Value", style="bold green")
-        table.add_row("ID", str(self.id))
-        table.add_row("Name", self.name)
-        table.add_row("Tags", ", ".join(self.tags) if self.tags else "[dim]No tags[/dim]")
-        table.add_row("Stage", f"[yellow]{self.stage}[/yellow]")
-        table.add_row("Priority", "[bold red]High[/bold red]" if self.priority else "[dim]Low[/dim]")
-        table.add_row("URL", f"[blue underline]{self.url}[/blue underline]")
-        return table
-
-
-def map_record_to_task(record: Dict) -> Task:
-    return Task(
-        id=record["id"],
-        name=record["name"],
-        tags=[tag["display_name"] for tag in record["tag_ids"]],
-        stage=record["stage_id"]["display_name"],
-        priority=record["priority"] == "1",
-    )
-
-
-def fetch_tasks(client: OdooClient, domain: List[tuple[str, str, str]]) -> List[Task]:
-    records = client.web_search_read(
-        "project.task",
-        domain,
-        spec={
-            "name": {},
-            "priority": {},
-            "tag_ids": {
-                "fields": {
-                    "display_name": {},
-                },
-            },
-            "stage_id": {
-                "fields": {
-                    "display_name": {},
-                },
-            },
-        },
-    )
-    return [map_record_to_task(record) for record in records]