From 47a3eed98b924845dc97db9a5e5893838d440e80 Mon Sep 17 00:00:00 2001 From: Hubert Van De Walle Date: Wed, 5 Mar 2025 13:57:58 +0100 Subject: [PATCH] Infer field specification and mapping from model definition --- config.py => src/config.py | 2 +- main.py => src/main.py | 17 +++----- src/model.py | 51 +++++++++++++++++++++++ odoo_client.py => src/odoo_client.py | 18 +++++---- src/tasks.py | 18 +++++++++ tasks.py | 60 ---------------------------- 6 files changed, 87 insertions(+), 79 deletions(-) rename config.py => src/config.py (97%) rename main.py => src/main.py (93%) create mode 100644 src/model.py rename odoo_client.py => src/odoo_client.py (75%) create mode 100644 src/tasks.py delete mode 100644 tasks.py diff --git a/config.py b/src/config.py similarity index 97% rename from config.py rename to src/config.py index 5b7e751..8e9349a 100644 --- a/config.py +++ b/src/config.py @@ -22,7 +22,7 @@ team = [ "nle", "abz", "hael", - #"roen", + "roen", "sbel", "wasa", "huvw", diff --git a/main.py b/src/main.py similarity index 93% rename from main.py rename to src/main.py index 9aa2c08..7ddce16 100644 --- a/main.py +++ b/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 = fetch_tasks(client, domain) + tasks = client.web_search_read(Task, 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) diff --git a/src/model.py b/src/model.py new file mode 100644 index 0000000..1a997b2 --- /dev/null +++ b/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) diff --git a/odoo_client.py b/src/odoo_client.py similarity index 75% rename from odoo_client.py rename to src/odoo_client.py index 72cbe98..6a73aa1 100644 --- a/odoo_client.py +++ b/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]]: - args = [domain, spec, *args] - res = self.rpc(model, "web_search_read", *args, **kwargs) + ) -> List[T]: + args = [domain, model.specification(), *args] + 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] diff --git a/src/tasks.py b/src/tasks.py new file mode 100644 index 0000000..26d3501 --- /dev/null +++ b/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}" diff --git a/tasks.py b/tasks.py deleted file mode 100644 index 6b7ea86..0000000 --- a/tasks.py +++ /dev/null @@ -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]