Кратко: запросы к API и разбор XML-ответов. Python

Этот пост предназначен в первую очередь для новичков в разработке, впервые столкнувшихся с необходимостью отправить post/get запросы к какому-нибудь API и проанализировать полученный в XML ответ. Постаралась собрать необходимы минимум в одном месте.

1. Непосредственно API

Оно может быть от ваших коллег, партнеров, заказчиков, сторонних сервисов. Разной степени готовности и актуальности. Но у него как правило есть заголовки запросов, параметры запросов, параметры ответа и статус-коды ответов. Например:

Если проверить сам метод на работоспособность, например с помощью curl - можно сэкономить себе много нервов и сил, особенно актуально если API допиливается одновременно с вашей разработкой. После чего можно воспользоваться например вот таким сервисом https://reqbin.com/req/python/c-xgafmluu/convert-curl-to-python-requests для того что бы перевести curl-запрос в код под либу requests.

Воспользуемся модулем CaseInsensitiveDict из requests.structures для того что бы собирать заголовки запроса в привычные headers. Параметры запроса определим в параметре files. Тогда для метода с первого скрина post-запрос будет выглядеть следующим образом:

import requests
from requests.structures import CaseInsensitiveDict

base_url = 'https://myaccount.ispringlearn.ru'
headers = CaseInsensitiveDict()
headers["Host"] = 'api-learn.ispringlearn.ru'
headers["X-Auth-Account-Url"] = "https://myaccount.ispringlearn.ru/"
headers["X-Auth-Email"] = " email@email.com"
headers["X-Auth-Password"] = "12345Q"

id_test_course = '14b847e8-c10f-11ea-b4e1-ae33e75597e9'
id_test_enrollment = 'e1f2c50e-c1ae-11ea-8592-a6eabe1809b'
url = f"{base_url}/statistics/course"
files = {'courseId': (None, f'{id_test_course}'),
         'enrollmentIds': (None, f'{id_test_enrollment}'),
         'status': (None, 'automatic')         
        }
responce = requests.post(url=url, headers=headers, files=files)
if responce.status_code != 200:
    raise ValueError(f"Request add_user failed {responce.status_code}")
resp_xml_content = responce.content
print(f"{resp_xml_content}")

Вариант с ответом в json в этой статье разбирать не буду, по нему написано очень много материалов, поэтому остановлюсь подробнее на XML.

2. XML. Чтение и разбор

После получения XML-портянки на этапе отладки не лишним будет проверить визуально данные в ней. Например с помощью сервиса https://jsonformatter.org/xml-parser.

В приходящем респонсе байтовая кодировка(необходимая для раскладывания xml по дереву) находится только в атрибуте content. перепишем для дальнейшей работы его в отдельную переменную responce_xml_content.

В дебагере очень похоже выглядит атрибут text у полученного responce, но он там существует как utf-8.

Большая часть мануалов по парсингу xml написана под чтение из файла и под библиотеку etree. И метод для строки из переменной fromstring в каждом классе работает несколько по разному.

Поэтому оптимальным считаю использование etree из модуля lxml. С ним проверка существования пользователя get-запросом и добавление пользователя post-запросом выглядит лаконично.

import re, os
import requests
from requests.structures import CaseInsensitiveDict
from config import Config
from lxml import etree
from loguru import logger


