92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
class OdooClient:
|
|
def __init__(self, url: str, database: str, username: str, password: str):
|
|
self.url = url.rstrip("/")
|
|
self.database = database
|
|
self.username = username
|
|
self.password = password
|
|
self.uid: int | None = None
|
|
|
|
async def _jsonrpc(self, endpoint: str, params: dict[str, Any]) -> Any:
|
|
payload = {
|
|
"jsonrpc": "2.0",
|
|
"method": "call",
|
|
"params": params,
|
|
"id": 1,
|
|
}
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(
|
|
f"{self.url}{endpoint}",
|
|
json=payload,
|
|
timeout=30.0,
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
if "error" in result:
|
|
raise Exception(f"Odoo error: {result['error']}")
|
|
return result.get("result")
|
|
|
|
async def authenticate(self) -> int:
|
|
self.uid = await self._jsonrpc("/web/session/authenticate", {
|
|
"db": self.database,
|
|
"login": self.username,
|
|
"password": self.password,
|
|
})
|
|
return self.uid
|
|
|
|
async def search_read(
|
|
self,
|
|
model: str,
|
|
domain: list | None = None,
|
|
fields: list[str] | None = None,
|
|
limit: int = 0,
|
|
order: str = "",
|
|
) -> list[dict[str, Any]]:
|
|
return await self._jsonrpc("/web/dataset/call_kw", {
|
|
"model": model,
|
|
"method": "search_read",
|
|
"args": [domain or []],
|
|
"kwargs": {
|
|
"fields": fields or [],
|
|
"limit": limit,
|
|
"order": order,
|
|
},
|
|
})
|
|
|
|
async def get_projects(self) -> list[dict[str, Any]]:
|
|
return await self.search_read(
|
|
model="project.project",
|
|
domain=[("active", "=", True)],
|
|
fields=["name", "task_count", "color"],
|
|
)
|
|
|
|
async def get_tasks(self, project_id: int | None = None) -> list[dict[str, Any]]:
|
|
domain: list = [("active", "=", True)]
|
|
if project_id:
|
|
domain.append(("project_id", "=", project_id))
|
|
return await self.search_read(
|
|
model="project.task",
|
|
domain=domain,
|
|
fields=[
|
|
"name", "project_id", "stage_id", "user_ids",
|
|
"priority", "date_deadline", "kanban_state",
|
|
],
|
|
)
|
|
|
|
async def get_calendar_events(
|
|
self, date_from: str, date_to: str
|
|
) -> list[dict[str, Any]]:
|
|
return await self.search_read(
|
|
model="calendar.event",
|
|
domain=[
|
|
("start", ">=", f"{date_from} 00:00:00"),
|
|
("start", "<=", f"{date_to} 23:59:59"),
|
|
],
|
|
fields=["name", "start", "stop", "location", "description", "partner_ids"],
|
|
order="start asc",
|
|
)
|