Spaces:
Runtime error
Runtime error
File size: 3,224 Bytes
932db78 36b9958 932db78 9a3c8f5 932db78 0fd4a4d 932db78 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 |
"""This module proivdes wrapper functionality for the imgur API"""
import dotenv, os
from PIL import Image
import asyncio
from aiohttp import ClientSession
from typing import Optional, Union, Tuple
from .utils import image_to_b64_string, bytes_to_image
dotenv.load_dotenv()
API_ENDPOINTS = {
'upload': 'https://api.imgur.com/3/upload/',
'download': 'http://i.imgur.com/',
'info': 'https://api.imgur.com/3/image/',
'delete': 'https://api.imgur.com/3/image/',
'auth': f'https://api.imgur.com/oauth2/token'
}
# get access and refresh token
async def get_tokens():
session = ClientSession()
r = await session.request(
method='post',
url=API_ENDPOINTS['auth'],
data={
'refresh_token': os.getenv("IS3_REFRESH_TOKEN"),
'client_id': os.getenv("IS3_CLIENT_ID"),
'client_secret': os.getenv("IS3_CLIENT_SECRET"),
'grant_type': 'refresh_token',
}
)
r = await r.json(content_type='application/json')
return r['access_token'], r['refresh_token']
ACCESS_TOKEN, REFRESH_TOKEN = asyncio.run(get_tokens())
# AUTH_HEADER = {'Authorization': f"Client-ID {os.getenv('IS3_CLIENT_ID')}"}
AUTH_HEADER = {'Authorization': f"Bearer {ACCESS_TOKEN}"}
class ImgurClient:
"""Class to interact with various API endpoints"""
def __init__(self, session: Optional[ClientSession] = None) -> None:
self._session = session or ClientSession()
async def __aenter__(self):
return self
async def __aexit__(self, *err):
await self._session.close()
async def _request(self, method: str, url: str, *args, **kwargs) -> Union[dict, bytes]:
"""Make a request with the specified method to the endpoint. All requests
should either return raw image data as bytes or other data as JSON"""
async with self._session.request(method, url, *args, **kwargs) as resp:
content_type = resp.content_type
if content_type == 'image/png':
return await resp.read()
elif content_type == 'application/json':
return (await resp.json())['data']
else:
raise RuntimeError(f'Unexpected response content-type "{content_type}"')
async def upload_image(self, img: Image.Image) -> Tuple[str, str]:
"""Upload an image and return img id and deletehash"""
data = image_to_b64_string(img)
r = await self._request(
method='post',
url=API_ENDPOINTS['upload'],
headers=AUTH_HEADER,
data={'image': data, 'type': 'base64'}
)
return r['id'], r['deletehash']
async def download_image(self, image_id: str) -> Image.Image:
"""Download the image and return the data as bytes."""
url = API_ENDPOINTS['download'] + image_id + '.png'
data = await self._request('get', url)
return bytes_to_image(data)
async def delete_image(self, deletehash: str) -> None:
"""Delete an image using a deletehash string"""
url = API_ENDPOINTS['delete'] + deletehash
await self._request('delete', url, headers=AUTH_HEADER)
|