Skip to content
Extraits de code Groupes Projets
Valider 65326682 rédigé par Juliana's avatar Juliana
Parcourir les fichiers

[FIX] Correction call api

parent d2ac9af8
Aucune branche associée trouvée
Aucune étiquette associée trouvée
Aucune requête de fusion associée trouvée
......@@ -4,6 +4,7 @@
import requests
import logging
import pytz
import json
from datetime import datetime, date
......@@ -71,7 +72,7 @@ class ApiEnedis(models.AbstractModel):
print("--- token ---", str(token))
return token
def enedis_get_by_url(self, url, call_type, query=None):
def enedis_get_by_url(self, url, call_type, query=None, token=None):
""" Création de la requête à Enedis
:param url: action API Enedis
......@@ -81,10 +82,10 @@ class ApiEnedis(models.AbstractModel):
@return response au format JSON
"""
_logger.info("Calling %s" % url)
if not token:
token = self.access_token()
url_enedis = self.env.user.company_id.url_enedis
# url_enedis = 'https://gw.hml.api.enedis.fr/'
header = {
'Content-Type': 'application/json;charset=UTF-8',
......@@ -111,17 +112,12 @@ class ApiEnedis(models.AbstractModel):
)
response = False
try:
response.raise_for_status()
except Exception as error:
raise ValueError('Autentication failed: {}'.format(error))
# Gestion erreur API
if response.status_code not in [200]:
try:
message = response.json().get('content')
message = response.json().get('error_description')
except:
message = response.content
message = response.error_description
raise exceptions.Warning(
"L'appel url '%s' a échoué\n"
"- Code erreur : %d\n"
......@@ -131,7 +127,7 @@ class ApiEnedis(models.AbstractModel):
message))
return response
def definitive_load_curves(self, date_start, date_end, type, usage_point_name=None):
def definitive_load_curves(self, date_start, date_end, type, usage_point_name=None, token=None):
""" Fonction permettant d'appeler l'API Enedis et retourne les courbes
de chare en fonction d'un intervalle de date et d'un type
:param date_start: une date de début
......@@ -155,11 +151,11 @@ class ApiEnedis(models.AbstractModel):
('enercoop_operation_id', '=', self.id)
])
if not log_id:
self.load_data(url, date_start, date_end, type, usage_point_name)
self.load_data(url, date_start, date_end, type, usage_point_name, token)
return True
def load_data(self, url, date_start, date_end, type_courbe=None, usage_point_name=None):
def load_data(self, url, date_start, date_end, type_courbe=None, usage_point_name=None, token=None):
""" Read function
:param url : URL à appeler
date_start: une date de début
......@@ -184,7 +180,8 @@ class ApiEnedis(models.AbstractModel):
response = self.enedis_get_by_url(
url=url,
call_type='get',
query=query)
query=query,
token=token)
# get the data
curves = response.json().get('curves')
......@@ -247,7 +244,8 @@ class ApiEnedis(models.AbstractModel):
response = self.enedis_get_by_url(
url=url,
call_type='get',
query=[])
query=[],
token=None)
usage_points = response.json().get('usage_points')
......
......@@ -57,19 +57,19 @@ class EnercoopOperation(models.Model):
# Actions
# ------------------------------------------------------
def get_curves(self, date_start, date_end):
# Ask token to API
token = self.access_token()
# Load consommation data by PRM
for delivery_counter_id in self.enercoop_delivery_operation_ids:
self.definitive_load_curves(date_start, date_end, 'cons', delivery_counter_id.enercoop_counter_id.name)
self.definitive_load_curves(date_start, date_end, 'autocons', delivery_counter_id.enercoop_counter_id.name)
self.definitive_load_curves(date_start, date_end, 'cons', delivery_counter_id.enercoop_counter_id.name, token=token)
self.definitive_load_curves(date_start, date_end, 'autocons', delivery_counter_id.enercoop_counter_id.name, token=token)
# Load production data by PRM
for injection_counter_id in self.enercoop_injection_operation_ids:
self.definitive_load_curves(date_start, date_end, 'prod', injection_counter_id.enercoop_counter_id.name)
self.definitive_load_curves(date_start, date_end, 'prod', injection_counter_id.enercoop_counter_id.name, token=token)
# Load surplus data by operation
self.definitive_load_curves(date_start, date_end, 'surplus', None)
# self.definitive_load_curves(date_start, date_end)
self.definitive_load_curves(date_start, date_end, 'surplus', None, token=token)
def get_perimeter(self):
for operation in self:
......
0% Chargement en cours ou .
You are about to add 0 people to the discussion. Proceed with caution.
Veuillez vous inscrire ou vous pour commenter