77 lines
2.1 KiB
Python
77 lines
2.1 KiB
Python
import logging
|
|
|
|
from settings import MASTODON_INSTANCE, MASTODON_ACCESS_TOKEN
|
|
|
|
API_STATUSES = '/api/v1/statuses'
|
|
API_STATUS_CONTEXT = '/api/v1/statuses/%(id)s/context'
|
|
API_STATUS_FAVORITES = '/api/v1/statuses/%(id)s/favourited_by'
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class Mastodon:
|
|
def __init__(self, aiosession, instance=None, access_token=None):
|
|
self.aiosession = aiosession
|
|
self.instance = instance if instance is not None else MASTODON_INSTANCE
|
|
self.access_token = access_token if access_token is not None else MASTODON_ACCESS_TOKEN
|
|
|
|
async def get(self, instance, request, params={}):
|
|
async with self.aiosession.get(
|
|
'https://{}{}'.format(
|
|
instance, request
|
|
),
|
|
params=params,
|
|
headers={
|
|
'Authorization': f'Bearer {self.access_token}'
|
|
} if self.access_token else {}
|
|
) as response:
|
|
if not response.ok:
|
|
logger.error('{} {}'.format(request, response.status))
|
|
raise Exception('{} {}'.format(request, response.status))
|
|
|
|
return await response.json()
|
|
|
|
async def post(self, instance, request, data):
|
|
async with self.aiosession.post(
|
|
'https://{}{}'.format(
|
|
instance,
|
|
request,
|
|
),
|
|
json=data,
|
|
headers={
|
|
'Authorization': f'Bearer {self.access_token}'
|
|
} if self.access_token else {}
|
|
) as response:
|
|
|
|
if not response.ok:
|
|
logger.error('{} {} {}'.format(instance, request, response.status))
|
|
raise Exception('{} {} {}'.format(instance, request, response.status))
|
|
|
|
return await response.json()
|
|
|
|
async def statusPost(self, status, user, in_reply_to_id=None):
|
|
visibility = user.mastodon_visibility()
|
|
|
|
return await self.post(
|
|
self.instance,
|
|
API_STATUSES,
|
|
data={
|
|
'status': status,
|
|
'visibility': visibility,
|
|
'in_reply_to_id': in_reply_to_id
|
|
})
|
|
|
|
async def statusContext(self, id):
|
|
return await self.get(
|
|
self.instance,
|
|
API_STATUS_CONTEXT % {
|
|
'id': id
|
|
}
|
|
)
|
|
|
|
async def statusFavorites(self, id):
|
|
return await self.get(
|
|
self.instance,
|
|
API_STATUS_FAVORITES % {
|
|
'id' : id
|
|
}
|
|
)
|