REST API v2

Работа с REST API

Пример поиска абонентов по коду

Python

Импортируем необходимые модули:

import json
import logging
import sys
from http import HTTPStatus
from urllib import parse as urllib_parse

import requests

Задаём параметры подключения к офису оператора связи (если планируется значительное количество запросов, стоит установить отдельную копию приложения hoper, чтобы работа с REST API не вызывала задержек в работе операторов через веб-интерфейс):

hoper_url = 'https://hydra.hoper.url/rest/v2/'
hoper_login = '########'
hoper_password = '*********'
http_timeout = 60

Задаём параметры поиска. Строка поиска - latera, тип 2001 - поиск по абонентам

search_string = 'latera'
search_subtype = 2001

Инициализируем логгер с выводом в stdout

logger = logging.getLogger()
logger.setLevel(logging.INFO)
stdout_logger = logging.StreamHandler()
logger.addHandler(stdout_logger)

Создаём HTTP-сессию в которой устанавливаем заголовки Content-Type и Accept

http_session = requests.Session()
http_session.headers.update(
    {
        'Content-Type': 'application/json',
        'Accept':       'application/json',
    },
)

Для работы с REST необходимо авторизоваться и получить токен для дальнейшего выполнения запросов.

Чтобы получить токен, подготавливаем запрос {"session": {"login": "########", "password": "********"}} и отправляем его по адресу /rest/v2/login

Если авторизация не выполнилась успешно, завершаем работу скрипта

auth_url = urllib_parse.urljoin(hoper_url, 'login')
auth_params = {'session': {'login': hoper_login, 'password': hoper_password}}
response = http_session.post(
    auth_url,
    timeout=http_timeout,
    data=json.dumps(auth_params),
)


if response.status_code != HTTPStatus.CREATED:
    logger.error(
        'Auth error ({0}): {1}'.format(response.status_code, response.content),
    )
    sys.exit(1)

Получаем токен, который сервер выдал в теле ответа

auth_result = json.loads(response.content)
auth_token = auth_result['session']['token']

и сохраняем его HTTP-сессии в заголовке Authorization со значением Token token=RECEIVED_TOKEN

http_session.headers.update(
    {'Authorization': 'Token token={0}'.format(auth_token)},
)

Подготавливаем и отправляем запрос на поиск абонентов

organizations_url = urllib_parse.urljoin(
    hoper_url,
    'search?result_subtype_id={0}&query={1}'.format(
        search_subtype,
        search_string,
        ),
)
response = http_session.get(
    organizations_url,
    timeout=http_timeout,
)

Если запрос выполнен успешно, по каждой найденной сущности выведем её идентификатор и наименование из результата поиска

if response.status_code == HTTPStatus.OK:
    search_results = json.loads(response.content)['search_results']
    for entity in search_results:
        logger.info(
            'Customer `{0}`: `{1}`'.format(
                entity['n_entity_id'],
                entity['vc_result_name'],
            ),
        )

Если запрос выполнился с ошибкой, выведем предупреждение с полученным от сервера ответом

else:
    logger.warning(
        'Invalid response ({0}): {1}'.format(
            response.status_code,
            response.content,
        ),
    )

Полная версия скрипта:

rest-example.py

"""Hydra REST v2 example."""
import json
import logging
import sys
from http import HTTPStatus
from urllib import parse as urllib_parse

import requests

hoper_url = 'https://hydra.hoper.url/rest/v2/'
hoper_login = '########'
hoper_password = '********'
http_timeout = 60

search_string = 'latera'
search_subtype = 2001

logger = logging.getLogger()
logger.setLevel(logging.INFO)
stdout_logger = logging.StreamHandler()
logger.addHandler(stdout_logger)

http_session = requests.Session()
http_session.headers.update(
    {
        'Content-Type': 'application/json',
        'Accept':       'application/json',
    },
)

auth_url = urllib_parse.urljoin(hoper_url, 'login')
auth_params = {'session': {'login': hoper_login, 'password': hoper_password}}
response = http_session.post(
    auth_url,
    timeout=http_timeout,
    data=json.dumps(auth_params),
)

if response.status_code != HTTPStatus.CREATED:
    logger.error(
        'Auth error ({0}): {1}'.format(response.status_code, response.content),
    )
    sys.exit(1)

auth_result = json.loads(response.content)
auth_token = auth_result['session']['token']

http_session.headers.update(
    {'Authorization': 'Token token={0}'.format(auth_token)},
)

organizations_url = urllib_parse.urljoin(
    hoper_url,
    'search?result_subtype_id={0}&query={1}'.format(
        search_subtype,
        search_string,
    ),
)
response = http_session.get(
    organizations_url,
    timeout=http_timeout,
)

if response.status_code == HTTPStatus.OK:
    search_results = json.loads(response.content)['search_results']
    for entity in search_results:
        logger.info(
            'Customer `{0}`: `{1}`'.format(
                entity['n_entity_id'],
                entity['vc_result_name'],
            ),
        )
else:
    logger.warning(
        'Invalid response ({0}): {1}'.format(
            response.status_code,
            response.content,
        ),
    )

Page tree

REST API v2

Работа с REST API

Python