class ApiRequest:
    def init(self, new_user) -> None:
        self.base_url = Config.base_url
        self.headers = CaseInsensitiveDict()
        self.headers["Host"] = Config.Host
        self.headers["X-Auth-Account-Url"] = Config.X_Auth_Account_Url
        self.headers["X-Auth-Email"] = Config.X_Auth_Email
        self.headers["X-Auth-Password"] = Config.X_Auth_Password
        self.new_user = new_user
        self.default_department_id = Config.default_department_id
        self.dueDate = Config.dueDate
        self.re_login = re.compile('(?P<login>\w+)@', re.M | re.S)
        
    def check_exist_user(self) -> bool:
        """example of get request"""
        url = f"{self.base_url}/user"
        email = self.new_user.email
        resp = requests.get(url, headers=self.headers)
        if resp.status_code != 200:
            raise ValueError(f"Request check_exist_user failed {resp.status_code}")
        resp_xml_content = resp.content
        tree = etree.XML(resp_xml_content)
        user_by_email = tree.xpath(
            f'/response/userProfile/fields/field[name = "EMAIL" and value = "{email}"]')
        if user_by_email:
            logger.info(f"Пользователя с email {email} еще не существует")
            return False
        logger.info(f"Пользователь с email {email} уже существует")
        self.new_user.user_id = (tree.xpath(
            f".//userProfile[./fields/field/name[contains(text(), 'EMAIL')] and ./fields/field/value[contains(text(), '{email}')]]/userId"))[
            0].text
        return True


    def add_user(self) -> bool:
        """example of post request"""
        url = f"{self.base_url}/user"
        login = self.re_login.search(self.new_user.email).group('login')

        files = {
            'departmentId': (None, f'{self.default_department_id}'),
            'fields[email]': (None, f'{self.new_user.email}'),
            'fields[login]': (None, f'{login}'),
        }

        response = requests.post(url=url, headers=self.headers, files=files)

        if response.status_code != 201:
            raise ValueError(f"Request add_user failed {response.status_code}")
        resp_xml_content = resp.content
        try:
            self.new_user.user_id = etree.XML(resp_xml_content).text
            logger.info(f"Add new user successful. User_id: {self.new_user.user_id}")
            return True
        except Exception as ex:
            logger.error(f"Error get user_id from responce {ex}")
            return False

3. Анализ XML. XPath или Bs4

XPath - язык на котором необходимо будет доставать цепочки элементов. И на нем необходимо будет писать выражения для поиска элемента по параметрам ответа.

Примечательно что родители знают о своих детях, но дети не знают о своих родителях. Напрямую, задав путь до проверки атрибута элемента на пятом уровне вложенности, в случае если такой найден, забрать атрибут с его третьего уровня вложенности не получится.

Один из немногих мануалов по тому как с ним обращаться http://www.k-press.ru/cs/2001/2/XPath/XPath4.asp

Например для того что бы достать из списка пользователей id пользователя по искомому email, находящегося в атрибуте value в элементе в котором атрибут name со значением email выражение будет выглядеть так:

self.new_user.user_id = (tree.xpath(
            f".//userProfile[./fields/field/name[contains(text(), 'EMAIL')] and ./fields/field/value[contains(text(), '{email}')]]/userId"))[
            0].text

Есть еще вариант использовать для этого Bs4. Тогда тот же поиск будет выглядеть так:

import bs4
import re
soup = bs4.BeautifulSoup(html, 'lxml')
found_user = soup.find_all('value', text=re.compile(f'{email}'))
userProfile = (found_user[0]).parent.parent.parent
userid = (userProfile.find('userid')).text

4. Размещаем эти запросы на flask.

Рекоммендуется использовать стандартную структуру страниц.

В api.py расположим основной обработчик страниц. Для обработки post-запроса с json-ом и примером формы. Для валидации самым лаконичным решением будет pydantic,из которого потребуется BaseModel, ValidationError, validator

Если потребуется styles.css то располагаться ему следует в /static/css.

from flask import Flask, jsonify, request, render_template, url_for, escape, make_response
from flask_wtf import FlaskForm
from loguru import logger
from pydantic import BaseModel, ValidationError, validator
from wtforms import StringField
import re, os
from api_requests import ApiRequest


DEBUG = True
SECRET_KEY = ''

app = Flask(__name__)
app.config.from_object(__name__)


class User(BaseModel): 
    """
    Модель пользователя. С примером валидации данных в JSON
    """

    name: str
    surname: str
    email: str
    phone: str
    user_id: str = ''
    course_id: str

    @validator("name")
    def validate_name(cls, name: str) -> str:
        if not re.search('^(?P<name>[а-яА-Я]{2,})($|)', name):
            raise ValueError("Incorrect name")
        return name.capitalize()

    @validator("surname")
    def validate_surname(cls, surname: str) -> str:
        if not re.search('^(?P<surname>[а-яА-Я]{2,})($|)', surname):
            raise ValueError("Incorrect surname")
        return surname.capitalize()

    @validator("phone")
    def validate_phone(cls, phone: str) -> str:
        founds_phone = re.search('^(?P<phone>(?P<pre>\+7|8|)(?P<num>\d{10}))($|)', phone, re.M|re.S)
        if founds_phone is None:
            raise ValueError("Incorrect phone")
        phone = f"8{founds_phone.group('num')}"
        return phone

    @validator("email")
    def validate_email(cls, email: str) -> str:
        if not re.search('^(?P<email>(?P<mnaim>\S+)\@(?P<host>\S+)\.(?P<domain>\w+))($|)', email):
            raise ValueError("Incorrect email")
        return email.lower()


