hueso/Alpaca
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Simplified connection_handler

Browse Source
This commit is contained in:
jeffser
2024-07-07 20:41:25 -06:00
parent 35869c0fbf
commit 6cd813c146
3 changed files with 35 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
src/connection_handler.py
View File

@@ -1,6 +1,6 @@
# connectionhandler.py
# connection_handler.py
import json, requests
#OK=200 response.status_code
url = None
bearer_token = None
@@ -13,44 +13,13 @@ def get_headers(include_json:bool) -> dict:
return headers if len(headers.keys()) > 0 else None
def simple_get(connection_url:str) -> dict:
try:
response = requests.get(connection_url, headers=get_headers(False))
if response.status_code == 200:
return {"status": "ok", "text": response.text, "status_code": response.status_code}
else:
return {"status": "error", "status_code": response.status_code}
except Exception as e:
return {"status": "error", "status_code": 0}
return requests.get(connection_url, headers=get_headers(False))
def simple_post(connection_url:str, data) -> dict:
try:
response = requests.post(connection_url, headers=get_headers(True), data=data, stream=False)
if response.status_code == 200:
return {"status": "ok", "text": response.text, "status_code": response.status_code}
else:
return {"status": "error", "status_code": response.status_code}
except Exception as e:
return {"status": "error", "status_code": 0}
return requests.post(connection_url, headers=get_headers(True), data=data, stream=False)
def simple_delete(connection_url:str, data) -> dict:
try:
response = requests.delete(connection_url, headers=get_headers(False), json=data)
if response.status_code == 200:
return {"status": "ok", "status_code": response.status_code}
else:
return {"status": "error", "text": "Failed to delete", "status_code": response.status_code}
except Exception as e:
return {"status": "error", "status_code": 0}
return requests.delete(connection_url, headers=get_headers(False), json=data)
def stream_post(connection_url:str, data, callback:callable) -> dict:
try:
response = requests.post(connection_url, headers=get_headers(True), data=data, stream=True)
if response.status_code == 200:
for line in response.iter_lines():
if line:
callback(json.loads(line.decode("utf-8")))
return {"status": "ok", "status_code": response.status_code}
else:
return {"status": "error", "status_code": response.status_code}
except Exception as e:
return {"status": "error", "status_code": 0}
return requests.post(connection_url, headers=get_headers(True), data=data, stream=True)