Skip to content

Commit

Permalink
Add HistoryResource from #232
Browse files Browse the repository at this point in the history
Also directly validates results to `HistoryItem` instead of going through a dict.
  • Loading branch information
pederhan committed May 29, 2024
1 parent 1c8b117 commit 11d9f44
Showing 1 changed file with 36 additions and 31 deletions.
67 changes: 36 additions & 31 deletions mreg_cli/api/history.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import datetime
import json
from enum import Enum
from typing import Any
from typing import Any, Self

from pydantic import BaseModel, Field, field_validator

Expand All @@ -16,25 +16,36 @@


class HistoryResource(str, Enum):
"""History resources."""
"""History resources for the API.
Host = "host"
Group = "group"
HostPolicy_Role = "hostpolicy_role"
HostPolicy_Atom = "hostpolicy_atom"
Names represent resource names.
Values represent resource relations.
Access resource names and relation with the `resource()` and `relation()` methods.
"""

Host = "hosts"
Group = "groups"
HostPolicy_Role = "roles"
HostPolicy_Atom = "atoms"

@classmethod
def _missing_(cls, value: Any) -> HistoryResource:
v = str(value).lower()
for resource in cls:
if resource.value == v:
return resource
elif resource.name.lower() == v:
return resource
raise ValueError(f"Unknown resource {value}")

def relation(self) -> str:
"""Provide the relation of the resource."""
if self == HistoryResource.Host:
return "hosts"
if self == HistoryResource.Group:
return "groups"
if self == HistoryResource.HostPolicy_Role:
return "roles"
if self == HistoryResource.HostPolicy_Atom:
return "atoms"
"""Get the resource relation."""
return self.value

raise ValueError(f"Unknown resource {self}")
def resource(self) -> str:
"""Get the resource name."""
return self.name.lower()


class HistoryItem(BaseModel):
Expand Down Expand Up @@ -110,30 +121,24 @@ def output_multiple(cls, basename: str, items: list[HistoryItem]) -> None:
item.output(basename)

@classmethod
def get(cls, name: str, resource: HistoryResource) -> list[HistoryItem]:
def get(cls, name: str, resource: HistoryResource) -> list[Self]:
"""Get history items for a resource."""

def _get_history(params: dict[str, str]) -> list[dict[str, Any]]:
return get_typed(Endpoint.History, list[dict[str, Any]], params=params)

resource_value = resource.value
params: dict[str, str] = {"resource": resource_value, "name": name}
ret = _get_history(params)
params: dict[str, str] = {"resource": resource.resource(), "name": name}
ret = get_typed(Endpoint.History, list[cls], params=params)
if len(ret) == 0:
raise EntityNotFound(f"No history found for {name}")

model_ids = ",".join({str(i["model_id"]) for i in ret})
model_ids = ",".join({str(i.mid) for i in ret})
params = {
"resource": resource_value,
"resource": resource.resource(),
"model_id__in": model_ids,
}
ret = _get_history(params)
ret = get_typed(Endpoint.History, list[cls], params=params)

data_relation = resource.relation()
params = {
"data__relation": data_relation,
"data__relation": resource.relation(),
"data__id__in": model_ids,
}
ret.extend(_get_history(params))
ret.extend(get_typed(Endpoint.History, list[cls], params=params))

return [cls.model_validate(i) for i in ret]
return ret

0 comments on commit 11d9f44

Please sign in to comment.