Toot-Mastodon-CLI-TUI-clien.../toot/http.py

88 lines
2.3 KiB
Python
Raw Normal View History

2017-12-30 14:15:51 +01:00
from requests import Request, Session
from toot.exceptions import NotFoundError, ApiError
from toot.logging import log_request, log_response
2017-12-30 14:15:51 +01:00
def send_request(request, allow_redirects=True):
log_request(request)
with Session() as session:
prepared = session.prepare_request(request)
settings = session.merge_environment_settings(prepared.url, {}, None, None, None)
response = session.send(prepared, allow_redirects=allow_redirects, **settings)
2017-12-30 14:15:51 +01:00
log_response(response)
return response
2017-12-30 14:15:51 +01:00
def _get_error_message(response):
"""Attempt to extract an error message from response body"""
try:
data = response.json()
if "error_description" in data:
return data['error_description']
if "error" in data:
return data['error']
except Exception:
pass
return "Unknown error"
def process_response(response):
if not response.ok:
error = _get_error_message(response)
2017-12-30 14:15:51 +01:00
if response.status_code == 404:
raise NotFoundError(error)
raise ApiError(error)
return response
def get(app, user, url, params=None):
url = app.base_url + url
headers = {"Authorization": "Bearer " + user.access_token}
request = Request('GET', url, headers, params=params)
response = send_request(request)
2017-12-30 14:15:51 +01:00
return process_response(response)
def anon_get(url, params=None):
request = Request('GET', url, None, params=params)
response = send_request(request)
2017-12-30 14:15:51 +01:00
return process_response(response)
def post(app, user, url, data=None, files=None, allow_redirects=True, headers={}):
2017-12-30 14:15:51 +01:00
url = app.base_url + url
headers["Authorization"] = "Bearer " + user.access_token
2017-12-30 14:15:51 +01:00
request = Request('POST', url, headers, files, data)
response = send_request(request, allow_redirects)
return process_response(response)
2017-12-30 14:15:51 +01:00
2018-06-14 10:40:16 +02:00
def delete(app, user, url, data=None):
url = app.base_url + url
headers = {"Authorization": "Bearer " + user.access_token}
request = Request('DELETE', url, headers=headers, data=data)
response = send_request(request)
return process_response(response)
def anon_post(url, data=None, files=None, allow_redirects=True):
request = Request('POST', url, {}, files, data)
response = send_request(request, allow_redirects)
2017-12-30 14:15:51 +01:00
return process_response(response)