aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/vk/api.py
blob: 392d4b83ff8f55c7602f1b1113715ff9f2801a9d (plain) (blame)
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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
# Copyright (c) 2015 Egor Tensin <Egor.Tensin@gmail.com>
# This file is part of the "VK scripts" project.
# For details, see https://github.com/egor-tensin/vk-scripts.
# Distributed under the MIT License.

from collections import Iterable
from collections.abc import Mapping
from enum import Enum
import json
from urllib.error import URLError
import urllib.parse
from urllib.request import urlopen

import vk.error
from vk.user import User

def _split_url(url):
    return urllib.parse.urlsplit(url)[:3]

def _is_empty_param_value(value):
    return isinstance(value, str) and not value

def _filter_empty_params(params, empty_params=False):
    if empty_params:
        return params
    if isinstance(params, Mapping):
        return {name: value for name, value in params.items() if not _is_empty_param_value(value)}
    elif isinstance(params, Iterable):
        return [(name, value) for name, value in params if not _is_empty_param_value(value)]
    else:
        raise TypeError()

def _build_url(scheme, host, path, params=None, empty_params=False):
    if params is None:
        params = {}
    if isinstance(params, Mapping) or isinstance(params, Iterable):
        params = _filter_empty_params(params, empty_params)
        params = urllib.parse.urlencode(params)
    elif isinstance(params, str):
        pass
    else:
        raise TypeError()
    path = urllib.parse.quote(path)
    return urllib.parse.urlunsplit((scheme, host, path, params, ''))

def _join_param_values(values):
    if isinstance(values, str):
        return values
    elif isinstance(values, Iterable):
        return ','.join(map(str, values))
    else:
        return values

def _join_path(base, url):
    if not base.endswith('/'):
        base += '/'
    return urllib.parse.urljoin(base, url)

class Version(Enum):
    V5_73 = '5.73'
    DEFAULT = V5_73

    def __str__(self):
        return self.value

class Language(Enum):
    EN = 'en'
    DEFAULT = EN

    def __str__(self):
        return self.value

class Method(Enum):
    USERS_GET = 'users.get'
    FRIENDS_GET = 'friends.get'

    def __str__(self):
        return self.value

class CommonParameters(Enum):
    VERSION = 'v'
    LANGUAGE = 'lang'

    def __str__(self):
        return self.value

class API:
    _ROOT_URL = 'https://api.vk.com/method/'

    _SCHEME, _HOST, _ROOT_PATH = _split_url(_ROOT_URL)

    def __init__(self, version=Version.DEFAULT, lang=Language.DEFAULT):
        self._common_params = {
            CommonParameters.VERSION: version,
            CommonParameters.LANGUAGE: lang,
        }

    def _build_method_url(self, method, **params):
        path = _join_path(self._ROOT_PATH, str(method))
        params.update(self._common_params)
        return _build_url(self._SCHEME, self._HOST, path, params)

    def _call_method(self, method, **params):
        url = self._build_method_url(method, **params)
        #print(url)
        try:
            with urlopen(url) as response:
                response = json.loads(response.read().decode())
                #print(response)
                if 'response' not in response:
                    raise vk.error.InvalidAPIResponseError(response)
                return response['response']
        except (ConnectionError, URLError) as e:
            raise vk.error.APIConnectionError(str(e)) from e

    def _filter_response_with_users(self, user_list, deactivated_users=True):
        user_list = map(User.from_api_response, user_list)
        if deactivated_users:
            return user_list
        return [user for user in user_list if not user.is_deactivated()]

    def users_get(self, user_ids, fields=(), deactivated_users=True):
        return self._filter_response_with_users(self._call_method(
            Method.USERS_GET,
            user_ids=_join_param_values(user_ids),
            fields=_join_param_values(fields)), deactivated_users)

    def friends_get(self, user_id, fields=(), deactivated_users=True):
        response = self._call_method(
            Method.FRIENDS_GET,
            user_id=user_id,
            fields=_join_param_values(fields))
        if 'items' not in response:
            raise vk.error.InvalidAPIResponseError(response)
        return self._filter_response_with_users(response['items'], deactivated_users)