Skip to content

Commit

Permalink
[Fix] Add refresh token in Onisep API. (#20)
Browse files Browse the repository at this point in the history
fix: Implement refresh token for oniseop API
  • Loading branch information
Angel-Dijoux authored Dec 30, 2023
1 parent aae075c commit 01659ec
Show file tree
Hide file tree
Showing 3 changed files with 34 additions and 15 deletions.
17 changes: 13 additions & 4 deletions src/business_logic/formation/__init__.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,20 @@
from enum import Enum
import os

from src.business_logic.formation.scrap.utils.get_onisep_token import get_token
from src.business_logic.formation.scrap.utils.get_onisep_token import (
BearerToken,
get_token,
)


class HeaderKey(Enum):
APPLICATION_ID = "Application-ID"
AUTHORIZATION = "Authorization"


ONISEP_URL = "https://api.opendata.onisep.fr/api/1.0/dataset/"

HEADERS = {
"Application-ID": os.environ.get("ONISEP_APP_ID"),
"Authorization": "Bearer " + get_token(),
HEADERS: dict[HeaderKey, BearerToken | str] = {
HeaderKey.APPLICATION_ID.value: os.environ.get("ONISEP_APP_ID"),
HeaderKey.AUTHORIZATION.value: get_token(),
}
13 changes: 10 additions & 3 deletions src/business_logic/formation/scrap/utils/get_onisep_data.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,10 @@
# Idéo-Formations initiales en France
# https://opendata.onisep.fr/data/5fa591127f501/2-ideo-formations-initiales-en-france.htm
import requests
from src.business_logic.formation import HEADERS, ONISEP_URL
from src.business_logic.formation import HEADERS, ONISEP_URL, HeaderKey
from src.business_logic.formation.exceptions import NoOnisepAPIException
from src.business_logic.formation.scrap.utils.get_onisep_token import get_token
from src.constants.http_status_codes import HTTP_200_OK, HTTP_401_UNAUTHORIZED


DATASET = "5fa591127f501"
Expand All @@ -11,8 +13,13 @@
def get_onisep_data(params: str) -> dict:
url = ONISEP_URL + DATASET + params
response = requests.get(url, headers=HEADERS)
if response.status_code == 200:
if response.status_code == HTTP_200_OK:
return response.json()
if response.status_code == HTTP_401_UNAUTHORIZED:
HEADERS[HeaderKey.AUTHORIZATION.value] = get_token()
response = requests.get(url, headers=HEADERS)
if response.status_code == HTTP_200_OK:
return response.json()
raise NoOnisepAPIException(
f"status: {response.status_code} \n message : Onisep API is down. \n dataset : {DATASET} \n headers : {HEADERS} "
f"\n status: {response.status_code} \n message : Onisep API is down. \n dataset : {DATASET} \n headers : {HEADERS} "
)
19 changes: 11 additions & 8 deletions src/business_logic/formation/scrap/utils/get_onisep_token.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import os
from loguru import logger
from typing import Literal
import requests

from src.business_logic.formation.exceptions import NoOnisepAPIException


URL = "https://api.opendata.onisep.fr/api/1.0/login"

Expand All @@ -13,13 +15,14 @@ def _get_form_data() -> dict[str, str]:
}


def get_token() -> str | None:
BearerToken = Literal["Bearer"]


def get_token() -> BearerToken:
response = requests.post(URL, data=_get_form_data())
if response.status_code == 200:
data = response.json()
return data.get("token")
else:
logger.warning(
f"Failed to get onisep token. Status code: {response.status_code}"
)
return None
return f"Bearer {data.get('token')}"
raise NoOnisepAPIException(
f"Failed to get onisep token. Status code: {response.status_code}"
)

0 comments on commit 01659ec

Please sign in to comment.