Кратко: запросы к 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 или руками заполняя формы на веб-странице.

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