158 lines
5.7 KiB
Python
158 lines
5.7 KiB
Python
import requests
|
|
|
|
# Wrapper base para realizar requests (GET/POST/PUT/DELETE) à API da PlugNotas de forma padronizada
|
|
class Api:
|
|
def __init__(self, api_key):
|
|
self.api_key = api_key
|
|
|
|
def get(self, url):
|
|
try:
|
|
headers = {
|
|
"x-api-key": self.api_key
|
|
}
|
|
response = requests.get(url, headers=headers, timeout=30)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
resp_data = {}
|
|
if 'response' in locals() and response is not None:
|
|
try:
|
|
resp_data = response.json()
|
|
except ValueError:
|
|
resp_data = response.text
|
|
return {"error": str(e), "response": resp_data}
|
|
|
|
def get_paginated(self, url, max_pages=100):
|
|
try:
|
|
headers = {
|
|
"x-api-key": self.api_key
|
|
}
|
|
all_data = []
|
|
current_url = url
|
|
page = 0
|
|
|
|
while current_url and page < max_pages:
|
|
separator = "&" if "?" in current_url else "?"
|
|
request_url = current_url if page == 0 else current_url
|
|
|
|
response = requests.get(request_url, headers=headers, timeout=30)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
|
|
if isinstance(result, list):
|
|
all_data.extend(result)
|
|
break
|
|
elif isinstance(result, dict):
|
|
page_data = result.get("data", [])
|
|
if isinstance(page_data, list):
|
|
all_data.extend(page_data)
|
|
elif page_data:
|
|
all_data.append(page_data)
|
|
|
|
hash_proxima = result.get("hashProximaPagina")
|
|
if hash_proxima:
|
|
separator = "&" if "?" in url else "?"
|
|
current_url = f"{url}{separator}hashProximaPagina={hash_proxima}"
|
|
page += 1
|
|
else:
|
|
break
|
|
else:
|
|
break
|
|
|
|
return {"data": all_data, "total_paginas": page + 1, "total_registros": len(all_data)}
|
|
except requests.exceptions.RequestException as e:
|
|
resp_data = {}
|
|
if 'response' in locals() and response is not None:
|
|
try:
|
|
resp_data = response.json()
|
|
except ValueError:
|
|
resp_data = response.text
|
|
return {"error": str(e), "response": resp_data}
|
|
|
|
def post(self, url, data):
|
|
try:
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"x-api-key": self.api_key
|
|
}
|
|
response = requests.post(url, headers=headers, json=data, timeout=30)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
resp_data = {}
|
|
if 'response' in locals() and response is not None:
|
|
try:
|
|
resp_data = response.json()
|
|
except ValueError:
|
|
resp_data = response.text
|
|
return {"error": str(e), "response": resp_data}
|
|
|
|
def put(self, url, data):
|
|
try:
|
|
headers = {
|
|
"x-api-key": self.api_key
|
|
}
|
|
response = requests.put(url, headers=headers, json=data, timeout=30)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": str(e)}
|
|
|
|
def patch(self, url, data):
|
|
try:
|
|
headers = {
|
|
"x-api-key": self.api_key
|
|
}
|
|
response = requests.patch(url, headers=headers, json=data, timeout=30)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": str(e)}
|
|
|
|
def delete(self, url):
|
|
try:
|
|
headers = {
|
|
"x-api-key": self.api_key
|
|
}
|
|
response = requests.delete(url, headers=headers, timeout=30)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": str(e)}
|
|
|
|
def post_file(self, url, file):
|
|
try:
|
|
headers = {
|
|
"x-api-key": self.api_key
|
|
}
|
|
response = requests.post(url, headers=headers, files=file, timeout=60)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": str(e)}
|
|
|
|
def post_form_data(self, url, data):
|
|
try:
|
|
headers = {
|
|
"x-api-key": self.api_key
|
|
}
|
|
response = requests.post(url, headers=headers, files=data, timeout=60)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": str(e)}
|
|
|
|
def put_form_data(self, url, data):
|
|
try:
|
|
headers = {
|
|
"x-api-key": self.api_key
|
|
}
|
|
response = requests.put(url, headers=headers, files=data, timeout=60)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": str(e)}
|
|
|
|
# api = Api("2da392a6-79d2-4304-a8b7-959572c7e44d")
|
|
# response = api.get("https://api.sandbox.plugnotas.com.br/certificado")
|
|
# print(response)
|