import requests;
import jwt;
import urllib;
import json
from pyarialib.ariaresponse import AriaResponse
from pyarialib.persona import Persona;
from pyarialib.sei import Sei;
[documentos]class Aria:
_tokens = {}
def __set_query_param(self, url: str, key: str, value: str) -> str:
partes_url = url.split('?')
base_url = partes_url[0]
query_string = partes_url[0] if len(partes_url) >= 2 else None
query_params = urllib.parse.parse_qs(query_string)
if key in query_params:
query_params[key] = value
else:
query_params[key] = [value]
encoded_query_params = urllib.parse.urlencode(query_params, doseq=True)
return f"{base_url}?{encoded_query_params}"
def __autentica(self) -> None:
token_antigo = self._tokens.get(self.username)
if token_antigo != None:
self.token = token_antigo
else:
data = {'login': self.username, 'password': self.password}
headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
resultado = requests.post(self.base_url + ("/" if self.base_url[-1] != "/" else "") + "token/", json=data, headers=headers)
resultado_json = resultado.json()
if resultado_json.get("status") == "ok":
self.token = resultado_json.get("token")
self._tokens[self.username] = self.token
else:
if self._tokens.get(self.username) != None:
self._tokens.pop(self.username)
raise Exception("Erro na autenticação.")
def __init__(self, username = None, password = None, base_url="https://apiapex.tesouro.gov.br/aria/", token = None):
if token != None:
self.token = token
self.revalida_token = False
elif username != None and password != None:
self.base_url = base_url
self.username = username
self.password = password
self.__autentica()
self.revalida_token = True
else:
self.token = None
self.revalida_token = False
def __fetch(self, method, body, json_body, headers, url, data_append = None, auto_pagination = False) -> AriaResponse:
if len(body) > 0:
resultado = requests.request(method=method, url=url, headers=headers, data=body)
else:
resultado = requests.request(method=method, url=url, headers=headers, json=json_body)
dados_json = resultado.json()
if not auto_pagination:
return AriaResponse(resultado.text, resultado.status_code)
if dados_json.get("page") != None and dados_json.get("pageSize") != None and dados_json.get("registros") != None:
if len(dados_json.get('registros', [])) < dados_json.get('pageSize') and data_append != None:
nova_response = AriaResponse(text=data_append, status_code=resultado.status_code)
return nova_response
elif len(dados_json.get('registros', [])) < dados_json.get('pageSize') and data_append == None:
return AriaResponse(resultado.text, resultado.status_code)
elif dados_json.get('page') and len(dados_json.get('registros', [])) == dados_json.get('pageSize'):
next_page_url = self.__set_query_param(url, 'page', str(dados_json.get('page') + 1))
if data_append != None:
dados_append_json = json.loads(data_append)
dados_json['registros'] += dados_append_json.get('registros', [])
return self.__fetch(method=method, body=body, json_body=json_body, headers=headers, url=next_page_url, data_append=json.dumps(dados_json), auto_pagination=auto_pagination)
else:
return AriaResponse(resultado.text, resultado.status_code)
else:
return AriaResponse(resultado.text, resultado.status_code)
[documentos] def request(self, method : str, version : int, project : str, endpoint : str, body : str = "", json_body : dict = {}, headers : dict = {}, query_string_params : dict = {}, auto_pagination : bool = False) -> AriaResponse:
"""
Realiza uma requisição ao ARIA.
Args:
method (string): método da requisição.
version (number): número da versão da api.
project (string): nome do projeto.
endpoint (string): código do endpoint.
body (str, optional): corpo da requisição em formato de string. Defaults to "".
json_body (dict, optional): corpo da requisição em formato de dict, enviado como json. Defaults to {}.
headers (dict, optional): headers da requisição. Defaults to {}.
query_string_params (dict, optional): parâmetros via querystring. Defaults to {}.
Returns:
any: resultado da requisição.
"""
if self.revalida_token:
try:
jwt.decode(self.token, algorithms=["HS512"], options={"verify_signature": False, "verify_exp": True})
except jwt.ExpiredSignatureError:
self.__autentica()
if self.token != None:
headers["Authorization"] = "Bearer " + self.token
query_string = ""
for key in query_string_params:
if query_string == "":
query_string += "?"
else:
query_string += "&"
query_string += key + "=" + str(query_string_params.get(key))
url = self.base_url + ("/" if self.base_url[-1] != "/" else "") + "v" + str(version) + "/" + project + "/custom/" + endpoint + query_string
resultado = self.__fetch(method=method, body=body, json_body=json_body, headers=headers, url=url, auto_pagination=auto_pagination)
return resultado
[documentos] def get_persona(self) -> Persona:
persona = Persona(self)
[documentos] def get_sei(self) -> Sei:
sei = Sei(self)
return Sei(self)