hubert/odoo-employee-schedule
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix bugs + use actions auth session get/set for sessions

Browse Source
This commit is contained in:
Hubert Van De Walle
2026-09-16 15:43:14 +02:00
parent 12440a11f9
commit 35337afdf0
3 changed files with 199 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/odoo_employee_schedule/cli.py
+4 -9
View File
@@ -13,6 +13,8 @@ from .client import OdooClient, OdooError
from .render import render_gantt
from .schedule import DAY_NAMES, get_schedules, resolve_employees
BASE_URL = "https://www.odoo.com"
def parse_date(value: str) -> date:
return date.fromisoformat(value)
@@ -80,13 +82,6 @@ def main() -> int:
default=None,
help="End date, YYYY-MM-DD, inclusive (default: same as --date-from)",
)
parser.add_argument("--url", default="https://www.odoo.com", help="Odoo base URL")
parser.add_argument(
"--session-file",
default=Path(__file__).resolve().parents[2] / "session_id.txt",
type=Path,
help="Path to a file containing the session_id cookie value",
)
parser.add_argument(
"--tz",
default=None,
@@ -130,8 +125,8 @@ def main() -> int:
parser.error("date_from must be before or equal to date_to")
try:
client = OdooClient.from_session_file(args.url, args.session_file, debug=args.debug)
except (FileNotFoundError, ValueError) as exc:
client = OdooClient.from_actions(BASE_URL, debug=args.debug)
except ValueError as exc:
print(f"error: {exc}", file=sys.stderr)
return 1
src/odoo_employee_schedule/client.py
+38 -7
View File
@@ -1,10 +1,17 @@
"""Minimal JSON-RPC client for Odoo's /web/dataset/call_kw route, authenticated via a session_id cookie."""
"""Minimal JSON-RPC client for Odoo's /web/dataset/call_kw route.
The session_id cookie is managed by the `actions` CLI's stored client profiles
(`actions auth session get/set <profile>`) rather than a local file: we read the
initial cookie from there, and if the server ever rotates it (a new Set-Cookie for
session_id on any response), we persist the new value back to the same profile so
the next run picks it up.
"""
from __future__ import annotations
import json
import subprocess
import sys
from pathlib import Path
import requests
@@ -17,20 +24,31 @@ class OdooError(RuntimeError):
self.data = data or {}
def _run_actions(*args: str) -> str:
try:
result = subprocess.run(["actions", *args], capture_output=True, text=True)
except FileNotFoundError as exc:
raise ValueError("the `actions` command isn't available on PATH") from exc
if result.returncode != 0:
raise ValueError((result.stderr or result.stdout).strip() or f"`actions {' '.join(args)}` failed")
return result.stdout.strip()
class OdooClient:
def __init__(self, base_url: str, session_id: str, debug: bool = False):
def __init__(self, base_url: str, session_id: str, profile: str, debug: bool = False):
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
self.session.cookies.set("session_id", session_id)
self.profile = profile
self.debug = debug
self._request_count = 0
@classmethod
def from_session_file(cls, base_url: str, session_file: Path, debug: bool = False) -> "OdooClient":
session_id = session_file.read_text().strip()
def from_actions(cls, base_url: str, profile: str = "odoo", debug: bool = False) -> "OdooClient":
session_id = _run_actions("auth", "session", "get", profile)
if not session_id:
raise ValueError(f"{session_file} is empty; put a session_id cookie value in it")
return cls(base_url, session_id, debug=debug)
raise ValueError(f"`actions auth session get {profile}` returned an empty session_id")
return cls(base_url, session_id, profile, debug=debug)
def _jsonrpc(self, endpoint: str, params: dict) -> dict:
self._request_count += 1
@@ -40,6 +58,7 @@ class OdooClient:
+ json.dumps(params, default=str),
file=sys.stderr,
)
previous_session_id = self.session.cookies.get("session_id")
resp = self.session.post(
f"{self.base_url}{endpoint}",
json={"jsonrpc": "2.0", "method": "call", "params": params},
@@ -47,6 +66,7 @@ class OdooClient:
timeout=30,
)
resp.raise_for_status()
self._persist_rotated_session_id(previous_session_id)
payload = resp.json()
if "error" in payload:
error = payload["error"]
@@ -54,6 +74,17 @@ class OdooClient:
raise OdooError(message, error.get("data"))
return payload.get("result")
def _persist_rotated_session_id(self, previous_session_id: str | None) -> None:
new_session_id = self.session.cookies.get("session_id")
if not new_session_id or new_session_id == previous_session_id:
return
if self.debug:
print(f"[debug] session_id rotated, persisting to actions profile {self.profile!r}", file=sys.stderr)
try:
_run_actions("auth", "session", "set", self.profile, new_session_id)
except ValueError as exc:
print(f"warning: failed to persist rotated session_id: {exc}", file=sys.stderr)
def call_kw(self, model: str, method: str, args: list, kwargs: dict | None = None):
return self._jsonrpc(
"/web/dataset/call_kw",
src/odoo_employee_schedule/schedule.py
+157 -136
View File
@@ -1,45 +1,33 @@
"""Compute several employees' per-day working schedules over a date range, in a
constant, small number of requests regardless of how many employees are asked for.
Two things have to be combined to get a correct answer, because Odoo keeps
them separate:
The weekly working-hours pattern is read directly from `resource.calendar.attendance`
(the raw dayofweek/day_period/week_type rows), not from the Gantt view's
`get_gantt_data`/`_gantt_unavailability`. That method branches on whether an employee
has an `hr.version` with a `contract_date_start` set: with one, it correctly computes
the pattern via `resource.calendar._unavailable_intervals_batch`; without one (e.g.
independent/freelance employees who are tracked as `hr.employee` but have no formal
contract), it falls back to `resource.resource._get_unavailable_intervals`, which for
such employees reports every day as unavailable. Reading the attendance rows ourselves
sidesteps that branch entirely and works the same way regardless of contract status.
1. The "unavailability" data behind the Gantt view
(`hr.leave.report.calendar.get_gantt_data`, called with a domain/groupby
that is *only* `employee_id` -- that specific shape triggers a
`hr_holidays_gantt` code path, `_get_gantt_data_groupby_employee`, that
fills in non-working time from the employee's working-hours calendar
even when they have zero `hr.leave` records in range). This gives the
weekly working-hours pattern (e.g. "doesn't work Fridays") merged with
company-wide/public holidays. The domain accepts `employee_id in [...]`
just as well as `employee_id = X`, and the result is keyed per employee,
so one call covers the whole list of employees.
2. The employees' own *validated* `hr.leave` records. These are
deliberately NOT included in (1): in the Gantt view a validated leave
is shown as its own colored bar on the employee's row, not as part of
the gray "unavailable" background, so the server-side unavailability
calculation explicitly filters those out (see
`hr_holidays_gantt/models/hr_leave.py::_gantt_unavailability`, which
restricts the leave domain to `resource_id = False`, i.e. company-wide
holidays only).
We get these from the SAME `get_gantt_data` call's `records`, rather
than a separate `hr.leave.search_read` -- `hr.leave` itself is normally
row-restricted by an `ir.rule` to "your own leaves + leaves of people
you manage", and Odoo record rules filter silently rather than raising,
so a plain search_read for a colleague outside that scope just comes
back empty with no error to catch. `hr.leave.report.calendar` is a
SQL-view-backed model (`_auto = False`) without that per-employee row
restriction, so its rows show the same colleague leave bars the Gantt
view itself already displays to any employee.
Public holidays are company/date-range-wide, not employee-specific, so
that lookup is naturally already a single shared call.
Three other things get merged in:
- Each employee's own leave requests, from a plain `hr.leave.report.calendar.search_read`
(not `hr.leave.search_read`: `hr.leave` is normally row-restricted by an `ir.rule` to
"your own leaves + leaves of people you manage", and record rules filter silently
rather than raising, so a colleague outside that scope just comes back empty with no
error to catch; `hr.leave.report.calendar` is a SQL-view-backed model without that
restriction).
- Company-wide (or per-calendar) public holidays, from `resource.calendar.leaves`.
- `work_entry_type_id` (the specific leave type name) needs `hr_holidays.group_hr_holidays_user`,
and unlike a plain `read()`, `web_read` raises `AccessError` if you request a field you
don't have group access to rather than silently omitting it, so it's never requested and
every leave is labeled generically as "Time Off".
"""
from __future__ import annotations
import math
import re
from collections import defaultdict
from dataclasses import dataclass, field
@@ -99,6 +87,7 @@ class ResolvedEmployee:
id: int
name: str
requested_as: str
calendar_id: int | None
def resolve_employees(client: OdooClient, identifiers: list[str]) -> list[ResolvedEmployee]:
@@ -122,29 +111,35 @@ def resolve_employees(client: OdooClient, identifiers: list[str]) -> list[Resolv
for tag in tag_tokens:
atoms.append(("name", "=ilike", f"%({tag})"))
rows = client.call_kw("hr.employee", "search_read", [or_domain(atoms), ["id", "name"]])
rows = client.call_kw(
"hr.employee", "search_read", [or_domain(atoms), ["id", "name", "resource_calendar_id"]]
)
by_id = {r["id"]: r["name"] for r in rows}
by_id = {r["id"]: r for r in rows}
by_tag: dict[str, list[dict]] = defaultdict(list)
for r in rows:
m = TAG_RE.search(r["name"])
if m:
by_tag[m.group(1).lower()].append(r)
def _make(row: dict, token: str) -> ResolvedEmployee:
calendar = row["resource_calendar_id"]
return ResolvedEmployee(row["id"], row["name"], token, calendar[0] if calendar else None)
resolved = []
problems = []
for token in identifiers:
if token.isdigit():
emp_id = int(token)
if emp_id in by_id:
resolved.append(ResolvedEmployee(emp_id, by_id[emp_id], token))
resolved.append(_make(by_id[emp_id], token))
else:
problems.append(f"{token!r}: no employee with that id")
continue
matches = by_tag.get(token.lower(), [])
if len(matches) == 1:
resolved.append(ResolvedEmployee(matches[0]["id"], matches[0]["name"], token))
resolved.append(_make(matches[0], token))
elif not matches:
problems.append(f"{token!r}: no employee name ends with '({token})'")
else:
@@ -174,105 +169,133 @@ class LeavePeriod:
validated: bool
def _fetch_gantt_data_batch(
def _fetch_leaves(
client: OdooClient, employee_ids: list[int], date_from: date, date_to: date, tz: ZoneInfo
) -> tuple[dict[int, list[tuple[datetime, datetime]]], dict[int, list[LeavePeriod]]]:
"""Weekly working-hours pattern + company-wide holidays (from `unavailabilities`) AND
each employee's own leave requests (from `records`), both from one call. See the module
docstring for why leave data comes from here rather than a plain `hr.leave.search_read`.
"""
) -> dict[int, list[LeavePeriod]]:
"""Every employee's own leave requests overlapping the range, in one call."""
start_utc = _local_midnight_to_utc_naive(date_from, tz)
stop_utc = _local_midnight_to_utc_naive(date_to + timedelta(days=1), tz)
result = client.call_kw(
"hr.leave.report.calendar",
"get_gantt_data",
[
[("employee_id", "in", employee_ids)],
["employee_id"],
{
"id": {},
"employee_id": {},
"start_datetime": {},
"stop_datetime": {},
"state": {},
},
],
{
"unavailability_fields": ["employee_id"],
"start_date": start_utc.strftime("%Y-%m-%d %H:%M:%S"),
"stop_date": stop_utc.strftime("%Y-%m-%d %H:%M:%S"),
"scale": "week",
},
)
raw_unavail = result["unavailabilities"].get("employee_id", {})
pattern_by_employee = {}
for emp_id in employee_ids:
items = raw_unavail.get(str(emp_id), [])
pattern_by_employee[emp_id] = [(_to_local(item["start"], tz), _to_local(item["stop"], tz)) for item in items]
# work_entry_type_id (and description, leave_id) require
# hr_holidays.group_hr_holidays_user -- unlike a plain read(), web_read
# (which get_gantt_data uses internally) raises AccessError if you
# request a field you don't have group access to, rather than silently
# omitting it, so those aren't in the read_specification above and
# every leave is labeled generically.
leaves_by_employee: dict[int, list[LeavePeriod]] = defaultdict(list)
for rec in result.get("records", []):
employee = rec.get("employee_id")
start = rec.get("start_datetime")
stop = rec.get("stop_datetime")
if not employee or not start or not stop:
continue
state = rec.get("state")
if state == "refuse":
continue
# employee_id normally comes back as [id, name], but resolving the
# name requires read access to hr.employee itself -- without that,
# web_read degrades it to a bare id instead of raising.
employee_id = employee[0] if isinstance(employee, (list, tuple)) else employee
leaves_by_employee[employee_id].append(
LeavePeriod(
start=_to_local(start, tz),
stop=_to_local(stop, tz),
label="Time Off",
validated=state == "validate",
)
)
return pattern_by_employee, dict(leaves_by_employee)
def _fetch_public_holidays(
client: OdooClient, date_from: date, date_to: date, tz: ZoneInfo
) -> list[tuple[datetime, datetime, str]]:
"""Company-wide (non-employee-specific) holidays, for labeling purposes only. Already shared across employees."""
rows = client.call_kw(
"resource.calendar.leaves",
"hr.leave.report.calendar",
"search_read",
[
[
("resource_id", "=", False),
("date_from", "<=", f"{date_to} 23:59:59"),
("date_to", ">=", f"{date_from} 00:00:00"),
("employee_id", "in", employee_ids),
("state", "not in", ("cancel", "refuse")),
("start_datetime", "<", stop_utc.strftime("%Y-%m-%d %H:%M:%S")),
("stop_datetime", ">=", start_utc.strftime("%Y-%m-%d %H:%M:%S")),
],
["name", "date_from", "date_to"],
["employee_id", "start_datetime", "stop_datetime", "state"],
],
)
out: dict[int, list[LeavePeriod]] = defaultdict(list)
for r in rows:
employee = r["employee_id"]
# employee_id normally comes back as [id, name], but resolving the
# name requires read access to hr.employee itself -- without that,
# it degrades to a bare id instead of raising.
employee_id = employee[0] if isinstance(employee, (list, tuple)) else employee
out[employee_id].append(
LeavePeriod(
start=_to_local(r["start_datetime"], tz),
stop=_to_local(r["stop_datetime"], tz),
label="Time Off",
validated=r["state"] == "validate",
)
)
return out
def _fetch_public_holidays(
client: OdooClient, calendar_ids: list[int], date_from: date, date_to: date, tz: ZoneInfo
) -> list[tuple[datetime, datetime, str]]:
"""Company-wide or per-calendar (non-employee-specific) holidays, for labeling
purposes only. Already shared across employees using the same calendars.
"""
start_utc = _local_midnight_to_utc_naive(date_from, tz)
stop_utc = _local_midnight_to_utc_naive(date_to + timedelta(days=1), tz)
domain = or_domain([("calendar_id", "=", False), ("calendar_id", "in", calendar_ids)]) + [
("resource_id", "=", False),
("date_from", "<", stop_utc.strftime("%Y-%m-%d %H:%M:%S")),
("date_to", ">=", start_utc.strftime("%Y-%m-%d %H:%M:%S")),
]
rows = client.call_kw("resource.calendar.leaves", "search_read", [domain, ["name", "date_from", "date_to"]])
return [(_to_local(r["date_from"], tz), _to_local(r["date_to"], tz), r["name"] or "Public Holiday") for r in rows]
# A half-day leave's date_from/date_to is the employee's actual half-day work
# block (Odoo computes it from their real attendance hours), which is small
# next to a generic 00:00-12:00/12:00-24:00 window -- measuring "how much of
# the window is covered" dilutes it away. Checking a fixed anchor point per
# half-day instead sidesteps that: does the leave/pattern-off time actually
# cover the point in time this half of the day is centered on.
@dataclass
class CalendarInfo:
two_weeks: bool
# (weekday 0=Monday..6=Sunday, day_period, week_type '0'/'1' or None)
rows: list[tuple[int, str, str | None]]
def _fetch_attendance_patterns(client: OdooClient, calendar_ids: list[int]) -> dict[int, CalendarInfo]:
"""Each calendar's raw weekly attendance pattern, for every calendar in use, in one call."""
if not calendar_ids:
return {}
rows = client.call_kw(
"resource.calendar.attendance",
"search_read",
[
[("calendar_id", "in", calendar_ids)],
["calendar_id", "dayofweek", "day_period", "week_type", "two_weeks_calendar"],
],
)
by_calendar: dict[int, CalendarInfo] = {}
for r in rows:
cal_id = r["calendar_id"][0]
info = by_calendar.setdefault(cal_id, CalendarInfo(two_weeks=r["two_weeks_calendar"], rows=[]))
info.rows.append((int(r["dayofweek"]), r["day_period"], r["week_type"] or None))
return by_calendar
def _week_type(day: date) -> str:
"""Same formula as `resource.calendar.attendance.get_week_type`: parity of the
number of weeks since January 1 of year 1, so an even week always follows an odd
one even in a 53-week year.
"""
return str(int(math.floor((day.toordinal() - 1) / 7) % 2))
def _pattern_for_day(calendar: CalendarInfo | None, day: date) -> tuple[bool, bool]:
"""(morning_working, afternoon_working) from the calendar's raw attendance rows.
An employee with no calendar at all has no data to say otherwise, so defaults to
available both halves rather than defaulting to always off.
"""
if calendar is None:
return True, True
weekday = day.weekday()
week_type = _week_type(day) if calendar.two_weeks else None
morning = afternoon = False
for dow, period, wt in calendar.rows:
if dow != weekday:
continue
if calendar.two_weeks and wt != week_type:
continue
if period == "full_day":
morning = afternoon = True
elif period == "morning":
morning = True
elif period == "afternoon":
afternoon = True
return morning, afternoon
# A half-day leave's start/stop is the employee's actual half-day work block, which
# is small next to a generic 00:00-12:00/12:00-24:00 window -- measuring "how much of
# the window is covered" dilutes it away. Checking a fixed anchor point per half-day
# instead sidesteps that.
MORNING_ANCHOR_HOUR = 10
AFTERNOON_ANCHOR_HOUR = 15
def _anchor(d: date, tz: ZoneInfo, hour: int) -> datetime:
return datetime.combine(d, time(hour, 0), tzinfo=tz)
def _point_covered(intervals: list[tuple[datetime, datetime]], point: datetime) -> bool:
return any(start <= point < stop for start, stop in intervals)
@@ -301,13 +324,12 @@ def _build_days(
date_from: date,
date_to: date,
tz: ZoneInfo,
pattern_intervals: list[tuple[datetime, datetime]],
calendar: CalendarInfo | None,
leave_periods: list[LeavePeriod],
holidays: list[tuple[datetime, datetime, str]],
explain: bool,
) -> list[DaySchedule]:
validated_leave_intervals = [(p.start, p.stop) for p in leave_periods if p.validated]
unavailable_intervals = pattern_intervals + validated_leave_intervals
days = []
d = date_from
@@ -315,19 +337,16 @@ def _build_days(
morning_start = datetime.combine(d, time(0, 0), tzinfo=tz)
noon = datetime.combine(d, time(12, 0), tzinfo=tz)
day_end = datetime.combine(d, time(23, 59, 59), tzinfo=tz)
morning_anchor = datetime.combine(d, time(MORNING_ANCHOR_HOUR, 0), tzinfo=tz)
afternoon_anchor = datetime.combine(d, time(AFTERNOON_ANCHOR_HOUR, 0), tzinfo=tz)
morning_working = not _point_covered(unavailable_intervals, morning_anchor)
afternoon_working = not _point_covered(unavailable_intervals, afternoon_anchor)
pattern_morning, pattern_afternoon = _pattern_for_day(calendar, d)
morning_working = pattern_morning and not _point_covered(
validated_leave_intervals, _anchor(d, tz, MORNING_ANCHOR_HOUR)
)
afternoon_working = pattern_afternoon and not _point_covered(
validated_leave_intervals, _anchor(d, tz, AFTERNOON_ANCHOR_HOUR)
)
# Reason categories are computed regardless of `explain` -- the gantt
# grid's colors need them even when text notes are suppressed. Only
# the specific name (leave type / holiday name) and free-text notes
# are gated behind `explain`, since the public-holiday lookup itself
# is skipped to save a request when `explain` is off (see
# get_schedules): in that case a public holiday can't be told apart
# from a plain weekly day off and falls back to that category.
morning_reason = REASON_WORKING
afternoon_reason = REASON_WORKING
morning_label = None
@@ -378,13 +397,15 @@ def get_schedules(
"""Compute the per-day schedule for every employee, in a constant number of requests."""
tz = ZoneInfo(tz_name)
employee_ids = [e.id for e in employees]
calendar_ids = sorted({e.calendar_id for e in employees if e.calendar_id})
pattern_by_employee, leaves_by_employee = _fetch_gantt_data_batch(client, employee_ids, date_from, date_to, tz)
calendars_by_id = _fetch_attendance_patterns(client, calendar_ids)
leaves_by_employee = _fetch_leaves(client, employee_ids, date_from, date_to, tz)
holidays: list[tuple[datetime, datetime, str]] = []
if explain:
try:
holidays = _fetch_public_holidays(client, date_from, date_to, tz)
holidays = _fetch_public_holidays(client, calendar_ids, date_from, date_to, tz)
except OdooError:
pass
@@ -393,7 +414,7 @@ def get_schedules(
date_from,
date_to,
tz,
pattern_by_employee.get(emp.id, []),
calendars_by_id.get(emp.calendar_id) if emp.calendar_id else None,
leaves_by_employee.get(emp.id, []),
holidays,
explain,