...
Code Block |
---|
|
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,
datajson=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) |
Получаем токен, который сервер выдал в теле Логируем ответ сервера и получаем токен из тела ответа
Code Block |
---|
|
logger.debug(response.content)
auth_result = json.loads(response.content)
auth_token = auth_result['session']['token']
|
...
Code Block |
---|
|
organizations_url = urllib_parse.urljoin(
hoper_url, 'search')
response = 'search?result_subtype_id={0}&query={1}'.formathttp_session.get(
organizations_url,
search_subtype timeout=http_timeout,
params={
'result_subtype_id': search_stringsubtype,
),
)
response = http_session.get('query': organizations_url, search_string
timeout=http_timeout,}
)
|
Если запрос выполнен успешно, по каждой найденной сущности выведем её идентификатор и наименование из результата поиска
...
Code Block |
---|
language | py |
---|
title | rest-example.py |
---|
linenumbers | true |
---|
collapse | true |
---|
|
"""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,
datajson=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)
logger.debug(response.content)
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')
response = hoperhttp_session.get(
organizations_url,
'search?result_subtype_id={0}&query={1}'.format(timeout=http_timeout,
params={
'result_subtype_id': search_subtype,
search_string,'query': ), ) response = http_session.get( organizationssearch_url,string
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,
),
)
|
PHP
Задаём параметры подключения к офису оператора связи:
...