class RegistrationForm(FlaskForm): 
    """Вадиции данных пользователя из POST формы"""

    name = StringField()
    surname = StringField()
    email = StringField()
    phone = StringField()
    course_id = StringField()


@app.route("/api/v1/register_post", methods=["POST"])
def isp_registration():
    """Метод для регистрации пользователя в isp post-запросом"""
    
    logger.debug(f'request.form = {request.form}')
    if request.data:
        try:
            new_user = User.parse_raw(request.data)
            processing_requests = ApiRequest(new_user)
            processing_requests.api_requests()
            logger.info(f'Данные пользователя: {new_user}')
            return new_user.json(ensure_ascii=False, indent=2)
        except ValidationError as e:
            logger.debug(f'Ошибки валидации JSON: {e.json()}')
            return e.json(), 400

@app.route("/api/register_form", methods=['POST', "GET"])
def ispring_registration_from_form():
    """example http://127.0.0.1:5000/api/register_form?course_id=cc624f10-cf4b-11eb-82aa-4e8520552baf"""

    logger.debug(f'request.form = {request.form}')
    if request.method == 'POST':
        form = RegistrationForm()
        if form.validate_on_submit():
            new_user = User(name=form.name.data, surname=form.surname.data, email=form.email.data,
                            phone=form.phone.data, course_id=form.course_id.data)
            req = ApiRequest(new_user)
            req.api_requests()
            logger.debug(f'Данные пользователя: {new_user}')
            return make_response(new_user.json(ensure_ascii=False, indent=2), status_code=200)

        logger.debug(f'Ошибки валидации формы: {form.errors}')
        return jsonify(form.errors), 400
    return render_template('test_register_form1.html', course_id=request.args.get('course_id'))



if __name__ == '__main__':
    app.config["WTF_CSRF_ENABLED"] = True 
    app.run(host='0.0.0.0', debug=False, port=os.getenv("HOST")) 

Стандартное место расположения шаблонов страниц - /templates. Base.html для наследования общего стиля.

<!DOCTYPE html>
<html lang="ru">
<head>
    <meta charset="UTF-8">
    <link type="text/css" href="{{ url_for('static', filename='css/styles.css') }}" rel="stylesheet"/>
    {% block title -%}
        {% if title %}
            <title>Форма регистрации - {{ title }}</title>
        {% else %}
            <title>Форма регистрации</title>
        {% endif %}
    {% endblock %}
</head>
<body>
{% block content -%}
{% block mainmenu -%}
<ul class="mainmenu">
</ul>
{% endblock mainmenu -%}
<div class="clear"></div>
<div class="content">
    {% if title %}
        <h1>Форма регистрации - {{ title }}</h1>
    {% else %}
        <h1>Форма регистрации </h1>
    {% endif %}
{% endblock content -%}
</div>
</body>
</html>

Форму регистрации сделаем например в таком виде

<!DOCTYPE html>
<html lang="ru">
<head>
    <meta charset="UTF-8">
    <title> Форма регистрации</title>
</head>
<body>

<div>
    <h1>Регистрация нового пользователя</h1>
    <form action="http://127.0.0.1:5000/api/register_form" method="post">
        <p>
            <input id="course_id" type="text" name="course_id" hidden value={{ course_id }}>
        </p>
        <p>
            <label for="id_user">name</label>
            <input id="id_user" type="text" name="name">
        </p>
        <p>
            <label for="id_surname">surname</label>
            <input id="id_surname" type="text" name="surname">
        </p>
        <p>
            <label for="id_email">email</label>
            <input id="id_email" type="text" name="email">
        </p>
        <p>
            <label for="id_phone">phone</label>
            <input id="id_phone" type="text" name="phone">
        </p>
        <button>Регистрация</button>
    </form>
</div>

</body>
</html>

После чего мы можем например отправлять post-запросы к нашему сервису, таким же образом, curl-ом, через postman или руками заполняя формы на веб-странице.

Весь использованный код расположу здесь