Этот пост предназначен в первую очередь для новичков в разработке, впервые столкнувшихся с необходимостью отправить post/get запросы к какому-нибудь API и проанализировать полученный в XML ответ. Постаралась собрать необходимы минимум в одном месте.
Если проверить сам метод на работоспособность, например с помощью 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.
В приходящем респонсе байтовая кодировка(необходимая для раскладывания 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': (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
[HEADING=1]3. Анализ XML. XPath или Bs4[/HEADING]
XPath - язык на котором необходимо будет доставать цепочки элементов. И на нем необходимо будет писать выражения для поиска элемента по параметрам ответа.
Примечательно что родители знают о своих детях, но дети не знают о своих родителях. Напрямую, задав путь до проверки атрибута элемента на пятом уровне вложенности, в случае если такой найден, забрать атрибут с его третьего уровня вложенности не получится.
Один из немногих мануалов по тому как с ним обращаться [URL]http://www.k-press.ru/cs/2001/2/XPath/XPath4.asp[/URL]
[IMG]https://habrastorage.org/r/w1560/getpro/habr/upload_files/24e/11a/8ae/24e11a8aefe610e7a9074f73d47d6f61.png[/IMG]
Например для того что бы достать из списка пользователей 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
[HEADING=1]4. Размещаем эти запросы на flask.[/HEADING]
Рекоммендуется использовать стандартную структуру страниц.
[IMG]https://habrastorage.org/r/w1560/getpro/habr/upload_files/077/13d/098/07713d0987eb1648cb4f80e6b8ccdaa2.png[/IMG]
В 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 или руками заполняя формы на веб-странице.
https://habr.com/ru/post/570852/
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': (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
[HEADING=1]3. Анализ XML. XPath или Bs4[/HEADING]
XPath - язык на котором необходимо будет доставать цепочки элементов. И на нем необходимо будет писать выражения для поиска элемента по параметрам ответа.
Примечательно что родители знают о своих детях, но дети не знают о своих родителях. Напрямую, задав путь до проверки атрибута элемента на пятом уровне вложенности, в случае если такой найден, забрать атрибут с его третьего уровня вложенности не получится.
Один из немногих мануалов по тому как с ним обращаться [URL]http://www.k-press.ru/cs/2001/2/XPath/XPath4.asp[/URL]
[IMG]https://habrastorage.org/r/w1560/getpro/habr/upload_files/24e/11a/8ae/24e11a8aefe610e7a9074f73d47d6f61.png[/IMG]
Например для того что бы достать из списка пользователей 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
[HEADING=1]4. Размещаем эти запросы на flask.[/HEADING]
Рекоммендуется использовать стандартную структуру страниц.
[IMG]https://habrastorage.org/r/w1560/getpro/habr/upload_files/077/13d/098/07713d0987eb1648cb4f80e6b8ccdaa2.png[/IMG]
В 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 или руками заполняя формы на веб-странице.
https://habr.com/ru/post/570852/