BANKS FAQ : Пример API клиента

c = CashoffAPIClient('https://developer.cashoff.ru', 'test', 'test')

c.request('/users')
c.request('/users/add', body={'ext_id': '123456'})

c.request('/accounts/5612', method='DELETE')
c.request('/accounts', {'profile_id': 5235})

c.request('/profiles/add', body={'provider_key': 'sber'}, user_id=2512)

import hashlib
import hmac
import json
import time
from base64 import b64encode
from http.client import HTTPConnection, HTTPSConnection
from urllib.parse import urlencode, urlsplit, quote


class CashoffAPIClient:
    """Базовый клиент для работы с CASHOFF API через python3."""

    def __init__(self, host, login, secret, use_sign=True):
        """
        :param host: Адрес, с указанием протокола
        :param login: логин (ServiceID)
        :param secret: пароль (PrivateKey)
        :param use_sign: использовать для аутентификации подпись; В случае False будет использовано http basic auth
        """
        url_parts = urlsplit(host)
        connection_cls = HTTPConnection if url_parts.scheme == 'http' else HTTPSConnection
        self.connection = connection_cls(url_parts.hostname, url_parts.port)

        self.login = login
        self.secret = secret
        self.use_sign = use_sign

    def request(self, url, params=None, body=None, *, user_id=None, session=None, method=None):
        """
        Выполнить GET/POST запрос
        :param url: относительный путь к запросу, как в документации
        :param params: get-параметры запроса в виде dict()
        :param body: тело, если POST; автоматически конвертируется в json
        :param user_id: идентификатор юзера; подставится в заголовок co-user-id
        :param session: идентификатор сессии; подставится в заголовок co-session, будет использоваться вместо подписи
        :param method: http-метод; по умолчанию будет подставлен get/post, в зависимости от body
        :return: пара (http код, распарсеное тело ответа/текст в случае неудачи)
        """
        url = self._format_request_url(url, params)
        if method is None:
            method = 'GET' if body is None else 'POST'

        if body is not None:
            body = json.dumps(body, ensure_ascii=False)
        headers = self._make_headers(url, body, user_id, session)
        if body is not None:
            body = body.encode('utf-8')

        try:
            self.connection.request(method, url, body, headers)
        except ConnectionResetError:
            self.connection.close()
            self.connection.request(method, url, body, headers)

        response = self.connection.getresponse()
        result = response.read().decode('utf-8')
        result = result and json.loads(result)

        return response.status, result

    def _make_headers(self, url, body, user_id=None, session=None):
        """Генерирует заголовки для запроса"""
        headers = {
            'Content-type': 'application/json',
        }

        if session:
            headers['co-session'] = session
        else:
            if self.use_sign:
                headers['co-auth'] = self._make_sign(url, body, user_id)
            else:
                headers['co-auth'] = 'http_basic'
                headers['Authorization'] = self._make_http_basic()

        if user_id is not None:
            headers['co-user-id'] = str(user_id)

        return headers

    def _make_sign(self, url, body, user_id):
        timestamp = int(time.time())
        msg = (
                self.login +
                str(timestamp) +
                (str(user_id) if user_id is not None else '') +
                self._get_url_for_sign(url) +
                (body or '')
        )
        signature = hmac.new(
            self.secret.encode('utf-8'), msg.encode('utf-8'), hashlib.sha1
        )
        return f'sha1|{self.login}:{timestamp}:{signature.hexdigest()}'

    def _make_http_basic(self):
        auth_data = quote(self.login) + ':' + quote(self.secret)
        return 'Basic ' + b64encode(auth_data.encode()).decode()

    def _get_url_for_sign(self, url):
        parts = urlsplit(url)
        url_for_sign = parts.path
        if parts.query:
            url_for_sign += '?' + parts.query
        return url_for_sign

    def _format_request_url(self, url, params):
        if url.startswith('/'):
            url = url[1:]
        if params is not None:
            url += '?' + urlencode(params)
        return '/api/v2/' + url