I’ve figured it out on my own. I am not sure if this is the best way to deal with this, but it works and might help someone in the future. The approach is based on Spotify’s native integration. This checks are not done on setup(), but I decided to do it on the sensor itself. However, the logic could be adapted.
When a Sensor is created, I also create an API class. This API has access_token and refresh_token as attributes:
def __init__(self, sensor, client_id, client_secret):
"""Instantiates a new OuraApi class.
Args:
sensor: Oura sensor to which this api is linked.
client_id: Client id for Oura API.
client_secret: Client secret for Oura API.
"""
self._sensor = sensor
self._client_id = client_id
self._client_secret = client_secret
self._access_token = None
self._refresh_token = None
If when fetching the data these parameters are not set up, we try fetching them from a local storage file:
def get_sleep_data(self, start_date, end_date=None):
"""Fetches data for a sleep OuraEndpoint and date.
Args:
start_date: Day for which to fetch data(YYYY-MM-DD).
end_date: Last day for which to retrieve data(YYYY-MM-DD).
If same as start_date, leave empty.
Returns:
Dictionary containing Oura sleep data.
None if the access token was not found or authorized.
"""
if not self._access_token:
self._get_access_token_data_from_file()
# If after fetching the token, it is still not available, the update should
# not go through. This is most likely at the OAuth set up stage and may
# require input from the user.
if not self._access_token:
return None
The file could be in 3 status:
- The OAuth has not happened for this sensor. Hence, the OAuth needs to start.
- The file contains “code”. This means that this is on the second step of authentication and the code needs to get exchanged for a token.
- The file contains “access_code”, in which case this was already authenticated and we can just use the credentials normally.
def _get_access_token_data_from_file(self):
"""Gets credentials data from the credentials file."""
if not os.path.isfile(self.token_file_name):
self._get_authentication_code()
return
with open(self.token_file_name, 'r') as token_file:
token_data = json.loads(token_file.read()) or {}
if token_data.get('code'):
self._get_access_token_from_code(token_data.get('code'))
return
if token_data.get('access_token') and token_data.get('refresh_token'):
self._access_token = token_data.get('access_token')
self._refresh_token = token_data.get('refresh_token')
logging.error('Unable to retrieve access token from file data.')
If the file does not exit, we trigger the OAuth flow to get the access code:
def _get_authentication_code(self):
"""Gets authentication code."""
base_url = self._sensor._hass.config.api.base_url
callback_url = f'{base_url}{views.AUTH_CALLBACK_PATH}'
state = self._sensor.name
authorize_params = {
'client_id': self._client_id,
'duration': 'temporary',
'redirect_uri': callback_url,
'response_type': 'code',
'scope': 'email personal daily',
'state': state,
}
authorize_url = '{}?{}'.format(
self._get_api_endpoint(OuraEndpoints.AUTHORIZE),
urllib.parse.urlencode(authorize_params))
self._sensor.create_oauth_view(authorize_url)
This will do three things:
- Create a permanent notification with the link to authorize.
- This link includes a callback URL which is the URL that will receive the code after approving the access.
- It creates a view to handle the callback.
def create_oauth_view(self, authorize_url):
"""Creates a view and message to obtain authorization token.
Args:
authorize_url: Authorization URL.
"""
self._hass.http.register_view(views.OuraAuthCallbackView(self))
self._hass.components.persistent_notification.create(
'In order to authorize Home-Assistant to view your Oura Ring data, '
'you must visit: '
f'<a href="{authorize_url}" target="_blank">{authorize_url}</a>',
title=SENSOR_NAME,
notification_id=f'oura_setup_{self._name}')
On the second call, we are creating a notification for the user. When the user clicks the “authorize_url”, it gets redirected to Oura website and comes back to the callback URL.
This call is handled by my OuraAuthCallbackView. This is a view that is set to handle the call exactly on the callback URL.
This view is responsible for receiving the code and storing it on the cache file for this sensor. As we are using sensor name as part of the file name, we could have different access tokens for different sensors (aka: different rings / accounts).
Note: This view part is what I really was struggling with. I was unsure how to handle the callback URL to continue the OAuth flow within the Home-Assistant ecosystem.
# Views configuration.
AUTH_CALLBACK_NAME = 'api:oura'
AUTH_CALLBACK_PATH = '/oura/oauth/setup'
class OuraAuthCallbackView(http.HomeAssistantView):
"""Oura Authorization Callback View.
Methods:
get: Handles get requests to given view.
"""
requires_auth = False
url = AUTH_CALLBACK_PATH
name = AUTH_CALLBACK_NAME
def __init__(self, sensor):
"""Initializes view.
Args:
sensor: Sensor which initialized the OAuth process.
"""
self._sensor = sensor
@core.callback
def get(self, request):
"""Handles Oura OAuth callbacks.
Stores code from Oura API into cache token file.
This code will be read by the API and use it to retrieve access token.
"""
code = request.query.get('code')
code_data = {'code': code}
sensor_name = request.query.get('state')
token_file_name = self._sensor._api.token_file_name
with open(token_file_name, 'w+') as token_file:
token_file.write(json.dumps(code_data))
self._sensor.update() # Forces exchanging code for access token.
return self.json_message(
f'Oura OAuth code {code} for sensor.{sensor_name} stored in '
f'{token_file_name}. The sensor API will use this code to retrieve '
'the access token, store it and start fetching data on the next '
'update. No further action is required from your side. Any errors on '
'retrieving the API token will be logged. If you ever want to restart '
f'this OAuth process, simply delete the file {token_file_name} within '
'the /config/ directory.')
After storing the code in a file, the view updates the sensor. This will trigger again the OAuth process. However, this time, we have a token cache file with “code” on it. As a result, the code will call _get_access_token_from_code to exchange the code for an access token.
def _get_access_token_from_code(self, code):
"""Requests and stores an access token for the given code.
Args:
code: Oura OAuth code.
"""
access_token_data = self._request_access_token_with_code(code)
self._store_access_token_data(access_token_data)
def _request_access_token_with_code(self, code):
"""Sends a request to get access token with new OAuth code.
Args:
code: Oura code to fetch access token.
Returns:
Response from requesting new token.
Most likely: Access and refresh token data in a dictionary.
"""
request_auth = requests.auth.HTTPBasicAuth(self._client_id,
self._client_secret)
base_url = self._sensor._hass.config.api.base_url
callback_url = f'{base_url}{views.AUTH_CALLBACK_PATH}'
request_data = {
'grant_type': 'authorization_code',
'code': code,
'redirect_uri': callback_url,
}
request_headers = {
'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8',
}
request_url = self._get_api_endpoint(OuraEndpoints.TOKEN)
response = requests.post(
request_url, auth=request_auth, data=request_data,
headers=request_headers)
return response.json()
Once this call is made, we should have received the access token and refresh tokens. The next step is to store them and update the API class properties.
def _store_access_token_data(self, access_token_data):
"""Validates and stores access token data into file.
Args:
access_token_data: Dictionary containing access token and refresh token.
"""
if 'access_token' not in access_token_data:
logging.error('Oura API was unable to retrieve new API token.')
return
if 'refresh_token' not in access_token_data:
if self._refresh_token:
access_token_data['refresh_token'] = self._refresh_token
else:
logging.error(
'Refresh token not available. Oura API will become unauthorized.')
return
self._access_token = access_token_data['access_token']
self._refresh_token = access_token_data['refresh_token']
At this stage, the OAuth is completed, the update sensor will continue and will start showing data from the API.
At a later update, unrelated to this one, this access token will be expired. In that case, we need to use the refresh_token and exchange it for a new access token.
The expiration can be validated during the call stages when received an authorized request.
def get_sleep_data(self, start_date, end_date=None):
"""Fetches data for a sleep OuraEndpoint and date.
Args:
start_date: Day for which to fetch data(YYYY-MM-DD).
end_date: Last day for which to retrieve data(YYYY-MM-DD).
If same as start_date, leave empty.
Returns:
Dictionary containing Oura sleep data.
None if the access token was not found or authorized.
"""
if not self._access_token:
self._get_access_token_data_from_file()
# If after fetching the token, it is still not available, the update should
# not go through. This is most likely at the OAuth set up stage and may
# require input from the user.
if not self._access_token:
return None
retries = 0
while retries < _MAX_API_RETRIES:
api_url = self._get_api_endpoint(OuraEndpoints.SLEEP,
start_date=start_date)
response = requests.get(api_url)
response_data = response.json()
if not response_data:
retries += 1
continue
if response_data.get('message') == 'Unauthorized':
retries += 1
self._refresh_access_token()
continue
return response_data
return None
If the response fails with “Unauthorized”, it most likely means that the token expired. (Do note that this might be different on your API). If that’s the case, we trigger the process to refresh the access token.
This basically consists of exchanging the refresh token for a new access token.
def _refresh_access_token(self):
"""Gets a new access token using refresh token."""
access_token_data = self._request_access_token_with_refresh_token()
self._store_access_token_data(access_token_data)
def _request_access_token_with_refresh_token(self):
"""Sends a request to get access token with the existing refresh token.
Returns:
Response from requesting new token.
Most likely: Access and refresh token data in a dictionary.
"""
request_auth = requests.auth.HTTPBasicAuth(self._client_id,
self._client_secret)
request_data = {
'grant_type': 'refresh_token',
'refresh_token': self._refresh_token,
}
request_url = self._get_api_endpoint(OuraEndpoints.TOKEN)
response = requests.post(request_url, auth=request_auth, data=request_data)
return response.json()
Once the new token is retrieve, we call _store_access_token_data to store to the temp file.