refactored and introduced more optionals for stricter use

Author: antdjohns

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
 

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 = []

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(

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.
+    """

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,45 +27,79 @@ 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"
         )
 
         required_options = {
             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",
             X_POLYGON_EDGE_IP_ADDRESS: "test",
             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",
             "X-Polygon-Edge-IP-Address": "test",
             "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))
+