Extract api functions to api.py
This commit is contained in:
parent
d7701bd2e6
commit
d53849fe4b
4 changed files with 117 additions and 111 deletions
|
@ -1,7 +1,8 @@
|
||||||
# -*- coding: utf-8 -*-
|
# -*- coding: utf-8 -*-
|
||||||
import requests
|
import requests
|
||||||
|
|
||||||
from toot import App, User, create_app, login, CLIENT_NAME, CLIENT_WEB, SCOPES
|
from toot import App, User, CLIENT_NAME, CLIENT_WEBSITE
|
||||||
|
from toot.api import create_app, login, SCOPES
|
||||||
|
|
||||||
|
|
||||||
class MockResponse:
|
class MockResponse:
|
||||||
|
@ -19,7 +20,7 @@ def test_create_app(monkeypatch):
|
||||||
def mock_post(url, data):
|
def mock_post(url, data):
|
||||||
assert url == 'https://bigfish.software/api/v1/apps'
|
assert url == 'https://bigfish.software/api/v1/apps'
|
||||||
assert data == {
|
assert data == {
|
||||||
'website': CLIENT_WEB,
|
'website': CLIENT_WEBSITE,
|
||||||
'client_name': CLIENT_NAME,
|
'client_name': CLIENT_NAME,
|
||||||
'scopes': SCOPES,
|
'scopes': SCOPES,
|
||||||
'redirect_uris': 'urn:ietf:wg:oauth:2.0:oob'
|
'redirect_uris': 'urn:ietf:wg:oauth:2.0:oob'
|
||||||
|
|
108
toot/__init__.py
108
toot/__init__.py
|
@ -1,10 +1,6 @@
|
||||||
# -*- coding: utf-8 -*-
|
# -*- coding: utf-8 -*-
|
||||||
|
|
||||||
import logging
|
|
||||||
import requests
|
|
||||||
|
|
||||||
from collections import namedtuple
|
from collections import namedtuple
|
||||||
from requests import Request, Session
|
|
||||||
|
|
||||||
App = namedtuple('App', ['base_url', 'client_id', 'client_secret'])
|
App = namedtuple('App', ['base_url', 'client_id', 'client_secret'])
|
||||||
User = namedtuple('User', ['username', 'access_token'])
|
User = namedtuple('User', ['username', 'access_token'])
|
||||||
|
@ -12,106 +8,4 @@ User = namedtuple('User', ['username', 'access_token'])
|
||||||
DEFAULT_INSTANCE = 'mastodon.social'
|
DEFAULT_INSTANCE = 'mastodon.social'
|
||||||
|
|
||||||
CLIENT_NAME = 'toot - Mastodon CLI Interface'
|
CLIENT_NAME = 'toot - Mastodon CLI Interface'
|
||||||
CLIENT_WEB = 'https://github.com/ihabunek/toot'
|
CLIENT_WEBSITE = 'https://github.com/ihabunek/toot'
|
||||||
|
|
||||||
SCOPES = 'read write follow'
|
|
||||||
|
|
||||||
logger = logging.getLogger('toot')
|
|
||||||
|
|
||||||
|
|
||||||
def _log_request(request, prepared_request):
|
|
||||||
logger.debug(">>> \033[32m{} {}\033[0m".format(request.method, request.url))
|
|
||||||
logger.debug(">>> DATA: \033[33m{}\033[0m".format(request.data))
|
|
||||||
logger.debug(">>> FILES: \033[33m{}\033[0m".format(request.files))
|
|
||||||
logger.debug(">>> HEADERS: \033[33m{}\033[0m".format(request.headers))
|
|
||||||
|
|
||||||
|
|
||||||
def _log_response(response):
|
|
||||||
logger.debug("<<< \033[32m{}\033[0m".format(response))
|
|
||||||
logger.debug("<<< \033[33m{}\033[0m".format(response.json()))
|
|
||||||
|
|
||||||
|
|
||||||
def _get(app, user, url, params=None):
|
|
||||||
url = app.base_url + url
|
|
||||||
headers = {"Authorization": "Bearer " + user.access_token}
|
|
||||||
|
|
||||||
response = requests.get(url, params, headers=headers)
|
|
||||||
response.raise_for_status()
|
|
||||||
|
|
||||||
return response.json()
|
|
||||||
|
|
||||||
|
|
||||||
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, prepared_request)
|
|
||||||
|
|
||||||
response = session.send(prepared_request)
|
|
||||||
|
|
||||||
_log_response(response)
|
|
||||||
|
|
||||||
response.raise_for_status()
|
|
||||||
|
|
||||||
return response.json()
|
|
||||||
|
|
||||||
|
|
||||||
def create_app(base_url):
|
|
||||||
url = base_url + '/api/v1/apps'
|
|
||||||
|
|
||||||
response = requests.post(url, {
|
|
||||||
'client_name': 'toot - Mastodon CLI Interface',
|
|
||||||
'redirect_uris': 'urn:ietf:wg:oauth:2.0:oob',
|
|
||||||
'scopes': SCOPES,
|
|
||||||
'website': 'https://github.com/ihabunek/toot',
|
|
||||||
})
|
|
||||||
|
|
||||||
response.raise_for_status()
|
|
||||||
|
|
||||||
data = response.json()
|
|
||||||
client_id = data.get('client_id')
|
|
||||||
client_secret = data.get('client_secret')
|
|
||||||
|
|
||||||
return App(base_url, client_id, client_secret)
|
|
||||||
|
|
||||||
|
|
||||||
def login(app, username, password):
|
|
||||||
url = app.base_url + '/oauth/token'
|
|
||||||
|
|
||||||
response = requests.post(url, {
|
|
||||||
'grant_type': 'password',
|
|
||||||
'client_id': app.client_id,
|
|
||||||
'client_secret': app.client_secret,
|
|
||||||
'username': username,
|
|
||||||
'password': password,
|
|
||||||
'scope': SCOPES,
|
|
||||||
})
|
|
||||||
|
|
||||||
response.raise_for_status()
|
|
||||||
|
|
||||||
data = response.json()
|
|
||||||
access_token = data.get('access_token')
|
|
||||||
|
|
||||||
return User(username, access_token)
|
|
||||||
|
|
||||||
|
|
||||||
def post_status(app, user, status, visibility='public', media_ids=None):
|
|
||||||
return _post(app, user, '/api/v1/statuses', {
|
|
||||||
'status': status,
|
|
||||||
'media_ids[]': media_ids,
|
|
||||||
'visibility': visibility,
|
|
||||||
})
|
|
||||||
|
|
||||||
|
|
||||||
def timeline_home(app, user):
|
|
||||||
return _get(app, user, '/api/v1/timelines/home')
|
|
||||||
|
|
||||||
|
|
||||||
def upload_media(app, user, file):
|
|
||||||
return _post(app, user, '/api/v1/media', files={
|
|
||||||
'file': file
|
|
||||||
})
|
|
||||||
|
|
110
toot/api.py
Normal file
110
toot/api.py
Normal file
|
@ -0,0 +1,110 @@
|
||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
|
||||||
|
import logging
|
||||||
|
import requests
|
||||||
|
|
||||||
|
from requests import Request, Session
|
||||||
|
|
||||||
|
from toot import App, User, CLIENT_NAME, CLIENT_WEBSITE
|
||||||
|
|
||||||
|
SCOPES = 'read write follow'
|
||||||
|
|
||||||
|
logger = logging.getLogger('toot')
|
||||||
|
|
||||||
|
|
||||||
|
def _log_request(request, prepared_request):
|
||||||
|
logger.debug(">>> \033[32m{} {}\033[0m".format(request.method, request.url))
|
||||||
|
logger.debug(">>> DATA: \033[33m{}\033[0m".format(request.data))
|
||||||
|
logger.debug(">>> FILES: \033[33m{}\033[0m".format(request.files))
|
||||||
|
logger.debug(">>> HEADERS: \033[33m{}\033[0m".format(request.headers))
|
||||||
|
|
||||||
|
|
||||||
|
def _log_response(response):
|
||||||
|
logger.debug("<<< \033[32m{}\033[0m".format(response))
|
||||||
|
logger.debug("<<< \033[33m{}\033[0m".format(response.json()))
|
||||||
|
|
||||||
|
|
||||||
|
def _get(app, user, url, params=None):
|
||||||
|
url = app.base_url + url
|
||||||
|
headers = {"Authorization": "Bearer " + user.access_token}
|
||||||
|
|
||||||
|
response = requests.get(url, params, headers=headers)
|
||||||
|
response.raise_for_status()
|
||||||
|
|
||||||
|
return response.json()
|
||||||
|
|
||||||
|
|
||||||
|
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, prepared_request)
|
||||||
|
|
||||||
|
response = session.send(prepared_request)
|
||||||
|
|
||||||
|
_log_response(response)
|
||||||
|
|
||||||
|
response.raise_for_status()
|
||||||
|
|
||||||
|
return response.json()
|
||||||
|
|
||||||
|
|
||||||
|
def create_app(base_url):
|
||||||
|
url = base_url + '/api/v1/apps'
|
||||||
|
|
||||||
|
response = requests.post(url, {
|
||||||
|
'client_name': CLIENT_NAME,
|
||||||
|
'redirect_uris': 'urn:ietf:wg:oauth:2.0:oob',
|
||||||
|
'scopes': SCOPES,
|
||||||
|
'website': CLIENT_WEBSITE,
|
||||||
|
})
|
||||||
|
|
||||||
|
response.raise_for_status()
|
||||||
|
|
||||||
|
data = response.json()
|
||||||
|
client_id = data.get('client_id')
|
||||||
|
client_secret = data.get('client_secret')
|
||||||
|
|
||||||
|
return App(base_url, client_id, client_secret)
|
||||||
|
|
||||||
|
|
||||||
|
def login(app, username, password):
|
||||||
|
url = app.base_url + '/oauth/token'
|
||||||
|
|
||||||
|
response = requests.post(url, {
|
||||||
|
'grant_type': 'password',
|
||||||
|
'client_id': app.client_id,
|
||||||
|
'client_secret': app.client_secret,
|
||||||
|
'username': username,
|
||||||
|
'password': password,
|
||||||
|
'scope': SCOPES,
|
||||||
|
})
|
||||||
|
|
||||||
|
response.raise_for_status()
|
||||||
|
|
||||||
|
data = response.json()
|
||||||
|
access_token = data.get('access_token')
|
||||||
|
|
||||||
|
return User(username, access_token)
|
||||||
|
|
||||||
|
|
||||||
|
def post_status(app, user, status, visibility='public', media_ids=None):
|
||||||
|
return _post(app, user, '/api/v1/statuses', {
|
||||||
|
'status': status,
|
||||||
|
'media_ids[]': media_ids,
|
||||||
|
'visibility': visibility,
|
||||||
|
})
|
||||||
|
|
||||||
|
|
||||||
|
def timeline_home(app, user):
|
||||||
|
return _get(app, user, '/api/v1/timelines/home')
|
||||||
|
|
||||||
|
|
||||||
|
def upload_media(app, user, file):
|
||||||
|
return _post(app, user, '/api/v1/media', files={
|
||||||
|
'file': file
|
||||||
|
})
|
|
@ -15,8 +15,9 @@ from itertools import chain
|
||||||
from argparse import ArgumentParser, FileType
|
from argparse import ArgumentParser, FileType
|
||||||
from textwrap import TextWrapper
|
from textwrap import TextWrapper
|
||||||
|
|
||||||
from .config import save_user, load_user, load_app, save_app, CONFIG_APP_FILE, CONFIG_USER_FILE
|
from toot import DEFAULT_INSTANCE
|
||||||
from . import create_app, login, post_status, timeline_home, upload_media, DEFAULT_INSTANCE
|
from toot.api import create_app, login, post_status, timeline_home, upload_media
|
||||||
|
from toot.config import save_user, load_user, load_app, save_app, CONFIG_APP_FILE, CONFIG_USER_FILE
|
||||||
|
|
||||||
|
|
||||||
class ConsoleError(Exception):
|
class ConsoleError(Exception):
|
||||||
|
|
Loading…
Reference in a new issue