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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
|
import json
import platform
import re
from typing import Any, Callable, Dict, Optional, Tuple, Union
from urllib.parse import urlparse
from . import __version__
from .http_adapter import HTTPConnectionError
from .urllib_http_adapter import URLLibHttpAdapter
class APIError(Exception):
def __init__(self, e: str, retryable: bool):
super().__init__(e)
self.retryable = retryable
class Api:
def __init__(
self,
url: str,
user: Optional[str],
install_id: str,
http_adapter: Any = None,
) -> None:
self.url = url
self.user = user
self.install_id = install_id
self.http_adapter = (
http_adapter if http_adapter is not None else URLLibHttpAdapter()
)
def hostname(self) -> Optional[str]:
return urlparse(self.url).hostname
def auth_url(self) -> str:
return f"{self.url}/connect/{self.install_id}"
def upload_url(self) -> str:
return f"{self.url}/api/asciicasts"
def upload_asciicast(self, path_: str) -> Tuple[Any, Any]:
with open(path_, "rb") as f:
try:
status, headers, body = self.http_adapter.post(
self.upload_url(),
files={"asciicast": ("ascii.cast", f)},
headers=self._headers(),
username=self.user,
password=self.install_id,
)
except HTTPConnectionError as e:
raise APIError(str(e), True) from e
if status not in (200, 201):
self._handle_error(status, body)
if (headers.get("content-type") or "")[0:16] == "application/json":
result = json.loads(body)
else:
result = {"url": body}
return result, headers.get("Warning")
def _headers(self) -> Dict[str, Union[Callable[[], str], str]]:
return {"user-agent": self._user_agent(), "accept": "application/json"}
@staticmethod
def _user_agent() -> str:
os = re.sub("([^-]+)-(.*)", "\\1/\\2", platform.platform())
return (
f"asciinema/{__version__} {platform.python_implementation()}"
f"/{platform.python_version()} {os}"
)
@staticmethod
def _handle_error(status: int, body: bytes) -> None:
errors = {
400: f"Invalid request: {body.decode('utf-8', 'replace')}",
401: "Invalid or revoked install ID",
404: (
"API endpoint not found. "
"This asciinema version may no longer be supported. "
"Please upgrade to the latest version."
),
413: "Sorry, the size of your recording exceeds the server-configured limit.",
422: f"Invalid asciicast: {body.decode('utf-8', 'replace')}",
503: "The server is down for maintenance.",
}
error = errors.get(status)
if not error:
if status >= 500:
error = (
"The server is having temporary problems. "
"Try again in a minute."
)
else:
error = f"HTTP status: {status}"
raise APIError(error, status >= 500)
|
