File: api.py

package info (click to toggle)
asciinema 2.0.2-1
  • links: PTS, VCS
  • area: main
  • in suites: buster
  • size: 392 kB
  • sloc: python: 1,304; sh: 53; makefile: 35
file content (85 lines) | stat: -rw-r--r-- 2,798 bytes parent folder | download | duplicates (2)
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
import platform
import re
import json
from urllib.parse import urlparse

from asciinema import __version__
from asciinema.urllib_http_adapter import URLLibHttpAdapter
from asciinema.http_adapter import HTTPConnectionError


class APIError(Exception):
    pass


class Api:

    def __init__(self, url, user, install_id, http_adapter=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):
        return urlparse(self.url).hostname

    def auth_url(self):
        return "{}/connect/{}".format(self.url, self.install_id)

    def upload_url(self):
        return "{}/api/asciicasts".format(self.url)

    def upload_asciicast(self, path):
        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))

        if status != 200 and status != 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):
        return {'User-Agent': self._user_agent(), 'Accept': 'application/json'}

    def _user_agent(self):
        os = re.sub('([^-]+)-(.*)', '\\1/\\2', platform.platform())

        return 'asciinema/%s %s/%s %s' % (__version__,
                                          platform.python_implementation(),
                                          platform.python_version(),
                                          os
                                          )

    def _handle_error(self, status, body):
        errors = {
            400: "Invalid request: %s" % body,
            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, your asciicast is too big.",
            422: "Invalid asciicast: %s" % body,
            503: "The server is down for maintenance. Try again in a minute."
        }

        error = errors.get(status)

        if not error:
            if status >= 500:
                error = "The server is having temporary problems. Try again in a minute."
            else:
                error = "HTTP status: %i" % status

        raise APIError(error)