Skip to content

Commit

Permalink
refactor: changed and Provenance use neat object and list
Browse files Browse the repository at this point in the history
  • Loading branch information
doctrino committed Jun 27, 2024
1 parent 5539b17 commit 0740cdf
Show file tree
Hide file tree
Showing 2 changed files with 28 additions and 4 deletions.
18 changes: 17 additions & 1 deletion cognite/neat/_shared.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,24 @@ def dump(self, aggregate: bool = True) -> dict[str, Any]:
"""Return a dictionary representation of the object."""
raise NotImplementedError()

def _repr_html(self) -> str:
return pd.Series(self.dump(aggregate=True)).to_frame(name="value")._repr_html_()

T_NeatObject = TypeVar("T_NeatObject", bound=NeatObject)

@dataclass(frozen=True)
class FrozenNeatObject:
"""A frozen neat object can be dumped to a dictionary."""

@abstractmethod
def dump(self, aggregate: bool = True) -> dict[str, Any]:
"""Return a dictionary representation of the object."""
raise NotImplementedError()

def _repr_html(self) -> str:
return pd.Series(self.dump(aggregate=True)).to_frame(name="value")._repr_html_()


T_NeatObject = TypeVar("T_NeatObject", bound=NeatObject | FrozenNeatObject)


class NeatList(list, Sequence[T_NeatObject]):
Expand Down
14 changes: 11 additions & 3 deletions cognite/neat/graph/stores/_provenance.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,14 @@


import uuid
from collections import UserList
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import datetime
from typing import TypeVar

from rdflib import PROV, RDF, Literal, URIRef

from cognite.neat._shared import FrozenNeatObject, NeatList
from cognite.neat.constants import DEFAULT_NAMESPACE


Expand Down Expand Up @@ -64,7 +64,7 @@ def as_triples(self):


@dataclass(frozen=True)
class Change:
class Change(FrozenNeatObject):
agent: Agent
activity: Activity
entity: Entity
Expand All @@ -81,11 +81,19 @@ def record(cls, activity: str, start: datetime, end: datetime, description: str)
entity = Entity(was_generated_by=activity, was_attributed_to=agent)
return cls(agent, activity, entity, description)

def dump(self, aggregate: bool = True) -> dict[str, str]:
return {
"Agent": self.agent.id_,
"Activity": self.activity.id_,
"Entity": self.entity.id_,
"Description": self.description,
}


T_Change = TypeVar("T_Change", bound=Change)


class Provenance(UserList[T_Change]):
class Provenance(NeatList[Change]):
def __init__(self, changes: Sequence[T_Change] | None = None):
super().__init__(changes or [])

Expand Down

0 comments on commit 0740cdf

Please sign in to comment.