1from __future__ import annotations 

2 

3from typing import TYPE_CHECKING 

4from urllib.parse import urljoin 

5 

6from requests import delete, get, post 

7from returns.pipeline import flow 

8from returns.result import safe 

9 

10from .types import JPHRequestOptions 

11 

12if TYPE_CHECKING: 

13 from typing import Dict, Optional 

14 

15 from requests.models import Response 

16 from returns.result import Result 

17 

18 from model import JPHType 

19 

20 

21class JPHClient: 

22 """Mostly generic python client to access JSONPlaceholder API on typicode.com 

23 We're using returns lib: https://github.com/dry-python/returns to catch all the little 

24 reasons that web requests might fail and robustly return wrapped objects (Success/Exception) 

25 Keep in mind that everything returned is wrapped""" 

26 

27 DEFAULT_ENDPOINT = "https://jsonplaceholder.typicode.com/" 

28 DEFAULT_HEADERS: Dict[str, str] = {} 

29 

30 _handlers = {"DELETE": delete, "GET": get, "POST": post} 

31 

32 def __init__( 

33 self, endpoint: Optional[str] = None, headers: Optional[Dict[str, str]] = None 

34 ): 

35 # NOTE: This is here to inject bad values to misconfigure the client 

36 

37 self._endpoint = endpoint or self.DEFAULT_ENDPOINT 

38 self._headers = headers or self.DEFAULT_HEADERS 

39 

40 @safe 

41 def _make_request(self, options: JPHRequestOptions) -> Response: 

42 """Dynamically selects the request handler and wraps it in a returns container for ease of use 

43 

44 Args: 

45 options (JPHRequestOptions): Options bag to generate the request with 

46 

47 Returns: 

48 Response: A regular requests.Response 

49 """ 

50 

51 with_or_without = ( 

52 f"{options.resource}/{options.resource_id}" 

53 if options.resource_id 

54 else options.resource 

55 ) 

56 

57 response = self._handlers[options.method]( 

58 urljoin(self._endpoint, with_or_without), 

59 data=(vars(options.data) if options.data else None), 

60 ) 

61 

62 response.raise_for_status() 

63 

64 return response 

65 

66 def _gen_flow( 

67 self, 

68 method: str, 

69 resource: str, 

70 resource_id: Optional[int] = None, 

71 data: Optional[JPHType] = None, 

72 ) -> Result[Response, Exception]: 

73 """Generate a flow with the options provided 

74 

75 Args: 

76 method (str): HTTP method as string 

77 resource (str): Desired resource on the API 

78 resource_id (Optional[int], optional): Optional resource identifier to locate. Defaults to None. 

79 data (Optional[JPHType], optional): Optional data object. For like POST requests. Defaults to None. 

80 

81 Returns: 

82 Result[Response, Exception]: Represents either a Response if Success or an Exception if Failure 

83 """ 

84 

85 return flow( 

86 JPHRequestOptions( 

87 method=method, resource=resource, resource_id=resource_id, data=data 

88 ), 

89 self._make_request, 

90 ) 

91 

92 def delete(self, resource: str, resource_id: int) -> Result[Response, Exception]: 

93 """Send an HTTP DELETE request via the client and wrap the Response in a Result 

94 

95 Args: 

96 resource (str): The desired resource 

97 resource_id (int): The identifier of the resource to delete 

98 

99 Returns: 

100 Result[Response, Exception]: Represents either a Response if Success or an Exception if Failure 

101 """ 

102 

103 return self._gen_flow("DELETE", resource, resource_id) 

104 

105 def get( 

106 self, resource: str, resource_id: Optional[int] = None 

107 ) -> Result[Response, Exception]: 

108 """Send an HTTP GET request via the client and wrap the Response in a Result 

109 

110 Args: 

111 resource (str): The desired resource 

112 resource_id (Optional[int], optional): Optional identifier of a specific Entity to locate. Defaults to None. 

113 

114 Returns: 

115 Result[Response, Exception]: Represents either a Response if Success or an Exception if Failure 

116 """ 

117 

118 return self._gen_flow("GET", resource, resource_id) 

119 

120 def post(self, resource: str, data: JPHType) -> Result[Response, Exception]: 

121 """Send an HTTP POST request via the client and wrap the Response in a Result 

122 

123 Args: 

124 resource (str): The desired resource 

125 data (JPHType): The payload object to post to the API 

126 

127 Returns: 

128 Result[Response, Exception]: Represents either a Response if Success or an Exception if Failure 

129 """ 

130 

131 return self._gen_flow("POST", resource, data=data)