1+ from typing import List , Dict , Any , Optional
2+ from .query import prepare_grit_query
3+
4+ class GritDomain :
5+ def __init__ (self , service , path : str ):
6+ self .api = service .session
7+ self .base_url = service .base_url .rstrip ("/" )
8+ self .path = path .strip ("/" )
9+
10+ def _url (self , endpoint : str = "" ) -> str :
11+ return f"{ self .base_url } { self .path } { endpoint } .rstrip ("/" )
12+
13+ def detail (self , id : str ) -> Optional [Dict [str , Any ]]:
14+ try :
15+ resp = self .api .get (self ._url (f"detail/{ id } ))
16+ resp .raise_for_status ()
17+ return resp .json ()
18+ except Exception as e :
19+ if hasattr (e , 'response' ) and e .response is not None and e .response .status_code == 404 :
20+ return None
21+ raise
22+
23+ def dead_detail (self , id : str ) -> Optional [Dict [str , Any ]]:
24+ try :
25+ resp = self .api .get (self ._url (f"detail/{ id } ))
26+ resp .raise_for_status ()
27+ return resp .json ()
28+ except Exception as e :
29+ if hasattr (e , 'response' ) and e .response is not None and e .response .status_code == 404 :
30+ return None
31+ raise
32+
33+ def list (self , filters : Optional [List [Dict [str , Any ]]] = None ,
34+ order : Optional [Dict [str , str ]] = None ,
35+ cursor : Optional [str ] = None ) -> Dict [str , Any ]:
36+
37+ query = prepare_grit_query ({
38+ "filters" : filters ,
39+ "order" : order ,
40+ "cursor" : cursor
41+ })
42+
43+ url = f"{ self ._url ('list' )} { query } if query else self ._url ('list' )
44+ resp = self .api .get (url )
45+ resp .raise_for_status ()
46+
47+ return {
48+ "data" : resp .json (),
49+ "cursor" : resp .headers .get ('x-page-cursor' , '' )
50+ }
51+
52+ def dead_list (self , filters : Optional [List [Dict [str , Any ]]] = None ,
53+ order : Optional [Dict [str , str ]] = None ,
54+ cursor : Optional [str ] = None ) -> Dict [str , Any ]:
55+
56+ query = prepare_grit_query ({
57+ "filters" : filters ,
58+ "order" : order ,
59+ "cursor" : cursor
60+ })
61+
62+ url = f"{ self ._url ('dead_list' )} { query } if query else self ._url ('list' )
63+ resp = self .api .get (url )
64+ resp .raise_for_status ()
65+
66+ return {
67+ "data" : resp .json (),
68+ "cursor" : resp .headers .get ('x-page-cursor' , '' )
69+ }
70+
71+ def list_all (self , filters : Optional [List [Dict [str , Any ]]] = None ,
72+ order : Optional [str ] = None ) -> List [Dict [str , Any ]]:
73+ data = []
74+ cursor = None
75+
76+ while True :
77+ result = self .list (filters = filters , order = order , cursor = cursor )
78+ data .extend (result ["data" ])
79+
80+ if not result ["cursor" ] or cursor == result ["cursor" ]:
81+ break
82+
83+ cursor = result ["cursor" ]
84+
85+ return data
86+
87+ def list_one (self , filters : Optional [List [Dict [str , Any ]]] = None ,
88+ order : Optional [Dict [str , str ]] = None ) -> Optional [Dict [str , Any ]]:
89+ query = prepare_grit_query ({
90+ "filters" : filters ,
91+ "order" : order
92+ })
93+ url = f"{ self ._url ('list_one' )} { query }
94+ resp = self .api .get (url )
95+ resp .raise_for_status ()
96+
97+ result = resp .json ()
98+
99+ if not result or not isinstance (result , dict ) or not result .keys ():
100+ return None
101+
102+ return result
103+
104+ def add (self , payload : Dict [str , Any ]) -> Dict [str , Any ]:
105+ resp = self .api .post (self ._url ("add" ), json = payload )
106+ resp .raise_for_status ()
107+ return resp .json ()
108+
109+ def bulk_add (self , payload : List [Dict [str , Any ]]) -> Dict [str , List [str ]]:
110+ chunk_size = 25
111+ results = {"ids" : []}
112+
113+ for i in range (0 , len (payload ), chunk_size ):
114+ chunk = payload [i :i + chunk_size ]
115+ response = self .api .post (self ._url ("bulk_add" ), json = chunk )
116+ response .raise_for_status ()
117+ data = response .json ()
118+ results ["ids" ].extend (data .get ("ids" , []))
119+
120+ return results
121+
122+ def edit (self , id : str , data : Dict [str , Any ]) -> Dict [str , Any ]:
123+ resp = self .api .patch (self ._url (f"edit/{ id } ), json = data )
124+ resp .raise_for_status ()
125+ return resp .json ()
126+
127+ def delete (self , id : str ):
128+ resp = self .api .delete (self ._url (f"delete/{ id } ))
129+ resp .raise_for_status ()
130+
131+ def bulk (self , ids : List [str ]) -> List [Dict [str , Any ]]:
132+ resp = self .api .post (self ._url ("bulk" ), json = {"ids" : ids })
133+ resp .raise_for_status ()
134+ return resp .json ()
135+
136+ def bulk_all (self , ids : List [str ]) -> List [Dict [str , Any ]]:
137+ results = []
138+ size = 25
139+
140+ for i in range (0 , len (ids ), size ):
141+ chunk = ids [i :i + size ]
142+ res = self .bulk (chunk )
143+ results .extend (res )
144+
145+ return results
146+
147+ def select_raw (self , query : str , params : Dict [str , Any ]) -> Dict [str , Any ]:
148+ resp = self .api .post (self ._url ("select_raw" ), json = { "query" : query , "params" : params })
149+ resp .raise_for_status ()
150+ return resp .json ()
0 commit comments