thin-pleroma-wrapper/api.py

33 lines
1.1 KiB
Python

import aiohttp
secure=True
def url(endpoint,version=1):
s='s' if secure else ''
return f'http{s}://{instance}/api/v{version}/{endpoint}'
#cs=aiohttp.ClientSession() # Fuck your warnings
#Fuck your errors too
async def call(method,endpoint,version=1,headers={},**kwargs):
async with aiohttp.ClientSession() as sess:
async with sess.request(method, url(endpoint,version), headers={'Authorization':f'Bearer {token}'}|headers, **kwargs) as r:
try: return await r.json()
except aiohttp.client_exceptions.ContentTypeError: raise Exception(await r.read())
async def post(text,images=[],json={},**kwargs):
j={'status':text}
pics=[]
for image in images:
f=aiohttp.FormData()
f.add_field('file',image[1],filename=image[0])
pics.append(await call('POST','media',2,data=f))
if pics: j['media_ids']=list(map(lambda x:x['id'],pics))
await call('POST','statuses',json=j|json,**kwargs)
async def stream(method,endpoint,params,needle,*args,**kwargs):
p={'limit':40}|params
while True:
page=await call(method,endpoint,params=p,*args,**kwargs)
if not page: break # Handles a particular edge case
for n in page: yield n
p|=needle(page)
if len(page)<40: break