Python СRUD(т)!

Привет! По своей работе для тестирования API использую Postman. Базовые операции работы с записью стандартные — CRUD (create, read, update, delete). На результаты операционных запросов уже накладываются разнообразные тесты. При первой операции — create мы создаём запись, с тела ответа response парсим уникальный идентификатор (допустим id) созданной в БД записи, заносим её в переменную среды Postman

var jsonData = pm.response.json();
var NameId = jsonData.id;
pm.globals.set("NameId", NameId);

для того чтобы в дальнейшем вызывать и использовать для следующих операций: read, update, delete.

Но что если нет Postman, а надо отправить запрос на создание записи с ей последующим удалением?  Для реализации буду использовать Python 3.*.

Первый блок кода: отправка запроса на создание записи, распарсивание из json response уникального идентификатора (id) с его записью в переменную. Приведенный ниже код у вас не получится воспроизвести (если конечно эту статью не читают мои коллеги), так как для примера рассматриваю свой рабочий проект. Больше важна идея и способ её реализации.

#подключаем библиотеку для http запросов
import requests
#json.dump() - метод записывает объект Python в файл в формате JSON
import json
#подключаем библиотеку для того чтобы распарсить json
import jsonpath
url = 'http://mcm-demo.galaktika.local/server/api'
headers = {'Content-type': 'application/json', 'Accept': 'text/plain'}
#основное тело запроса создающая запись
data = {
    "action": "BankUpdate",
    "module": "nsi",
    "data": {
        "address": "Adress",
        "bic": "000011",
        "id": None,
        "name": "Name bank",
        "record_state_id": "000e554e-767d-11ea-82ef-830b1270af13"
    }
}
#отпавляем post запрос, не забывая про авторизацию
res_create = requests.post(url, auth=('user','password'),data=json.dumps(data,ensure_ascii=False).encode('utf-8'), headers=headers)
#выводим результат (response) нашего запроса, формат - JSON
print(res_create.text)
#превращаем полученные данные JSON в объект Python
response_json = json.loads(res_create.text)
#парсим, записываем в переменную id наше уникальное значение записи (id)
id = jsonpath.jsonpath(response_json,'$.item.id')
#выводим в консоль значение
print(id)

Скрин консоли:

Второй блок кода: удаление записи.

data_del = {
    "action": "BankDelete",
    "module": "nsi",
    "data": {
        "id": id[0]
    }
}
res_del = requests.post(url, auth=('user','password'),data=json.dumps(data_del,ensure_ascii=False).encode('utf-8'),headers=headers)
print(res_del.text)

В тело запроса (data_del) в значение id поместили спарсенное в блоке 2 значение. Вывели значения response. Получили {«id»:0}. В моём случае данный ответ означает то что запись удалена.

Скрин консоли:

Вроде как на этом можно было бы закончить статью, но нет. А давайте получим id записи непосредственно с базы данных? По прежнему используем Python и язык структурированных запросов (SQL).  Напишу функцию подключения к базе данных и фукнцию которая будет осуществлять непосредственно сам запрос.

#импортируем библиотеку для работы с postgresql
import psycopg2
#функция подключения к БД
def create_connection(db_name, db_user, db_password, db_host, db_port):
    connection = None
    try:
        connection = psycopg2.connect(
            database=db_name,
            user=db_user,
            password=db_password,
            host=db_host,
            port=db_port,
        )
        print("Подключение к базе данных осуществлено")
    except OperationalError as e:
        print(f"Что-то '{e}' пошло не так")
    return connection

connection = create_connection(
    "you_db_name", "you_db_user", "you_db_password", "you_ip", "you_port"
)
#функция выполнения запроса к БД
def execute_query(connection, query):
    cursor = connection.cursor()
    result = None
    try:
        cursor.execute(query)
        #вывод 1 строки
        # result = cursor.fetchmany(1)
        result = cursor.fetchall()
        return result
    except OperationalError as e:
        print(f"Что-то '{e}' пошло не так")

select_banks = "SELECT * FROM nsi.banks WHERE name='Name bank';"
banks = execute_query(connection, select_banks)
#выведем строку в консоль
for banks in banks:
    print ("id = {} ".format(banks[0]))
    print ("record_state_id = {}".format(banks[1]))
    print("name = {}".format(banks[2]))
    print("address = {}".format(banks[3]))
    print("bic = {}".format(banks[4]))
print("=======")
id = banks[0]
print(id)

Про синтаксис импортированной библиотеки psycopg2 можно почитать по ссылке. Немного объясню sql запрос:

SELECT * FROM nsi.banks WHERE name='Name bank';

Выбрать всё из схема.таблица(адрес где расположена таблица с данными) по критерию (где) строка заголовка таблицы name имеет значение = Name bank.

В первой части статьи мы создали запись указав параметру name значение = Name bank.
Так эта запись выглядит в базе данных:

Вывод консоли результата работы программы:

В последующем значение id мы можем использовать дальше, в зависимости от целей. Использовать для просмотра, редактирования записи.
Понятно, то о чём писал выше — не относится непосредственно к тестированию тела response API. Операционные действия, не более. Можно ли использовать Python для тестирования API? Можно конечно. Но есть Postman, SoapUI которые создавались для этого. Успешно используются и будут дальше внедряться в процесс тестирования.