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
|
import logging
from abc import ABCMeta
from typing import Optional
from aiohttp import ClientConnectorError, ClientSession
from discord.ext.commands import CogMeta
from bot.constants import URLs
from bot.utils.redis_cache import RedisCache
log = logging.getLogger(__name__)
FAILED_REQUEST_ATTEMPTS = 3
__all__ = ['RedisCache', 'CogABCMeta', "send_to_paste_service"]
class CogABCMeta(CogMeta, ABCMeta):
"""Metaclass for ABCs meant to be implemented as Cogs."""
pass
async def send_to_paste_service(http_session: ClientSession, contents: str, *, extension: str = "") -> Optional[str]:
"""
Upload `contents` to the paste service.
`http_session` should be the current running ClientSession from aiohttp
`extension` is added to the output URL
When an error occurs, `None` is returned, otherwise the generated URL with the suffix.
"""
extension = extension and f".{extension}"
log.debug(f"Sending contents of size {len(contents.encode())} bytes to paste service.")
paste_url = URLs.paste_service.format(key="documents")
for attempt in range(1, FAILED_REQUEST_ATTEMPTS + 1):
try:
async with http_session.post(paste_url, data=contents) as response:
response_json = await response.json()
except ClientConnectorError:
log.warning(
f"Failed to connect to paste service at url {paste_url}, "
f"trying again ({attempt}/{FAILED_REQUEST_ATTEMPTS})."
)
continue
except Exception:
log.exception(
f"An unexpected error has occurred during handling of the request, "
f"trying again ({attempt}/{FAILED_REQUEST_ATTEMPTS})."
)
continue
if "message" in response_json:
log.warning(
f"Paste service returned error {response_json['message']} with status code {response.status}, "
f"trying again ({attempt}/{FAILED_REQUEST_ATTEMPTS})."
)
continue
elif "key" in response_json:
log.info(f"Successfully uploaded contents to paste service behind key {response_json['key']}.")
return URLs.paste_service.format(key=response_json['key']) + extension
log.warning(
f"Got unexpected JSON response from paste service: {response_json}\n"
f"trying again ({attempt}/{FAILED_REQUEST_ATTEMPTS})."
)
|
