Coverage for src/fluree_py/http/endpoint/query.py: 83%

1from dataclasses import dataclass, replace 

2from typing import Any, Self 

3 

4from fluree_py.http.mixin import ( 

5 CommitableMixin, 

6 RequestMixin, 

7 WithContextMixin, 

8 WithWhereMixin, 

9) 

10from fluree_py.http.protocol.endpoint.query import ( 

11 GroupByClause, 

12 HavingClause, 

13 OrderByClause, 

14 QueryBuilder, 

15 ActiveIdentity, 

16) 

17from fluree_py.types.query.select import SelectArray, SelectObject 

18from fluree_py.types.query.where import WhereClause 

19 

20 

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

22class QueryBuilderImpl( 

23 WithContextMixin["QueryBuilderImpl"], 

24 WithWhereMixin["QueryBuilderImpl"], 

25 RequestMixin, 

26 CommitableMixin["QueryBuilderImpl"], 

27 QueryBuilder, 

28): 

29 """Implementation of a query operation builder.""" 

30 

31 endpoint: str 

32 ledger: str 

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

34 where: WhereClause | None = None 

35 group_by: GroupByClause | None = None 

36 having: HavingClause | None = None 

37 order_by: OrderByClause | None = None 

38 opts: ActiveIdentity | None = None 

39 select_fields: dict[str, Any] | list[str] | None = None 

40 

41 def with_group_by(self, fields: GroupByClause) -> Self: 

42 """Add group by clause to the query.""" 

43 return replace(self, group_by=fields) 

44 

45 def with_having(self, condition: HavingClause) -> Self: 

46 """Add having clause to the query.""" 

47 return replace(self, having=condition) 

48 

49 def with_order_by(self, fields: OrderByClause) -> Self: 

50 """Add order by clause to the query.""" 

51 return replace(self, order_by=fields) 

52 

53 def with_opts(self, opts: ActiveIdentity) -> Self: 

54 """Add query options to the query.""" 

55 return replace(self, opts=opts) 

56 

57 def with_select(self, fields: SelectObject | SelectArray) -> Self: 

58 """Add select fields to the query.""" 

59 return replace(self, select_fields=fields) 

60 

61 def get_url(self) -> str: 

62 """Get the endpoint URL for the query operation.""" 

63 return self.endpoint 

64 

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

66 """Build the request payload for the query operation.""" 

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

68 if self.context: 

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

70 result |= {"from": self.ledger} 

71 if self.where: 

72 result["where"] = self.where 

73 if self.group_by: 

74 result["groupBy"] = self.group_by 

75 if self.having: 

76 result["having"] = self.having 

77 if self.order_by: 

78 result["orderBy"] = self.order_by 

79 if self.opts: 

80 result["opts"] = self.opts 

81 if self.select_fields: 

82 result["select"] = self.select_fields 

83 return result