Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Implicit Translation Table Population #7

Merged
merged 1 commit into from
Feb 24, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ pip install cnbc
## Utilization

### APIWrapper
The APIWrapper class is used to make requests to the CNBC API. <br>
The `APIWrapper` class is used to make requests to the CNBC API. <br>
Note: A majority of the CNBC API endpoints require parameters. These must be set by an additional instruction.

```python
Expand All @@ -36,12 +36,16 @@ api_wrapper = APIWrapper(
# The APIWrapper class will supply the required parameters for the configured CNBC API endpoint.
api_wrapper_params = api_wrapper.params
api_wrapper_params['symbol'] = 'AAPL'
api_wrapper.params = api_wrapper_params
# The APIWrapper class will make a request to the CNBC API and return the response in JSON.
json_response = api_wrapper.request()

# The APIWrapper class can be repurposed to make multiple requests to the CNBC API.
api_wrapper.endpoint = Endpoints.GET_SUMMARY
api_wrapper.params = {'issueIds': json_response['issueId']}
api_wrapper_params = api_wrapper.params
api_wrapper_params['issueIds'] = json_response['issueId']
json_response = api_wrapper.request()
```
```

#### Translate Endpoint
A majority of the CNBC API endpoints require an `issueId` or `issueIds` parameter. The translate endpoint is used to convert a symbol to an issueId. <br>
Note: The `APIWrapper` class contains a translation table which can be loaded and saved to a file to reduce the number of requests to the CNBC API.
52 changes: 51 additions & 1 deletion src/cnbc/api_wrapper.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""
A module to make API requests.
"""
import json
import requests

from .endpoints import Endpoints
Expand All @@ -25,10 +26,14 @@ def __init__(self, api_key: str, endpoint: Endpoints, timeout: int = 10):
self._headers: dict[str, str]
self._timeout: int

self._translate_table: dict[str, str]

self._endpoint, self._params = endpoint.value
self._headers = {'x-rapidapi-host': Endpoints.HOST.value, 'x-rapidapi-key': api_key}
self._timeout = timeout

self._translation_table = {}

def _safe_delete(self):
"""
Safely delete the attributes.
Expand Down Expand Up @@ -63,6 +68,7 @@ def params(self):
@params.setter
def params(self, params: dict[str, str]):
try:
# Update the parameters if the input parameters match the endpoint parameters.
if set(self._params.keys()) == set(params.keys()):
self._params.update(params)
else:
Expand Down Expand Up @@ -106,16 +112,60 @@ def timeout(self, timeout: int):
def timeout(self):
del self._timeout

@property
def translation_table(self):
"""
Get the translation table of the API request.
:return: The translation table of the API request.
"""
return self._translation_table

@translation_table.setter
def translation_table(self, translation_table: dict):
self._translation_table = translation_table

@translation_table.deleter
def translation_table(self):
del self._translation_table

def translation_table_load(self, file_path: str):
"""
Load the translation table from a file.
:param file_path: The file path of the translation table.
:return: None
"""
with open(file_path, "r") as file:
self._translation_table = json.load(file)

def translation_table_save(self, file_path: str):
"""
Save the translation table to a file.
:param file_path: The file path of the translation table.
:return: None
"""
with open(file_path, "w") as file:
json.dump(self._translation_table, file)

def request(self) -> dict:
"""
Make an API request.
:return: API response in JSON.
"""
# If the endpoint is the translate endpoint, then check if the symbol is in the translation table.
if self._endpoint == Endpoints.TRANSLATE.get_endpoint():
# If the symbol is in the translation table, then return a faux JSON response.
if issueId := self._translation_table.get(self._params['symbol']):
return {'issueId': issueId, 'errorMessage': '', 'errorCode': ''}

with requests.request("GET", self._endpoint,
headers=self._headers, params=self._params, timeout=self._timeout) as response:
try:
response.raise_for_status()
return response.json()
response_json = response.json()
# If the endpoint is the translate endpoint, then update the translation table.
if self._endpoint == Endpoints.TRANSLATE.get_endpoint():
self._translation_table[self._params['symbol']] = response_json['issueId']
return response_json
except requests.exceptions.HTTPError as e:
raise APIRequestException(response.status_code, response.text) from e
except requests.exceptions.RequestException as e:
Expand Down
44 changes: 44 additions & 0 deletions tests/cnbc/test_api_wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,3 +54,47 @@ def test_request(self, mock_request: MagicMock):
headers=endpoint.headers, params=endpoint.params, timeout=endpoint.timeout
)
self.assertEqual({"key": "value"}, response)

def test_request_translate_translation_table(self):
"""
Test the request method with the translate endpoint with a translation table.
:return: None
"""
translation_table = {'AAPL': '123'}
json_response_expected = {'issueId': '123', 'errorMessage': '', 'errorCode': ''}

api_wrapper = APIWrapper('API_KEY', Endpoints.TRANSLATE)
api_wrapper._translation_table = translation_table
api_wrapper_params = api_wrapper.params
api_wrapper_params['symbol'] = 'AAPL'
json_response = api_wrapper.request()

self.assertEqual(translation_table, api_wrapper.translation_table)
self.assertEqual(json_response_expected, json_response)

@patch('src.cnbc.api_wrapper.requests.request')
def test_request_translate_translation_table_empty(self, mock_request: MagicMock):
"""
Test the request method with the translate endpoint without a translation table.
:param mock_request: The mock request.
:return: None
"""
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {'issueId': '123', 'errorMessage': '', 'errorCode': ''}

mock_request.return_value.__enter__.return_value = mock_response

translation_table = {'AAPL': '123'}

api_wrapper = APIWrapper('API_KEY', Endpoints.TRANSLATE)
api_wrapper_params = api_wrapper.params
api_wrapper_params['symbol'] = 'AAPL'
json_response = api_wrapper.request()

mock_request.assert_called_once_with(
"GET", api_wrapper.endpoint,
headers=api_wrapper.headers, params=api_wrapper.params, timeout=api_wrapper.timeout
)
self.assertEqual(translation_table, api_wrapper.translation_table)
self.assertEqual(mock_response.json.return_value, json_response)
Loading