Блог

Я пытаюсь создать класс для управления данным API. Пока это работает, даже если мне нужно добавить кое-что, чтобы правильно обрабатывать ошибки и анализировать выходные данные. Пожалуйста, смотрите Ниже код до сих пор:

 class ApiRteClient:

urlAuth = "https://digital.iservices.rte-france.com/token/oauth"
API_URL = "https://digital.iservices.rte-france.com/open_api"

def __init__(self,privateKey):
    self.privateKey = privateKey
    if os.path.exists('token.pickle'):
        with open('token.pickle', 'rb') as token:
            credentials = pickle.load(token)
            self._expirationDate = credentials["expirationDate"]
            if datetime.now() > self._expirationDate:
                self._getAccessToken()
            else:
                self._accessToken = credentials["accessToken"]
                self.session = self._init_session()
    else:
        self._getAccessToken()

def _init_session(self):

    session = requests.session()
    session.headers.update({'Authorization': 'Bearer '   self._accessToken})

    return session

def _getAccessToken(self):

    headers = {}
    headers.update({"Authorization": "Basic "   self.privateKey})
    headers.update({"Content-Type": 'application/x-www-form-urlencoded'})
    try:
        response = requests.post(self.urlAuth, headers=headers)
    except:
        response.raise_for_status()

    self._accessToken = json.loads(response.content.decode())["access_token"]
    self._expirationDate = datetime.now()   timedelta(0,json.loads(response.content.decode())["expires_in"])
    tokenInfo = {}
    tokenInfo.update({"accessToken": self._accessToken})
    self.session = self._init_session()
    tokenInfo.update({"expirationDate": self._expirationDate})
    with open('token.pickle', 'wb') as token:
        pickle.dump(tokenInfo, token)

def _request(self,method,uri,**kwargs):

    kwargs['timeout'] = 10

    data = kwargs.get('data', None)
    if data and isinstance(data, dict):
        kwargs['data'] = data



    self.response = getattr(self.session, method)(uri, **kwargs)

    return self._handle_response()

def _handle_response(self):
    return self.response.json()

def _request_api(self,method,path,**kwargs):

    uri = self._create_actual_generation_api_uri(path)

    return self._request(method,uri,**kwargs)

def _get(self, path, **kwargs):

    return self._request_api('get', path, **kwargs)

def _create_actual_generation_api_uri(self,path):

    return self.API_URL   "/actual_generation/"   "v1"   "/"   path

def get_actual_generation_per_production_type(self,**kwargs):

    return self._get("actual_generations_per_production_type", **kwargs)

def get_actual_generation_per_production_type(self,**kwargs):

    return self._get("actual_generations_per_unit", **kwargs)

def get_actual_water_reserves(self,**kwargs):

    return self._get("water_reserves", **kwargs)

def get_actual_generationmix_15min_step(self,**kwargs):

    return self._get("generation_mix_15min_time_scale", **kwargs)
  

В параметре запросов у меня всегда будет указываться дата начала и окончания, и API имеет некоторые ограничения в отношении диапазона дней: 7 дней, 1 years….it зависит от метода.

Я хочу включить что-то, что позволит пользователю не управлять этим. Например, для запроса в течение одного месяца, который ограничен 7 днями, разделите его на разные 7-дневные запросы.

Я хотел бы использовать декоратор с параметром, но есть ли другие варианты?

С уважением