# coding=utf8
import logging
import logging.config
from vk.logs import LOGGING_CONFIG
from vk.utils import stringify_values, json_iter_parse, LoggingSession, str_type
from vk.exceptions import VkAuthError, VkAPIError
from vk.mixins import AuthMixin, InteractiveMixin
VERSION = '2.0'
logging.config.dictConfig(LOGGING_CONFIG)
logger = logging.getLogger('vk')
[docs]class Session(object):
API_URL = 'https://api.vk.com/method/'
def __init__(self, access_token=None):
logger.debug('API.__init__(access_token=%(access_token)r)', {'access_token': access_token})
self.access_token = access_token
self.access_token_is_needed = False
self.requests_session = LoggingSession()
self.requests_session.headers['Accept'] = 'application/json'
self.requests_session.headers['Content-Type'] = 'application/x-www-form-urlencoded'
@property
def access_token(self):
logger.debug('Check that we need new access token')
if self.access_token_is_needed:
logger.debug('We need new access token. Try to get it.')
self.access_token = self.get_access_token()
else:
logger.debug('Use old access token')
return self._access_token
@access_token.setter
def access_token(self, value):
self._access_token = value
if isinstance(value, str_type) and len(value) >= 12:
self.censored_access_token = '{}***{}'.format(value[:4], value[-4:])
else:
self.censored_access_token = value
logger.debug('access_token = %r', self.censored_access_token)
self.access_token_is_needed = not self._access_token
def get_user_login(self):
logger.debug('Do nothing to get user login')
def get_access_token(self):
"""
Dummy method
"""
logger.debug('API.get_access_token()')
return self._access_token
def make_request(self, method_request, captcha_response=None):
logger.debug('Prepare API Method request')
response = self.send_api_request(method_request)
# todo Replace with something less exceptional
response.raise_for_status()
# there are may be 2 dicts in one JSON
# for example: "{'error': ...}{'response': ...}"
for response_or_error in json_iter_parse(response.text):
if 'response' in response_or_error:
# todo Can we have error and response simultaneously
# for error in errors:
# logger.warning(str(error))
return response_or_error['response']
elif 'error' in response_or_error:
error_data = response_or_error['error']
error = VkAPIError(error_data)
if error.is_captcha_needed():
captcha_key = self.get_captcha_key(error.captcha_img)
if not captcha_key:
raise error
captcha_response = {
'sid': error.captcha_sid,
'key': captcha_key,
}
return self.make_request(method_request, captcha_response=captcha_response)
elif error.is_access_token_incorrect():
logger.info('Authorization failed. Access token will be dropped')
self.access_token = None
return self.make_request(method_request)
else:
raise error
def send_api_request(self, request, captcha_response=None):
url = self.API_URL + request._method_name
method_args = request._api._method_default_args.copy()
method_args.update(stringify_values(request._method_args))
access_token = self.access_token
if access_token:
method_args['access_token'] = access_token
if captcha_response:
method_args['captcha_sid'] = captcha_response['sid']
method_args['captcha_key'] = captcha_response['key']
timeout = request._api._timeout
response = self.requests_session.post(url, method_args, timeout=timeout)
return response
def get_captcha_key(self, captcha_image_url):
"""
Default behavior on CAPTCHA is to raise exception
Reload this in child
"""
return None
def auth_code_is_needed(self, content, session):
"""
Default behavior on 2-AUTH CODE is to raise exception
Reload this in child
"""
raise VkAuthError('Authorization error (2-factor code is needed)')
def auth_captcha_is_needed(self, content, session):
"""
Default behavior on CAPTCHA is to raise exception
Reload this in child
"""
raise VkAuthError('Authorization error (captcha)')
def phone_number_is_needed(self, content, session):
"""
Default behavior on PHONE NUMBER is to raise exception
Reload this in child
"""
logger.error('Authorization error (phone number is needed)')
raise VkAuthError('Authorization error (phone number is needed)')
class API(object):
def __init__(self, session, timeout=10, **method_default_args):
self._session = session
self._timeout = timeout
self._method_default_args = method_default_args
def __getattr__(self, method_name):
return Request(self, method_name)
def __call__(self, method_name, **method_kwargs):
return getattr(self, method_name)(**method_kwargs)
class Request(object):
__slots__ = ('_api', '_method_name', '_method_args')
def __init__(self, api, method_name):
self._api = api
self._method_name = method_name
def __getattr__(self, method_name):
return Request(self._api, self._method_name + '.' + method_name)
def __call__(self, **method_args):
self._method_args = method_args
return self._api._session.make_request(self)
class AuthSession(AuthMixin, Session):
pass
class InteractiveSession(InteractiveMixin, Session):
pass
class InteractiveAuthSession(InteractiveMixin, AuthSession):
pass