From 1f22d3590b33b4864af3b5f1f041756534aeb8d1 Mon Sep 17 00:00:00 2001 From: antdjohns <114414459+antdjohns@users.noreply.github.com> Date: Thu, 8 Dec 2022 15:10:11 -0800 Subject: [PATCH] refactored and introduced more optionals for stricter use --- examples/launchpad/README.md | 41 +++++++----- examples/launchpad/launchpad.py | 12 ++-- polygon/rest/base.py | 10 +-- polygon/rest/models/request.py | 108 ++++++++++++++++++------------ test_rest/models/test_requests.py | 57 ++++++++++++---- 5 files changed, 146 insertions(+), 82 deletions(-) diff --git a/examples/launchpad/README.md b/examples/launchpad/README.md index 19a68158..53301593 100644 --- a/examples/launchpad/README.md +++ b/examples/launchpad/README.md @@ -1,19 +1,28 @@ -# LaunchPad +# Launchpad Users of the Launchpad product will need to pass in certain headers in order to make API requests. -## EdgeHeaders -EdgeHeaders can be passed into request calls, the additional parameter is available in all reference client functions, -get_aggs, and snapshots + ```python -###### X-Polygon-Edge-ID -...[DESCRIPTION PENDING] -###### X-Polygon-Edge-IP-Address -...[DESCRIPTION PENDING] -###### X-Polygon-Edge-User-Agent -...[DESCRIPTION PENDING] +# import RESTClient +from polygon import RESTClient +from polygon.rest.models.request import RequestOptionBuilder -## Example +# create client +c = RESTClient(api_key="API_KEY") + +# create request options +options = RequestOptionBuilder().edge_headers( + edge_id="YOUR_EDGE_ID", # required + edge_ip_address="IP_ADDRESS", # required +) +# get response +res = c.get_aggs("AAPL", 1, "day", "2022-04-04", "2022-04-04", options=options) + +# do something with response + + ``` +Launchpad users can also provide the optional User Agent value describing their Edge User's origination request. ```python @@ -25,15 +34,15 @@ from polygon.rest.models.request import RequestOptionBuilder c = RESTClient(api_key="API_KEY") # create request options -options = RequestOptionBuilder().required_edge_headers( +options = RequestOptionBuilder().edge_headers( edge_id="YOUR_EDGE_ID", # required edge_ip_address="IP_ADDRESS" # required -).optional_edge_headers( - user_agent="USER_AGENT_ID" # optional -) +).update_edge_header( + edge_user="EDGE_USER" # optional + ) # get response -res = c.get_ticker_events("META", options=options) +res = c.get_aggs("AAPL", 1, "day", "2022-04-04", "2022-04-04", options=options) # do something with response diff --git a/examples/launchpad/launchpad.py b/examples/launchpad/launchpad.py index 1af727e1..ecb02ceb 100644 --- a/examples/launchpad/launchpad.py +++ b/examples/launchpad/launchpad.py @@ -11,16 +11,18 @@ def get_aggs_launchpad(): Example: `options = RequestOptionBuilder(edge_id="", edge_ip_address="") - or you can use the builder patten + or you can use the builder patten for future modifications to + underlying edge header dictionary. Example: options = RequestOptionBuilder() - .required_edge_headers(edge_id="EDGE_ID", edge_ip_address="EDGE_ID_ADDRESS") - .optional_edge_headers(user_agent="EDGE_USER_AGENT") + .edge_headers(edge_id="EDGE_ID", edge_ip_address="EDGE_ID_ADDRESS") + .update(edge_id="NEW") + options = options.update_edge_header(edge_ip_address="NEW_IP") """ options = ( RequestOptionBuilder() - .required_edge_headers(edge_id="EDGE_ID", edge_ip_address="EDGE_ID_ADDRESS") - .optional_edge_headers(user_agent="EDGE_USER_AGENT") + .edge_headers(edge_id="EDGE_ID", edge_ip_address="EDGE_ID_ADDRESS") + .update_edge_header(edge_user="EDGE_USER") ) trades = [] diff --git a/polygon/rest/base.py b/polygon/rest/base.py index afef424e..b376fe9b 100644 --- a/polygon/rest/base.py +++ b/polygon/rest/base.py @@ -79,16 +79,14 @@ def _get( params = {str(k): str(v) for k, v in params.items() if v is not None} logger.debug("_get %s params %s", path, params) - option = RequestOptionBuilder() if options is None else options + option = options if options is not None else RequestOptionBuilder() resp = self.client.request( "GET", self.BASE + path, fields=params, retries=self.retries, - headers=self._concat_headers( - option.edge_headers - ), # merge supplied headers with standard headers + headers=self._concat_headers(option.headers), ) if resp.status != 200: @@ -159,7 +157,9 @@ def _get_params( return params - def _concat_headers(self, headers: Dict[str, str]) -> Dict[str, str]: + def _concat_headers(self, headers: Optional[Dict[str, str]]) -> Dict[str, str]: + if headers is None: + return {**self.headers} return {**headers, **self.headers} def _paginate_iter( diff --git a/polygon/rest/models/request.py b/polygon/rest/models/request.py index adb2ace9..b732b5f9 100644 --- a/polygon/rest/models/request.py +++ b/polygon/rest/models/request.py @@ -20,65 +20,85 @@ def __init__( :param edge_ip_address: is a required Launchpad header. It denotes the originating IP Address of the Edge User :param edge_user: is an optional Launchpad header. It denotes the originating UserAgent of the Edge User requesting data. """ - self.edge_headers: Dict[str, str] = {} - self.__handle_edge_header_options( - edge_id=edge_id, edge_ip_address=edge_ip_address, edge_user=edge_user - ) + self.headers: Optional[Dict[str, str]] = None + if edge_id is not None and edge_ip_address is not None: + self.edge_headers( + edge_id=edge_id, edge_ip_address=edge_ip_address, edge_user=edge_user + ) - def __handle_edge_header_options( + def edge_headers( self, - edge_id: Optional[str], + edge_id: Optional[str] = None, edge_ip_address: Optional[str] = None, edge_user: Optional[str] = None, ): - edge_headers = {} - if edge_id is not None: - edge_headers[X_POLYGON_EDGE_ID] = edge_id - if edge_ip_address is not None: - edge_headers[X_POLYGON_EDGE_IP_ADDRESS] = edge_ip_address + """ + require_edge_headers adds required headers to the headers' dictionary + :param edge_id: is a required Launchpad header. It identifies the Edge User requesting data + :param edge_ip_address: is a required Launchpad header. It denotes the originating IP Address of the Edge User + requesting data + :param edge_user: user_agent: is an optional Launchpad header. It denotes the originating UserAgent of the Edge + User requesting data + :return ResponseOptionBuilder + """ + if edge_id is None or edge_ip_address is None: + raise RequestOptionError(f"edge_id and edge_ip_address required.") + + edge_headers: Dict[str, str] = { + X_POLYGON_EDGE_ID: edge_id, + X_POLYGON_EDGE_IP_ADDRESS: edge_ip_address, + } + if edge_user is not None: edge_headers[X_POLYGON_EDGE_USER_AGENT] = edge_user - self.__set_edge_headers(edge_headers) - def __set_edge_headers(self, headers: Dict[str, str]): - self.edge_headers = headers + self._add_to_edge_headers(**edge_headers) - def __add_to_edge_headers(self, **headers): - for k, v in headers.items(): - self.edge_headers[k] = v + return self - def required_edge_headers( + def update_edge_header( self, - edge_id: str, - edge_ip_address: str, + edge_id: Optional[str] = None, + edge_ip_address: Optional[str] = None, + edge_user: Optional[str] = None, ): """ - require_edge_headers adds required headers to the headers' dictionary + used to change individual edge elements of underlying headers' dictionary. :param edge_id: is a required Launchpad header. It identifies the Edge User requesting data :param edge_ip_address: is a required Launchpad header. It denotes the originating IP Address of the Edge User - requesting data. - :return: RequestOptionBuilder + requesting data + :param edge_user: user_agent: is an optional Launchpad header. It denotes the originating UserAgent of the Edge + User requesting data + :return: """ - self.__add_to_edge_headers( - **{ - X_POLYGON_EDGE_ID: edge_id, - X_POLYGON_EDGE_IP_ADDRESS: edge_ip_address, - } # object destructure is needed for correct key formatting. - ) - return self + if self.headers is None: + raise RequestOptionError( + "must set required fields prior to using update function." + ) + edge_headers: Dict[str, str] = {} + + if edge_id is not None: + edge_headers[X_POLYGON_EDGE_ID] = edge_id + + if edge_ip_address is not None: + edge_headers[X_POLYGON_EDGE_IP_ADDRESS] = edge_ip_address + + if edge_user is not None: + edge_headers[X_POLYGON_EDGE_USER_AGENT] = edge_user + + self._add_to_edge_headers(**edge_headers) - def optional_edge_headers( - self, - user_agent: str, - ): - """ - edge_user_agent_header is used to add the optional X-Polygon-Edge-User-Agent key to the header dictionary - :param user_agent: is an optional Launchpad header. It denotes the originating UserAgent of the Edge User requesting data. - :return: RequestOptionBuilder - """ - self.__add_to_edge_headers( - **{ - X_POLYGON_EDGE_USER_AGENT: user_agent, - } - ) return self + + def _add_to_edge_headers(self, **headers): + if self.headers is None: + self.headers = {} + + for k, v in headers.items(): + self.headers[k] = v + + +class RequestOptionError(Exception): + """ + Missing required option. + """ diff --git a/test_rest/models/test_requests.py b/test_rest/models/test_requests.py index 597a2415..e7233933 100644 --- a/test_rest/models/test_requests.py +++ b/test_rest/models/test_requests.py @@ -13,8 +13,8 @@ class RequestTest(unittest.TestCase): def test_empty_request_options(self): options = RequestOptionBuilder() - expected_edge_headers = {} - assert expected_edge_headers == options.edge_headers + expected_edge_headers = None + assert expected_edge_headers == options.headers def test_request_options_with_initialized_values(self): options = RequestOptionBuilder( @@ -27,10 +27,10 @@ def test_request_options_with_initialized_values(self): X_POLYGON_EDGE_USER_AGENT: "test", } - assert expected_object == options.edge_headers + assert expected_object == options.headers def test_request_options_builder(self): - options = RequestOptionBuilder().required_edge_headers( + options = RequestOptionBuilder().edge_headers( edge_id="test", edge_ip_address="test" ) @@ -38,8 +38,8 @@ def test_request_options_builder(self): X_POLYGON_EDGE_ID: "test", X_POLYGON_EDGE_IP_ADDRESS: "test", } - print(options.edge_headers, required_options) - self.assertDictEqual(required_options, options.edge_headers) + print(options.headers, required_options) + self.assertDictEqual(required_options, options.headers) all_options = { X_POLYGON_EDGE_ID: "test", @@ -47,20 +47,45 @@ def test_request_options_builder(self): X_POLYGON_EDGE_USER_AGENT: "test", } - options = options.optional_edge_headers("test") - self.assertDictEqual(all_options, options.edge_headers) + options = options.update_edge_header(edge_user="test") + self.assertDictEqual(all_options, options.headers) + + def test_header_update(self): - def test_header_combination(self): - client = RESTClient(api_key="test") options = RequestOptionBuilder( edge_id="test", edge_ip_address="test", edge_user="test" ) # this mocks the expected behavior during the request.get function call # in the BaseClient. - headers = client._concat_headers(options.edge_headers) + headers = options.headers + + expected_headers = { + "X-Polygon-Edge-ID": "test", + "X-Polygon-Edge-IP-Address": "test", + "X-Polygon-Edge-User-Agent": "test", + } + + self.assertDictEqual(headers, expected_headers) expected_headers = { + "X-Polygon-Edge-ID": "test2", + "X-Polygon-Edge-IP-Address": "test2", + "X-Polygon-Edge-User-Agent": "test2", + } + + options = options.update_edge_header( + edge_id="test2", edge_ip_address="test2", edge_user="test2" + ) + + self.assertDictEqual(options.headers, expected_headers) + + def test_clint_headers_concat(self): + client = RESTClient(api_key="test") + options = RequestOptionBuilder("test", "test", "test") + concat_headers = client._concat_headers(options.headers) + + expected = { "Authorization": "Bearer test", "User-Agent": "Polygon.io PythonClient/0.0.0", "X-Polygon-Edge-ID": "test", @@ -68,4 +93,12 @@ def test_header_combination(self): "X-Polygon-Edge-User-Agent": "test", } - self.assertDictEqual(headers, expected_headers) + self.assertDictEqual(concat_headers, expected) + + headers = None + expected = { + "Authorization": "Bearer test", + "User-Agent": "Polygon.io PythonClient/0.0.0", + } + + self.assertDictEqual(expected, client._concat_headers(headers))