Extract http methods - toot - Unnamed repository; edit this file 'description' to name the repository.
(DIR) Log
(DIR) Files
(DIR) Refs
(DIR) LICENSE
---
(DIR) commit 20eaf86b56fc3c28b66c7043cb60b238e38f8253
(DIR) parent 7bbc98363edfb64ca7e07bbe1dab7b46ab1210bd
(HTM) Author: Ivan Habunek <ivan@habunek.com>
Date: Sat, 30 Dec 2017 14:15:51 +0100
Extract http methods
Diffstat:
toot/api.py | 89 ++++++-------------------------
toot/http.py | 62 +++++++++++++++++++++++++++++++
2 files changed, 77 insertions(+), 74 deletions(-)
---
(DIR) diff --git a/toot/api.py b/toot/api.py
@@ -3,78 +3,19 @@
import re
import requests
-from requests import Request, Session
from urllib.parse import urlparse, urlencode
-from toot import CLIENT_NAME, CLIENT_WEBSITE
-from toot.utils import domain_exists
-from toot.logging import log_request, log_response
+from toot import http, CLIENT_NAME, CLIENT_WEBSITE
from toot.exceptions import ApiError, AuthenticationError, NotFoundError
+from toot.utils import domain_exists
SCOPES = 'read write follow'
-def _process_response(response):
- log_response(response)
-
- if not response.ok:
- error = "Unknown error"
-
- try:
- data = response.json()
- if "error_description" in data:
- error = data['error_description']
- elif "error" in data:
- error = data['error']
- except Exception:
- pass
-
- 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}
-
- log_request(Request('GET', url, headers, params=params))
-
- response = requests.get(url, params, headers=headers)
-
- return _process_response(response)
-
-
-def _unauthorized_get(url, params=None):
- log_request(Request('GET', url, None, params=params))
-
- response = requests.get(url, params)
-
- return _process_response(response)
-
-
-def _post(app, user, url, data=None, files=None):
- url = app.base_url + url
- headers = {"Authorization": "Bearer " + user.access_token}
-
- session = Session()
- request = Request('POST', url, headers, files, data)
- prepared_request = request.prepare()
-
- log_request(request)
-
- response = session.send(prepared_request)
-
- return _process_response(response)
-
-
def _account_action(app, user, account, action):
url = '/api/v1/accounts/{}/{}'.format(account, action)
- return _post(app, user, url).json()
+ return http.post(app, user, url).json()
def create_app(instance):
@@ -88,7 +29,7 @@ def create_app(instance):
'website': CLIENT_WEBSITE,
})
- return _process_response(response).json()
+ return http.process_response(response).json()
def login(app, username, password):
@@ -107,7 +48,7 @@ def login(app, username, password):
if response.is_redirect:
raise AuthenticationError()
- return _process_response(response).json()
+ return http.process_response(response).json()
def get_browser_login_url(app):
@@ -131,11 +72,11 @@ def request_access_token(app, authorization_code):
'redirect_uri': 'urn:ietf:wg:oauth:2.0:oob',
}, allow_redirects=False)
- return _process_response(response).json()
+ return http.process_response(response).json()
def post_status(app, user, status, visibility='public', media_ids=None):
- return _post(app, user, '/api/v1/statuses', {
+ return http.post(app, user, '/api/v1/statuses', {
'status': status,
'media_ids[]': media_ids,
'visibility': visibility,
@@ -143,7 +84,7 @@ def post_status(app, user, status, visibility='public', media_ids=None):
def timeline_home(app, user):
- return _get(app, user, '/api/v1/timelines/home').json()
+ return http.get(app, user, '/api/v1/timelines/home').json()
def _get_next_path(headers):
@@ -158,26 +99,26 @@ def timeline_generator(app, user):
next_path = '/api/v1/timelines/home'
while next_path:
- response = _get(app, user, next_path)
+ response = http.get(app, user, next_path)
yield response.json()
next_path = _get_next_path(response.headers)
def upload_media(app, user, file):
- return _post(app, user, '/api/v1/media', files={
+ return http.post(app, user, '/api/v1/media', files={
'file': file
}).json()
def search(app, user, query, resolve):
- return _get(app, user, '/api/v1/search', {
+ return http.get(app, user, '/api/v1/search', {
'q': query,
'resolve': resolve,
}).json()
def search_accounts(app, user, query):
- return _get(app, user, '/api/v1/accounts/search', {
+ return http.get(app, user, '/api/v1/accounts/search', {
'q': query,
}).json()
@@ -207,11 +148,11 @@ def unblock(app, user, account):
def verify_credentials(app, user):
- return _get(app, user, '/api/v1/accounts/verify_credentials').json()
+ return http.get(app, user, '/api/v1/accounts/verify_credentials').json()
def get_notifications(app, user):
- return _get(app, user, '/api/v1/notifications').json()
+ return http.get(app, user, '/api/v1/notifications').json()
def get_instance(app, user, domain):
@@ -221,7 +162,7 @@ def get_instance(app, user, domain):
url = "http://{}/api/v1/instance".format(domain)
try:
- return _unauthorized_get(url).json()
+ return http.unauthorized_get(url).json()
except NotFoundError:
raise ApiError(
"Instance info not found at {}.\n"
(DIR) diff --git a/toot/http.py b/toot/http.py
@@ -0,0 +1,62 @@
+import requests
+
+from toot.logging import log_request, log_response
+from requests import Request, Session
+from toot.exceptions import NotFoundError, ApiError
+
+
+def process_response(response):
+ log_response(response)
+
+ if not response.ok:
+ error = "Unknown error"
+
+ try:
+ data = response.json()
+ if "error_description" in data:
+ error = data['error_description']
+ elif "error" in data:
+ error = data['error']
+ except Exception:
+ pass
+
+ 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}
+
+ log_request(Request('GET', url, headers, params=params))
+
+ response = requests.get(url, params, headers=headers)
+
+ return process_response(response)
+
+
+def unauthorized_get(url, params=None):
+ log_request(Request('GET', url, None, params=params))
+
+ response = requests.get(url, params)
+
+ return process_response(response)
+
+
+def post(app, user, url, data=None, files=None):
+ url = app.base_url + url
+ headers = {"Authorization": "Bearer " + user.access_token}
+
+ session = Session()
+ request = Request('POST', url, headers, files, data)
+ prepared_request = request.prepare()
+
+ log_request(request)
+
+ response = session.send(prepared_request)
+
+ return process_response(response)