Coverage for mindsdb / integrations / handlers / frappe_handler / frappe_client.py: 0%

1import json 

2import requests 

3from typing import Dict, List 

4 

5 

6class FrappeClient(object): 

7 """Client to interact with the Frappe API. 

8 

9 Attributes: 

10 domain (str): Path to Frappe domain to use (e.g. https://mindsdbfrappe.com). 

11 access_token (str): Frappe authorization token to use for all API requests. 

12 """ 

13 

14 def __init__( 

15 self, 

16 domain: str, 

17 access_token: str): 

18 self.domain = domain 

19 self.base_url = f'{self.domain}/api' 

20 self.access_token = access_token 

21 

22 self.headers = { 

23 'Authorization': f'token {self.access_token}', 

24 } 

25 

26 def get_document(self, doctype: str, name: str) -> Dict: 

27 """Gets a document matching the given doctype from Frappe. 

28 

29 See https://frappeframework.com/docs/v14/user/en/api/rest#listing-documents 

30 Args: 

31 doctype (str): The document type to retrieve. 

32 name (str): Name of the document. 

33 """ 

34 document_response = requests.get( 

35 f'{self.base_url}/resource/{doctype}/{name}', 

36 headers=self.headers) 

37 if not document_response.ok: 

38 document_response.raise_for_status() 

39 return document_response.json()['data'] 

40 

41 def get_documents(self, doctype: str, limit: int = None, fields: List[str] = None, filters: List[List] = None) -> List[Dict]: 

42 """Gets all documents matching the given doctype from Frappe. 

43 

44 See https://frappeframework.com/docs/v14/user/en/api/rest#listing-documents 

45 Args: 

46 doctype (str): The document type to retrieve. 

47 limit (int): At most, how many messages to return. 

48 filters (List[List]): List of filters in the form [field, operator, value] e.g. ["amount", ">", 50] 

49 """ 

50 params = { 

51 'fields': json.dumps(["*"]) 

52 } 

53 if limit is not None: 

54 params['limit_page_length'] = limit 

55 if filters is not None: 

56 params['filters'] = json.dumps(filters) 

57 if fields is not None: 

58 params['fields'] = json.dumps(fields) 

59 documents_response = requests.get( 

60 f'{self.base_url}/resource/{doctype}/', 

61 params=params, 

62 headers=self.headers, 

63 allow_redirects=False) 

64 if documents_response.is_redirect: 

65 # We have to manually redirect to preserve the 'Authorization' header. 

66 # See https://github.com/request/request/pull/1184/commits/210b326fd8625f358e06c59dc11e74468b1de515. 

67 redirect_url = documents_response.headers.get('location', None) 

68 if redirect_url is None: 

69 raise requests.HTTPError('Could not find redirect URL') 

70 documents_response = requests.get( 

71 redirect_url, 

72 params=params, 

73 headers=self.headers, 

74 allow_redirects=False) 

75 

76 if not documents_response.ok: 

77 documents_response.raise_for_status() 

78 return documents_response.json()['data'] 

79 

80 def post_document( 

81 self, 

82 doctype: str, 

83 data: Dict): 

84 """Creates a new document of the given doctype. 

85 See https://frappeframework.com/docs/v14/user/en/api/rest#listing-documents 

86 

87 Args: 

88 doctype (str): Type of the document to create. 

89 data (Dict): Document object. 

90 """ 

91 post_response = requests.post( 

92 f'{self.base_url}/resource/{doctype}', 

93 json=data, 

94 headers=self.headers) 

95 if not post_response.ok: 

96 if 400 <= post_response.status_code < 600: 

97 raise requests.HTTPError(f'{post_response.reason}: {post_response.text}', response=post_response) 

98 return post_response.json()['data'] 

99 

100 def ping(self) -> bool: 

101 """Sends a basic request to the Frappe API to see if it succeeds. 

102 

103 Returns whether or not the connection to the Frappe API is valid. 

104 See https://frappeframework.com/docs/v14/user/en/api/rest#1-token-based-authentication 

105 """ 

106 

107 # No ping or similar endpoint exists, so we'll try getting the logged in user. 

108 user_response = requests.get( 

109 f'{self.base_url}/method/frappe.auth.get_logged_user', 

110 headers=self.headers) 

111 return user_response.ok