-
Notifications
You must be signed in to change notification settings - Fork 10
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* feat: add new AhjoSetting class for storing Ahjo-related settings into the DB * feat: add a class for retrieving and refreshing Ahjo access tokens * feat: add a Django command for refreshing the Ahjo access token * feat: add dummy function as basis for further AHJO features * feat: improve error messages * feat: add instructions to readme * fix: code formatting * fix: conflicting migrations, add timestamps * fix: migration conflict
- Loading branch information
Showing
10 changed files
with
496 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
31 changes: 31 additions & 0 deletions
31
backend/benefit/applications/management/commands/refresh_ahjo_token.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| from django.core.exceptions import ObjectDoesNotExist | ||
| from django.core.management.base import BaseCommand | ||
|
|
||
| from applications.models import AhjoSetting | ||
| from applications.services.ahjo_authentication import AhjoConnector | ||
|
|
||
|
|
||
| class Command(BaseCommand): | ||
| help = "Refresh the Ahjo token using the refresh_token stored in the database" | ||
|
|
||
| def handle(self, *args, **options): | ||
| try: | ||
| ahjo_auth_code = AhjoSetting.objects.get(name="ahjo_code").data | ||
| self.stdout.write(f"Retrieved auth code: {ahjo_auth_code}") | ||
| except ObjectDoesNotExist: | ||
| self.stdout.write( | ||
| "Error: Ahjo auth code not found in database. Please set the 'ahjo_code' setting." | ||
| ) | ||
| return | ||
|
|
||
| ahjo_connector = AhjoConnector() | ||
| if not ahjo_connector.is_configured(): | ||
| self.stdout.write( | ||
| "Error: Ahjo connector is not properly configured. Please check your settings." | ||
| ) | ||
| return | ||
| try: | ||
| token = ahjo_connector.refresh_token() | ||
| except Exception as e: | ||
| self.stdout.write(f"Failed to refresh Ahjo token: {e}") | ||
| self.stdout.write(f"Ahjo token refreshed, token valid until {token.expires_in}") |
26 changes: 26 additions & 0 deletions
26
backend/benefit/applications/migrations/0043_ahjosetting.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
| # Generated by Django 3.2.18 on 2023-10-13 12:38 | ||
|
|
||
| from django.db import migrations, models | ||
|
|
||
|
|
||
| class Migration(migrations.Migration): | ||
|
|
||
| dependencies = [ | ||
| ('applications', '0042_reviewstate_paper'), | ||
| ] | ||
|
|
||
| operations = [ | ||
| migrations.CreateModel( | ||
| name='AhjoSetting', | ||
| fields=[ | ||
| ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), | ||
| ('created_at', models.DateTimeField(auto_now_add=True, verbose_name='time created')), | ||
| ('modified_at', models.DateTimeField(auto_now=True, verbose_name='time modified')), | ||
| ('name', models.CharField(max_length=255, unique=True)), | ||
| ('data', models.JSONField()), | ||
| ], | ||
| options={ | ||
| 'abstract': False, | ||
| }, | ||
| ), | ||
| ] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
149 changes: 149 additions & 0 deletions
149
backend/benefit/applications/services/ahjo_authentication.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,149 @@ | ||
| from dataclasses import dataclass | ||
| from datetime import datetime, timedelta | ||
| from typing import Dict, Union | ||
|
|
||
| import requests | ||
| from django.conf import settings | ||
| from django.core.exceptions import ObjectDoesNotExist | ||
|
|
||
| from applications.models import AhjoSetting | ||
|
|
||
|
|
||
| @dataclass | ||
| class AhjoToken: | ||
| access_token: str = "" | ||
| refresh_token: str = "" | ||
| expires_in: datetime = datetime.now() | ||
|
|
||
|
|
||
| class AhjoConnector: | ||
| def __init__(self, requests_module: requests.Session = requests) -> None: | ||
| self.requests_module: requests = requests_module | ||
| self.token_url: str = settings.AHJO_TOKEN_URL | ||
| self.client_id: str = settings.AHJO_CLIENT_ID | ||
| self.client_secret: str = settings.AHJO_CLIENT_SECRET | ||
| self.redirect_uri: str = settings.AHJO_REDIRECT_URL | ||
|
|
||
| self.grant_type_for_auth_token: str = "authorization_code" | ||
| self.grant_type_for_refresh_token: str = "refresh_token" | ||
| self.headers: Dict[str, str] = { | ||
| "Content-Type": "application/x-www-form-urlencoded", | ||
| } | ||
| self.timout: int = 10 | ||
|
|
||
| def is_configured(self) -> bool: | ||
| """Check if all required config options are set""" | ||
| if ( | ||
| not self.token_url | ||
| or not self.client_id | ||
| or not self.client_secret | ||
| or not self.redirect_uri | ||
| ): | ||
| return False | ||
|
|
||
| return True | ||
|
|
||
| def get_access_token(self, auth_code: str) -> AhjoToken: | ||
| """Get access token from db first, then from Ahjo if not found or expired""" | ||
| token = self.get_token_from_db() | ||
| if token and not self._check_if_token_is_expired(token.expires_in): | ||
| return token | ||
| else: | ||
| return self.get_new_token(auth_code) | ||
|
|
||
| def get_new_token(self, auth_code: str) -> AhjoToken: | ||
| """Retrieve the initial access token from Ahjo API using the auth code, | ||
| this is only used when getting the initial token or when the token has expired. | ||
| """ | ||
| if not auth_code: | ||
| raise Exception("No auth code") | ||
| payload = { | ||
| "client_id": self.client_id, | ||
| "client_secret": self.client_secret, | ||
| "grant_type": self.grant_type_for_auth_token, | ||
| "code": auth_code, | ||
| "redirect_uri": self.redirect_uri, | ||
| } | ||
| return self.do_token_request(payload) | ||
|
|
||
| def refresh_token(self) -> AhjoToken: | ||
| """Refresh access token from Ahjo API using the refresh token of an existing token. | ||
| This should be used by, for example, a cron job to keep the token up to date. | ||
| """ | ||
| token = self.get_token_from_db() | ||
| if not token.refresh_token: | ||
| raise Exception("No refresh token") | ||
|
|
||
| payload = { | ||
| "client_id": self.client_id, | ||
| "client_secret": self.client_secret, | ||
| "grant_type": self.grant_type_for_refresh_token, | ||
| "refresh_token": token.refresh_token, | ||
| } | ||
|
|
||
| return self.do_token_request(payload) | ||
|
|
||
| def do_token_request(self, payload: Dict[str, str]) -> AhjoToken: | ||
| # Make the POST request | ||
| response = self.requests_module.post( | ||
| self.token_url, headers=self.headers, data=payload, timeout=self.timout | ||
| ) | ||
|
|
||
| # Check if the request was successful | ||
| if response.status_code == 200: | ||
| # Extract the access token from the JSON response | ||
| access_token = response.json().get("access_token", "") | ||
| expires_in = response.json().get("expires_in", "") | ||
| refresh_token = response.json().get("refresh_token", "") | ||
| expiry_datetime = self.convert_expires_in_to_datetime(expires_in) | ||
|
|
||
| token = AhjoToken( | ||
| access_token=access_token, | ||
| refresh_token=refresh_token, | ||
| expires_in=expiry_datetime, | ||
| ) | ||
| self.set_or_update_token(token) | ||
| return token | ||
| else: | ||
| raise Exception( | ||
| f"Failed to get or refresh token: {response.status_code} {response.content.decode()}" | ||
| ) | ||
|
|
||
| def get_token_from_db(self) -> Union[AhjoToken, None]: | ||
| """Get token from AhjoSetting table""" | ||
| try: | ||
| token_data = AhjoSetting.objects.get(name="ahjo_access_token").data | ||
| access_token = token_data.get("access_token", "") | ||
| refresh_token = token_data.get("refresh_token", "") | ||
| expires_in = token_data.get("expires_in", "") | ||
| return AhjoToken( | ||
| access_token=access_token, | ||
| refresh_token=refresh_token, | ||
| expires_in=datetime.fromisoformat(expires_in), | ||
| ) | ||
| except ObjectDoesNotExist: | ||
| return None | ||
|
|
||
| def _check_if_token_is_expired(self, expires_in: datetime) -> bool: | ||
| """Check if access token is expired""" | ||
| return expires_in < datetime.now() | ||
|
|
||
| def set_or_update_token( | ||
| self, | ||
| token: AhjoToken, | ||
| ) -> None: | ||
| """Save or update token data to AhjoSetting table""" | ||
|
|
||
| access_token_data = { | ||
| "access_token": token.access_token, | ||
| "refresh_token": token.refresh_token, | ||
| "expires_in": token.expires_in.isoformat(), | ||
| } | ||
|
|
||
| AhjoSetting.objects.update_or_create( | ||
| name="ahjo_access_token", defaults={"data": access_token_data} | ||
| ) | ||
|
|
||
| def convert_expires_in_to_datetime(self, expires_in: str) -> datetime: | ||
| """Convert expires_in seconds to datetime""" | ||
| return datetime.now() + timedelta(seconds=int(expires_in)) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.