Coverage for src/fluree_py/http/endpoint/history.py: 93%

1from dataclasses import dataclass, replace 

2from typing import Any 

3 

4from fluree_py.http.mixin import ( 

5 CommitableMixin, 

6 RequestMixin, 

7 WithContextMixin, 

8) 

9from fluree_py.http.protocol.endpoint import HistoryBuilder 

10from fluree_py.types.common import TimeClause 

11from fluree_py.types.http.history import HistoryClause 

12 

13 

14@dataclass(frozen=True, kw_only=True) 

15class HistoryBuilderImpl( 

16 RequestMixin, 

17 WithContextMixin["HistoryBuilderImpl"], 

18 CommitableMixin["HistoryBuilderImpl"], 

19 HistoryBuilder, 

20): 

21 """Implementation of a history query builder.""" 

22 

23 endpoint: str 

24 ledger: str 

25 context: dict[str, Any] | None = None 

26 history: HistoryClause | None = None 

27 t: TimeClause | None = None 

28 commit_details: bool | None = None 

29 

30 def with_history(self, history: HistoryClause) -> "HistoryBuilderImpl": 

31 """Add history clause to the query.""" 

32 return replace(self, history=history) 

33 

34 def with_t(self, t: TimeClause) -> "HistoryBuilderImpl": 

35 """Add time clause to the query.""" 

36 return replace(self, t=t) 

37 

38 def with_commit_details(self, commit_details: bool) -> "HistoryBuilderImpl": 

39 """Add commit details flag to the query.""" 

40 return replace(self, commit_details=commit_details) 

41 

42 def get_url(self) -> str: 

43 """Get the endpoint URL for the history query operation.""" 

44 return self.endpoint 

45 

46 def build_request_payload(self) -> dict[str, Any]: 

47 """Build the request payload for the history query operation.""" 

48 result: dict[str, Any] = {} 

49 if self.context: 

50 result["@context"] = self.context 

51 result |= {"from": self.ledger, "history": self.history, "t": self.t} 

52 if self.commit_details: 

53 result["commitDetails"] = self.commit_details 

54 return result