Простая реализация REST-приложения на примере интернет-магазина с помощью Python-фреймворка Flask, Google Colab, утилиты для тунеллирования ngrok и базы данных PostgreSQL в облачной платформе Neon.
Слой представления: создаем HTML-страницы
В прошлой статье я описывала основные функциональные требования к интернет-магазину, а также приводила схему физической модели данных и пример спецификации OpenAPI, описывающей REST API серверного приложения. Приложение имеет классическую трехзвенную архитектуру:
- Слой представления выполнен в виде пользовательского интерфейса веб-сайта, т.е. совокупности связанных HTML-страниц;
- Слой бизнес-логики реализуется Flask-приложением, запускаемом в Goggle Colab и туннелированном с помощью утилиты ngrok, чтобы получить доступ к локальному хосту этой виртуальной машины через внешний URL-адрес;
- Слой доступа к данным – это база данных PostgreSQL, развернутая в облачной платформе Neon.
Сперва создадим клиентскую часть, т.е. HTML-страницы. Фреймворк Flask использует шаблоны HTML-страниц для визуализации форм ввода и вывода данных. Эти шаблоны страниц, наполняемые динамически, т.е. значениями переменных, например, строками из БД и пр., должны храниться в директории /templates. А статические данные, например, таблица стилей для HTML-файлов, чтобы они выглядели единообразно, и картинки хранятся в директории /static.
Для этого в Google Colab нужен следующий код. Сперва надо установить библиотеки и импортировать пакеты:
#############################новая ячейка в Google Colab########################################################### # Установка необходимых библиотек !pip install flask !pip install pyngrok !pip install flask-ngrok !pip install pydantic[email] # Импорт модулей и библиотек import traceback import threading from flask_ngrok import run_with_ngrok from flask import Flask, session, jsonify, request, render_template, Response, make_response, redirect from pydantic import BaseModel, EmailStr, ValidationError from enum import Enum from typing import Union, Optional, List import os import psycopg2 import threading import tkinter as tk import jwt from datetime import date, datetime, timedelta
Затем создать директории для хранения статических файлов и динамических шаблонов HTML-страниц:
#############################новая ячейка в Google Colab########################################################### # создаем директорию с шаблонами !mkdir templates !mkdir static
Создадим шаблоны HTML-страниц в виде файлов с расширением .html. Например, для создания страницы аутентификации, т.е. входа в систему, используется следующий код:
#############################новая ячейка в Google Colab###########################################################
# создаем html-страницу входа в систему
%%writefile templates/login.html
<!DOCTYPE html>
<html>
<head>
<title>Войти в личный кабинет</title>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/4.5.2/css/bootstrap.min.css">
<link rel="stylesheet" type="text/css" href="{{ url_for('static', filename='styles.css') }}">
<script src="https://ajax.googleapis.com/ajax/libs/jquery/3.5.1/jquery.min.js"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/popper.js/1.16.0/umd/popper.min.js"></script>
<script src="https://maxcdn.bootstrapcdn.com/bootstrap/4.5.2/js/bootstrap.min.js"></script>
</head>
<body>
<div class="container mt-3">
<h2>Войти в личный кабинет</h2>
<table>
<tr>
<td>
<a href='/login'>Войти в ЛК</a></td>
</td>
<td>
<a href='/product'>Каталог товаров</a>
</td>
<td>
<a href='/provider'>Список поставщиков</a>
</td>
<td>
<a href='/order'>Заказы</a>
</td>
</tr>
</table>
<form method="POST" action="/login">
<div class="form-group">
<label for="login">Имя:</label>
<input type="email" class="form-control" id="login" placeholder="логин (емейл)" name="login" required>
</div>
<div class="form-group">
<label for="password">Пароль:</label>
<input type="password" class="form-control" id="password" placeholder="пароль" name="password" required>
</div>
<div class="form-group">
<label for="role">Роль:</label>
<select id="role" name="role">
<option value="customer">Покупатель</option>
<option value="manager">Менеджер интернет-магазина</option>
<option value="operator">Оператор склада</option>
</select>
</div>
<button type="submit" class="btn btn-primary">ВОЙТИ</button>
</form>
</div>
<img src="{{ url_for('static', filename='backpic.png') }}" style="display: block; margin-top: 20px;">
</body>
</html> 
Для страницы просмотра всех товаров нужно выводить их полный список. Для этого следует встроить в код HTML-шаблона переменные, т.е. атрибуты класса Product, заключив их в двойные фигурные скобки:
#############################новая ячейка в Google Colab###########################################################
# создаем html-страницу просмотра каталога товаров
%%writefile templates/product.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/4.5.2/css/bootstrap.min.css">
<link rel="stylesheet" type="text/css" href="{{ url_for('static', filename='styles.css') }}">
<title>Каталог товаров интернет-магазина</title>
</head>
<body>
<h1>Каталог товаров интернет-магазина:</h1>
<table>
<tr>
<td>
<a href='/login'>Войти в ЛК</a></td>
</td>
<td>
<a href='/product'>Каталог товаров</a>
</td>
<td>
<a href='/new_product'>Добавить товар</a>
</td>
<td>
<a href='/provider'>Список поставщиков</a>
</td>
<td>
<a href='/order'>Заказы</a>
</td>
</tr>
</table>
<table>
<thead>
<tr>
<th>Товар </th>
<th>Поставщик</th>
<th>Цена</th>
<th>Количество</th>
</tr>
</thead>
<tbody>
{% for product in products %}
<tr>
<td>{{ product.name }}</td>
<td>{{ product.provider }}</td>
<td>{{ product.price }}</td>
<td>{{ product.quantity }}</td>
<td>
<form method="POST" action="/product/{{ product.id }}">
<button value="{{ product.id }}" type="submit">Изменить</button>
</form>
</td>
</td>
</tr>
{% endfor %}
</tbody>
</table>
</body>
</html> 
Основы архитектуры и интеграции информационных систем
Код курса
OAIS
Ближайшая дата курса
20 января, 2025
Продолжительность
16 ак.часов
Стоимость обучения
36 000 руб.
Чтобы изменить конкретный товар методом PUT или удалить его методом DELETE, как запланировано в спецификации OpenAPI, приходится переопределять HTTP-метод, поскольку согласно стандарту HTML, к кнопкам с надписью «Изменить» может быть привязан только метод GET или POST. Эту замену приходится делать на странице изменения или удаления конкретного товара, и даже добавлять JavaScript для вызова соответствующих методов HTTP:
#############################новая ячейка в Google Colab###########################################################
# создаем html-страницу изменения товара
%%writefile templates/update_product.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<link rel="stylesheet" type="text/css" href="{{ url_for('static', filename='styles.css') }}">
<title>Изменение поставщика</title>
<script src="https://ajax.googleapis.com/ajax/libs/jquery/2.2.0/jquery.min.js"></script>
<meta http-equiv="Cache-Control" content="no-cache, no-store, must-revalidate">
<meta http-equiv="Pragma" content="no-cache">
<meta http-equiv="Expires" content="0">
</head>
<body>
<h1>Изменение товара</h1>
<table>
<tr>
<td>
<form id="form" method="POST" action="/product/{{ product.id }}" onSubmit="myFunctionName(); return false">
<input type="hidden" name="_method" value="PUT">
<label for="name">Название:</label>
<input type="text" class="form-control" id="name" placeholder="название товара" name="name" value="{{ product.name }}"><br><br>
<label for="provider">Поставщик:</label>
<input type="number" class="form-control" id="provider" placeholder="поставщик товара" name="provider" value="{{ product.provider }}"><br><br>
<label for="price">Цена:</label>
<input type="number" step="0.01" class="form-control" id="price" placeholder="цена товара" name="price" value="{{ product.price }}"><br><br>
<label for="quantity">Количество:</label>
<input type="number" class="form-control" id="quantity" placeholder="количество товара" name="quantity" value="{{ product.quantity }}"><br><br>
<button type='button' id='change-button'>Изменить</button>
</form>
</td>
<td>
<button id='delete-button'>Удалить</button>
</td>
</tr>
</table>
<img src="{{ url_for('static', filename='backpic.png') }}" style="display: block; margin-top: 20px;">
<script>
$(document).ready(function(){
$("#delete-button").on("click", () => {
$.ajax({
url: "/product/{{ product.id }}",
method: "DELETE",
headers: {
"X-my": "aaaaaaaaaaaaaaaaaaaaaaaa"
},
success : function () {
window.location.href = "/product";
}
});
});
$("#change-button").on("click", () => {
$.ajax({
url: "/product/{{ product.id }}",
method: "PUT",
data:$("#form").serialize(),
headers: {
"X-my": "aaaaaaaaaaaaaaaaaaaaaaaa"
},
success : function () {
window.location.href = "/product";
},
error: function (jqXHR, exception) {
if (jqXHR.status === 0) {
alert('Not connect. Verify Network.');
} else if (jqXHR.status == 404) {
alert('Requested page not found (404).');
} else if (jqXHR.status == 500) {
alert('Internal Server Error (500).');
} else if (exception === 'parsererror') {
alert('Requested JSON parse failed.');
} else if (exception === 'timeout') {
alert('Time out error.');
} else if (exception === 'abort') {
alert('Ajax request aborted.');
} else {
alert('Uncaught Error. ' + jqXHR.responseText);
}}
});
});
});
</script>
</body>
</html> Пример страницы, отрисованной по этому шаблону для ресурса, доступного по маршруту /product/83, где 83 – это идентификатор конкретного товара.
Все HTML-страницы полностью представлены в моем Github-репозитории. Закончив с представлениями, перейдем к модели, т.е. классам.
Основы архитектуры и интеграции информационных систем
Код курса
OAIS
Ближайшая дата курса
20 января, 2025
Продолжительность
16 ак.часов
Стоимость обучения
36 000 руб.
Классы модели данных и функции работы с БД
Пока в этом приложении классы модели данных я использую только для вывода списка товаров, поставщиков и заказов в соответствующих HTML-страницах. Чтобы отобразить массив таких объектов, считанных из базы данных, следует определить их классы, а затем агрегировать. Сперва построим модель в UML-диаграммы классов.
Эта диаграмма создана следующим PlantUML-скриптом:
@startuml
class Provider {
- id: int
- name: Optional[str]
- phone: Optional[str]
- email: Optional[EmailStr]
- address: Optional[str]
}
class Product {
- id: int
- name: Optional[str]
- provider: Optional[Provider]
- price: Optional[float]
- quantity: Optional[int]
}
class Order {
- id: int
- customer: str
- state: str
- delivery: Optional[str]
- sum: Optional[float]
- date: Optional[date]
}
class Welcome {
- products: List[Product]
- providers: List[Provider]
- orders: List[Order]
+ init(apps: List[Product], providers: List[Provider], orders: List[Order]) -> None
}
Provider --> Product
Welcome o--> Product
Welcome o--> Provider
Welcome o--> Order
@enduml Python-код по этой диаграмме:
#############################новая ячейка в Google Colab###########################################################
# Описание моделей данных
class Provider(BaseModel):
id: int
name: Optional[str] = None
phone: Optional[str] = None
email: Optional[EmailStr] = None
address: Optional[str] = None
class Product(BaseModel):
id: int
name: Optional[str] = None
provider: Optional[Provider] = None
price: Optional[float] = None
quantity: Optional[int] = None
class Order(BaseModel):
id: int
customer: str
state: str
delivery: Optional[str] = None
sum: Optional[float]
date: Optional[date]
class Welcome:
products: List[Product]
def init(self, apps: List[Product]) -> None:
self.products = products
providers: List[Provider]
def init(self, apps: List[Provider]) -> None:
self.providers = providers
orders: List[Order]
def init(self, apps: List[Order]) -> None:
self.orders = orders DDD, ООП и UML для аналитика
Код курса
BUML
Ближайшая дата курса
9 декабря, 2024
Продолжительность
16 ак.часов
Стоимость обучения
36 000 руб.
На этом работа с данными не закончена: необходимо определить функции работы с базой данных, которые непосредственно выполняют SQL-запросы. За это отвечает следующая ячейка Colab с Python-кодом:
#############################новая ячейка в Google Colab###########################################################
# Строка подключения к БД
global connection_string
connection_string='postgres://my_database' #сюда вставлять СВОЮ строку подключения к СВОЕЙ БД
conn = psycopg2.connect(connection_string)
# Функция получения всех записей из таблицы товаров для отображения каталога
def read_product_db(conn):
cursor = conn.cursor()
try:
cursor.execute("SELECT product.id, product.name, provider.name, product.price, product.quantity FROM product JOIN provider on provider.id=product.provider ORDER BY product.id DESC")
results = cursor.fetchall()
products = []
for row in results:
Product = {"id":row[0], "name": row[1], "provider": row[2], "price": row[3], "quantity": row[4]}
products.append(Product)
return products
except:
traceback.print_exc()
return 'Ошибка при получении товаров из базы данных'
finally:
cursor.close()
# Функция получения всех записей из таблицы заказов для отображения списка
def read_order_db(conn):
cursor = conn.cursor()
try:
cursor.execute("SELECT orders.id, customer.name, order_states.name, delivery.address, orders.sum, orders.date FROM orders JOIN customer on customer.id=orders.customer JOIN order_states on order_states.id=orders.state JOIN delivery on delivery.id=orders.delivery ORDER BY orders.id DESC")
results = cursor.fetchall()
orders = []
for row in results:
Order = {"id":row[0], "customer": row[1], "state": row[2], "delivery": row[3], "sum": row[4], "date": row[5]}
orders.append(Order)
return orders
except:
traceback.print_exc()
return 'Ошибка при получении заказов из базы данных'
finally:
cursor.close()
# Функция получения всех записей из таблицы поставщиков для отображения списка
def read_provider_db(conn):
cursor = conn.cursor()
try:
cursor.execute("SELECT provider.id, provider.name, provider.phone, provider.email, provider.address FROM provider ORDER BY provider.id DESC")
results = cursor.fetchall()
providers = []
for row in results:
Provider = {"id":row[0], "name": row[1], "phone": row[2], "email": row[3], "address": row[4]}
providers.append(Provider)
return providers
except:
traceback.print_exc()
return 'Ошибка при получении поставщиков из базы данных'
finally:
cursor.close()
# Функция поиска пользователя в базе данных
def find_user_in_db(conn, login, password, role):
cursor = conn.cursor()
try:
cursor.execute("SELECT * FROM users WHERE login = %s AND password = %s AND role=%s", (login, password, role))
result = cursor.fetchone()
if result:
user = {"id": result[0], "login": result[1], "password": result[2], "role": result[3]}
return user
else:
return None
except:
traceback.print_exc()
return 'Ошибка при получении пользователя из базы данных'
finally:
cursor.close()
# Функция регистрации пользователя (добавления записи в таблицу users)
def add_user_to_db(conn, user_data):
cursor = conn.cursor()
try:
cursor.execute("SELECT MAX(id) FROM users")
id = cursor.fetchone()[0] + 1
cursor.execute("INSERT INTO users (id, login, password, role) VALUES (%s, %s, %s, %s)", (id, user_data['login'], user_data['password'], user_data['role']))
conn.commit()
user = {"id": id, "login": user_data['login'], "password": user_data['password'], "role": user_data['role']}
return user
except Exception as e:
traceback.print_exc()
conn.rollback()
return None
finally:
cursor.close()
# Функция генерации JWT-токена
def generate_jwt_token(user_id, role, expiration_minutes):
if role=='manager':
# Задаем время жизни токена
expiration_time = datetime.utcnow() + timedelta(minutes=expiration_minutes)
# Создаем словарь с данными пользователя, которые будут включены в токен
payload = {
"user_id": user_id,
"role": role,
"exp": expiration_time
}
# Генерируем JWT токен с помощью секретного ключа
token = jwt.encode(payload, secret_key, algorithm="HS256")
return token
# Функция поиска токена в базе данных
def find_token_in_db(conn, sysuser):
cursor = conn.cursor()
try:
cursor.execute("SELECT token FROM jwts WHERE sysuser = %s::integer", (sysuser,))
token = cursor.fetchone()
print('find_token_in_db ', token)
return token
except:
traceback.print_exc()
return 'Ошибка при получении токена из базы данных'
finally:
cursor.close()
# Функция извлечения данных пользователя из токена
def extract_user_data_from_token(token):
try:
# Декодируем токен для получения данных пользователя
decoded_token = jwt.decode(token, secret_key, algorithms=["HS256"])
user_id = decoded_token["user_id"]
sysuser = user_id["id"]
if find_token_in_db(conn, sysuser):
return decoded_token
except jwt.exceptions.ExpiredSignatureError:
print("Время жизни токена истекло, надо снова войти в систему")
response = make_response(render_template('login.html'), 401)
return response
except jwt.exceptions.DecodeError:
return None
# Функция записи JWT-токена в таблицу jwts
def add_JWT_to_db(conn, JWTtime, token, sysuser):
cursor = conn.cursor()
try:
cursor.execute("SELECT MAX(id) FROM jwts")
id = cursor.fetchone()[0] + 1
cursor.execute("INSERT INTO jwts (id, published, token, sysuser) VALUES (%s, %s, %s, %s)", (id, JWTtime, token, sysuser))
conn.commit()
return token
except Exception as e:
traceback.print_exc()
conn.rollback()
return None
finally:
cursor.close()
# Функция добавления товара в таблицу product
def add_product_to_db(conn, name, provider, price, quantity):
cursor = conn.cursor()
try:
cursor.execute("SELECT MAX(id) FROM product")
id = cursor.fetchone()[0] + 1
cursor.execute("INSERT INTO product (id, name, provider, price, quantity) VALUES (%s::integer, %s, %s::integer, %s::double precision, %s::integer)", (id, name, provider, price, quantity))
conn.commit()
product = {"id":id, "name": name, "provider": provider, "price": price, "quantity": quantity}
return product
except Exception as e:
traceback.print_exc()
conn.rollback()
return None
finally:
cursor.close()
# Функция добавления поставщика в таблицу provider
def add_provider_to_db(conn, name, phone, email, address):
cursor = conn.cursor()
try:
cursor.execute("SELECT MAX(id) FROM provider")
id = cursor.fetchone()[0] + 1
conn.commit()
cursor.execute("INSERT INTO provider (id, name, phone, email, address) VALUES (%s::integer, %s, %s, %s, %s)", (id, name, phone, email, address))
conn.commit()
provider = {"id":id, "name": name, "phone": phone, "email": email, "address": address}
return provider
except Exception as e:
traceback.print_exc()
conn.rollback()
return None
finally:
cursor.close()
# Функция поиска поставщика в базе данных
def get_provider_from_db(conn, provider_id):
cursor = conn.cursor()
try:
cursor.execute("SELECT * FROM provider WHERE id = (%s::integer)", (provider_id,))
provider_data = cursor.fetchone()
if provider_data:
provider = {
"id": provider_data[0],
"name": provider_data[1],
"phone": provider_data[2],
"email": provider_data[3],
"address": provider_data[4]
}
return provider
else:
return None
except Exception as e:
traceback.print_exc()
return None
finally:
cursor.close()
# Функция обновления поставщика в базе данных
def update_provider_in_db(conn, id, name, phone, email, address):
cursor = conn.cursor()
try:
cursor.execute("UPDATE provider SET name = %s, phone = %s, email = %s, address = %s WHERE id = %s::integer", (name, phone, email, address, id))
conn.commit()
if cursor.rowcount > 0:
provider = {
"id": id,
"name": name,
"phone": phone,
"email": email,
"address": address
}
return provider
else:
return None
except Exception as e:
traceback.print_exc()
conn.rollback()
return None
finally:
cursor.close()
# Функция удаления поставщика в базе данных
def delete_provider_from_db(conn, provider_id):
cursor = conn.cursor()
try:
cursor.execute("DELETE FROM provider WHERE id = (%s::integer)", (provider_id,))
conn.commit()
if cursor.rowcount > 0:
return True
else:
return False
except Exception as e:
traceback.print_exc()
conn.rollback()
return False
finally:
cursor.close()
# Функция поиска товара в базе данных
def get_product_from_db(conn, product_id):
cursor = conn.cursor()
try:
cursor.execute("SELECT * FROM product WHERE id = (%s::integer)", (product_id,))
product_data = cursor.fetchone()
if product_data:
product = {
"id": product_data[0],
"name": product_data[1],
"provider": product_data[2],
"price": product_data[3],
"quantity": product_data[4]
}
print('get_product_from_db', product)
return product
else:
return None
except Exception as e:
traceback.print_exc()
return None
finally:
cursor.close()
# Функция обновления товара в базе данных
def update_product_in_db(conn, id, name, provider, price, quantity):
cursor = conn.cursor()
try:
cursor.execute("UPDATE product SET name = %s, provider = %s::integer, price = %s::double precision, quantity = %s::integer WHERE id = %s::integer", (name, provider, price, quantity, id))
conn.commit()
if cursor.rowcount > 0:
product = {
"id": id,
"name": name,
"provider": provider,
"price": price,
"quantity": quantity
}
print("update_product_in_db", product)
return product
else:
return None
except Exception as e:
traceback.print_exc()
conn.rollback()
return None
finally:
cursor.close()
# Функция удаления товара в базе данных
def delete_product_from_db(conn, product_id):
cursor = conn.cursor()
try:
cursor.execute("DELETE FROM product WHERE id = (%s::integer)", (product_id,))
conn.commit()
if cursor.rowcount > 0:
return True
else:
return False
except Exception as e:
traceback.print_exc()
conn.rollback()
return False
finally:
cursor.close()
# Функция поиска заказа в базе данных
def get_order_from_db(conn,order_id):
cursor = conn.cursor()
try:
cursor.execute("SELECT orders.id, customer.name, order_states.name, delivery.address, orders.sum, orders.date FROM orders JOIN customer on customer.id=orders.customer JOIN order_states on order_states.id=orders.state JOIN delivery on delivery.id=orders.delivery WHERE orders.id = (%s::integer)", (order_id,))
order_data = cursor.fetchone()
if order_data:
order = {
"id": order_data[0],
"customer": order_data[1],
"state": order_data[2],
"delivery":order_data[3],
"sum": order_data[4],
"date": order_data[5]
}
return order
else:
return None
except Exception as e:
traceback.print_exc()
return None
finally:
cursor.close()
# Функция обновления заказа в базе данных
def update_order_in_db(conn, id, state):
cursor = conn.cursor()
try:
cursor.execute("UPDATE orders SET state = %s::integer WHERE id = (%s::integer)", (state, id))
conn.commit()
cursor.execute("SELECT * FROM orders WHERE id = (%s::integer)", (id,))
order_data = cursor.fetchone()
if order_data:
order = {
"id": order_data[0],
"customer": order_data[1],
"state": order_data[2],
"delivery":order_data[3],
"sum": order_data[4],
"date": order_data[5]
}
print("update_order_in_db:", order)
return order
else:
return None
except Exception as e:
traceback.print_exc()
conn.rollback()
return None
finally:
cursor.close() Далее перейдем к самому интересному, т.е. определению маршрутов веб-прилоения и HTTP-методов работы с ними.
Основы архитектуры и интеграции информационных систем
Код курса
OAIS
Ближайшая дата курса
20 января, 2025
Продолжительность
16 ак.часов
Стоимость обучения
36 000 руб.
Создание Flask-приложения
Конечная точка REST-приложения во фреймворке Flask помечается декорированным методом @route, в параметрах которого указывается URL-адрес маршрута и методы доступа к нему. Например, следующий код показывает, что к маршруту /product можно обратиться методами GET и POST, при этом далее определяется функция get_product().
# Обработчик запросов на получение списка товаров:
@app.route('/product', methods=['GET', 'POST'])
def get_product():
# Подключение к PostgreSQL
conn = psycopg2.connect(connection_string)
try:
if request.method == 'GET':
products = read_product_db(conn)
return render_template('product.html', products=products)
else:
if request.content_type == 'application/json':
new_product = request.json
elif request.form:
new_product = request.form
name = new_product.get('name')
provider = new_product.get('provider')
price = new_product.get('price')
quantity = new_product.get('quantity')
# Добавление нового товара в базу данных
product = add_product_to_db(conn, name, provider, price, quantity)
return redirect('/product')
except Exception as e:
traceback.print_exc()
return f'Ошибка при получении товаров из базы данных: {e}'
finally:
conn.close() В блоке try выполняются действия в зависимости от типа запроса. Если это GET-запрос, то выполняется функция read_product_db для получения списка товаров из базы данных, а затем отображается шаблон ‘product.html’ с передачей списка товаров в него.
Метод GET используется для получения списка товаров из базы данных, а метод POST — для добавления нового товара, что доступно только аутентифицированному пользователю с ролью Менеджер. Как это реализовано с помощью JWT-токенов, зашитых в куки-файлы клиентских запросов, я писала здесь. При выполнении POST-запроса создается новый объект new_product с информацией о новом товаре из формы HTML-страницы или содержания JSON-документа, который может быть отправлен через Postman. Из объекта new_product извлекаются данные о товаре (название, поставщик, цена и количество), и используются в качестве параметров вызова функции add_product_to_db(). Эта функция непосредственно работает с БД и была описана ранее. После этого происходит перенаправление на страницу со списком товаров.
Функции обработки запроса заключены в блок try-except для отлова ошибок (исключений) без сбоя программы. При возникновении исключения выводится сообщение об ошибке, а затем закрывается соединение с базой данных в блоке finally.
Весь код Flask-приложения выглядит следующим образом:
#############################новая ячейка в Google Colab###########################################################
# Установка токена для авторизации в ngrok #@param {type:"string"}
auth_token = "my-token-for-ngrok" #сюда вставить СВОЙ ngrok-токен, полученный здесь https://dashboard.ngrok.com/
os.system(f"ngrok authtoken {auth_token}")
########################################################################################
# Задаем секретный ключ для подписи JWT
secret_key = "SecretKey4JWT" #здесь задать СВОЙ секретный ключ для формирования JWT-токена
# Задаем время жизни JWT-токена в минутах
exp_time=15
# Создание экземпляра Flask-приложения
app = Flask(__name__)
# Запуск приложения с использованием ngrok
run_with_ngrok(app)
# set the secret key для кодирования сессии:
app.secret_key = secret_key
# Обработчик запросов по корневому адресу
@app.route('/')
def hello():
return render_template('index.html')
# Обработчик запросов на получение списка товаров:
@app.route('/product', methods=['GET', 'POST'])
def get_product():
# Подключение к PostgreSQL
conn = psycopg2.connect(connection_string)
try:
if request.method == 'GET':
products = read_product_db(conn)
return render_template('product.html', products=products)
else:
if request.content_type == 'application/json':
new_product = request.json
elif request.form:
new_product = request.form
name = new_product.get('name')
provider = new_product.get('provider')
price = new_product.get('price')
quantity = new_product.get('quantity')
# Добавление нового товара в базу данных
product = add_product_to_db(conn, name, provider, price, quantity)
return redirect('/product')
except Exception as e:
traceback.print_exc()
return f'Ошибка при получении товаров из базы данных: {e}'
finally:
conn.close()
# Обработчик запросов на добавление нового товара:
@app.route('/new_product', methods=['POST', 'GET'])
def add_product():
# Получаем токен из заголовка HTTP-запроса
token = request.cookies.get('token')
# Подключение к PostgreSQL
conn = psycopg2.connect(connection_string)
try:
if token:
# Проверяем валидность токена и получаем данные пользователя
user_data = extract_user_data_from_token(token)
if user_data:
# Проверяем, что пользователь имеет роль "manager"
if user_data['role'] == 'manager':
if request.method == 'POST':
response = make_response(render_template('product.html'), 200)
return response
else:
return render_template('new_product.html')
else:
print("У вас нет доступа к этой странице")
response = make_response(render_template('login.html'), 403)
return response
else:
print("Неверный токен")
response = make_response(render_template('login.html'), 403)
return response
else:
print("Требуется авторизация")
response = make_response(render_template('login.html'), 401)
return response
#return redirect(url_for('login'))
except Exception as e:
traceback.print_exc()
return f'Ошибка при обработке запроса: {e}'
finally:
conn.close()
# Обработчик запросов на получение списка поставщиков:
@app.route('/provider', methods=['GET', 'POST'])
def get_provider():
# Подключение к PostgreSQL
conn = psycopg2.connect(connection_string)
try:
if request.method == 'GET':
providers = read_provider_db(conn)
return render_template('provider.html', providers=providers)
else:
if request.content_type == 'application/json':
new_provider = request.json
elif request.form:
new_provider = request.form
name = new_provider.get('name')
phone = new_provider.get('phone')
email = new_provider.get('email')
address = new_provider.get('address')
# Добавление нового товара в базу данных
provider = add_provider_to_db(conn, name, phone, email, address)
return redirect('/provider')
except Exception as e:
traceback.print_exc()
return f'Ошибка при получении поставщиков из базы данных: {e}'
finally:
conn.close()
# Обработчик запросов на добавление поставщика:
@app.route('/new_provider', methods=['POST', 'GET'])
def add_provider():
# Получаем токен из заголовка HTTP-запроса
token = request.cookies.get('token')
print('token ', token)
# Подключение к PostgreSQL
conn = psycopg2.connect(connection_string)
try:
if token:
# Проверяем валидность токена и получаем данные пользователя
user_data = extract_user_data_from_token(token)
if user_data:
# Проверяем, что пользователь имеет роль "manager"
if user_data['role'] == 'manager':
if request.method == 'POST':
response = make_response(render_template('provider.html'), 200)
return response
else:
return render_template('new_provider.html')
else:
print("У вас нет доступа к этой странице")
response = make_response(render_template('login.html'), 403)
return response
else:
print("Неверный токен")
response = make_response(render_template('login.html'), 403)
return response
else:
print("Требуется авторизация")
response = make_response(render_template('login.html'), 401)
return response
except Exception as e:
traceback.print_exc()
return f'Ошибка при обработке запроса: {e}'
finally:
conn.close()
# Обработчик входа в систему существующего пользователя:
@app.route('/login', methods=['POST', 'GET'])
def login():
# Подключение к PostgreSQL
conn = psycopg2.connect(connection_string)
try:
if request.method == 'POST':
if request.content_type == 'application/json':
user_data = request.json
elif request.form:
user_data = request.form
login = user_data.get('login')
password = user_data.get('password')
role = user_data.get('role')
# Поиск пользователя в базе данных
user = find_user_in_db(conn, login, password, role)
if user is not None:
# Генерация JWT-токена
if role =='manager':
token = generate_jwt_token(user, role, exp_time)
sysuser=user['id']
print('token', token)
# Запись JWT-токена в таблицу jwts
add_JWT_to_db(conn, datetime.utcnow(), token, sysuser)
# Добавление токена в заголовок запроса
response = make_response(render_template('index.html'), 200)
response.set_cookie('token', token)
return response
else:
return render_template('index.html')
else:
response = make_response(render_template('registration.html'), 401)
return response
# return redirect(url_for('registration'))
# return redirect('/registration')
else:
response = make_response(render_template('login.html'), 200)
return response
#return redirect('/login')
except Exception as e:
traceback.print_exc()
if request.method == 'POST':
return jsonify({"error": f'Ошибка аутентификации пользователя в базу данных: {e}'})
else:
return jsonify({"error": f'Сервис аутентификации недоступен, попробуйте позже: {e}'})
finally:
conn.close()
# Обработчик регистрации нового пользователя:
@app.route('/registration', methods=['POST', 'GET'])
def registration():
# Подключение к PostgreSQL
conn = psycopg2.connect(connection_string)
try:
if request.method == 'POST':
if request.content_type == 'application/json':
user_data = request.json
elif request.form:
user_data = request.form
login = user_data.get('login')
password = user_data.get('password')
role = user_data.get('role')
# Регистрация пользователя в базе данных
added_user = add_user_to_db(conn, user_data)
response = make_response(render_template('login.html'), 201)
return response
# return redirect('/login')
else:
response = make_response(render_template('registration.html'), 200)
return response
#return redirect('/registration')
except Exception as e:
traceback.print_exc()
if request.method == 'POST':
return jsonify({"error": f'Ошибка при добавлении пользователя в базу данных: {e}'})
else:
return jsonify({"error": f'Сервис регистрации недоступен, попробуйте позже: {e}'})
finally:
conn.close()
# Обработчик изменения поставщика по id:
@app.route('/provider/<int:id>', methods=['GET', 'POST', 'PUT', 'DELETE'])
def provider_ops(id):
# Получаем токен из заголовка HTTP-запроса
token = request.cookies.get('token')
print('token ', token)
# Подключение к PostgreSQL
conn = psycopg2.connect(connection_string)
try:
if token:
# Проверяем валидность токена и получаем данные пользователя
user_data = extract_user_data_from_token(token)
if user_data:
# Проверяем, что пользователь имеет роль "manager"
if user_data['role'] == 'manager':
provider = get_provider_from_db(conn, id)
if not provider:
return f'Поставщик с ID {id} не найден'
else:
if request.method == 'PUT':
if request.content_type == 'application/json':
updated_provider = request.json
elif request.form:
updated_provider = request.form
else:
updated_provider = request.args
if updated_provider.get('name'):
name = updated_provider.get('name')
else:
name = provider['name']
if updated_provider.get('phone'):
phone = updated_provider.get('phone')
else:
phone = provider['phone']
if updated_provider.get('email'):
email = updated_provider.get('email')
else:
email = provider['email']
if updated_provider.get('address'):
address = updated_provider.get('address')
else:
address = provider['address']
#Обновление данных поставщика в базе данных
if update_provider_in_db(conn, id, name, phone, email, address):
return ('', 200)
elif request.method == 'DELETE':
try:
delete_provider_from_db(conn, id)
response = make_response(render_template('provider.html'), 204)
return response
except Exception as e:
traceback.print_exc()
return f'Ошибка при удалении поставщика: {e}'
finally:
conn.close()
else:
provider = get_provider_from_db(conn, id)
if provider:
return render_template('update_provider.html', provider=provider)
else:
return f'Поставщик с ID {id} не найден'
else:
print("У вас нет доступа к этой странице")
response = make_response(render_template('login.html'), 403)
return response
else:
print("Неверный токен")
response = make_response(render_template('login.html'), 403)
return response
else:
print("Требуется авторизация")
response = make_response(render_template('login.html'), 401)
return response
except Exception as e:
traceback.print_exc()
return f'Ошибка при изменении поставщика: {e}'
finally:
conn.close()
# Обработчик изменения товара по id:
@app.route('/product/<int:id>', methods=['GET', 'POST', 'PUT', 'PATCH', 'DELETE'])
def product_ops(id):
# Получаем токен из заголовка HTTP-запроса
token = request.cookies.get('token')
print('token ', token)
# Подключение к PostgreSQL
conn = psycopg2.connect(connection_string)
try:
if token:
# Проверяем валидность токена и получаем данные пользователя
user_data = extract_user_data_from_token(token)
if user_data:
# Проверяем, что пользователь имеет роль "manager"
if user_data['role'] == 'manager':
product = get_product_from_db(conn, id)
if not product:
return f'Товар с ID {id} не найден'
else:
if request.method == 'PUT' or request.method == 'PATCH':
if request.content_type == 'application/json':
updated_product = request.json
elif request.form:
updated_product = request.form
else:
updated_product = request.args
if updated_product.get('name'):
name = updated_product.get('name')
else:
name = product['name']
if updated_product.get('provider'):
provider = updated_product.get('provider')
else:
provider = product['provider']
if updated_product.get('price'):
price = updated_product.get('price')
else:
price = product['price']
if updated_product.get('quantity'):
quantity = updated_product.get('quantity')
else:
quantity = product['quantity']
#Обновление данных поставщика в базе данных
update_product_in_db(conn, id, name, provider, price, quantity)
return ('', 200)
elif request.method == 'DELETE':
try:
delete_product_from_db(conn, id)
response = make_response(render_template('product.html'), 204)
return response
except Exception as e:
traceback.print_exc()
return f'Ошибка при удалении товара: {e}'
finally:
conn.close()
else:
product = get_product_from_db(conn, id)
if product:
return render_template('update_product.html', product=product)
else:
return f'Товар с ID {id} не найден'
else:
print("У вас нет доступа к этой странице")
response = make_response(render_template('login.html'), 403)
return response
else:
print("Неверный токен")
response = make_response(render_template('login.html'), 401)
return response
else:
print("Требуется авторизация")
response = make_response(render_template('login.html'), 401)
return response
except Exception as e:
traceback.print_exc()
return f'Ошибка при изменении поставщика: {e}'
finally:
conn.close()
# Обработчик запросов на получение списка заказов:
@app.route('/order', methods=['GET'])
def get_order():
# Получаем токен из заголовка HTTP-запроса
token = request.cookies.get('token')
print('token ', token)
# Подключение к PostgreSQL
conn = psycopg2.connect(connection_string)
try:
if token:
# Проверяем валидность токена и получаем данные пользователя
user_data = extract_user_data_from_token(token)
if user_data:
# Проверяем, что пользователь имеет роль "manager"
if user_data['role'] == 'manager':
orders = read_order_db(conn)
return render_template('order.html', orders=orders)
else:
print("Вы не менеджер, просмотр запрещен")
response = make_response(render_template('login.html'), 401)
return response
else:
print("Надо войти как менеджер")
response = make_response(render_template('login.html'), 403)
return response
else:
print("Надо войти в систему")
response = make_response(render_template('login.html'), 403)
return response
except Exception as e:
traceback.print_exc()
return f'Ошибка при получении заказов из базы данных: {e}'
finally:
conn.close()
# Обработчик изменения заказа по id:
@app.route('/order/<int:id>', methods=['GET', 'POST', 'PUT'])
def order_ops(id):
# Получаем токен из заголовка HTTP-запроса
token = request.cookies.get('token')
print('token ', token)
# Подключение к PostgreSQL
conn = psycopg2.connect(connection_string)
try:
if token:
# Проверяем валидность токена и получаем данные пользователя
user_data = extract_user_data_from_token(token)
if user_data:
# Проверяем, что пользователь имеет роль "manager"
if user_data['role'] == 'manager':
order = get_order_from_db(conn, id)
print(order)
if not order:
return f'Заказ с ID {id} не найден'
else:
if request.method == 'PUT':
if request.content_type == 'application/json':
updated_order_state = request.json
elif request.form:
updated_order_state = request.form
else:
updated_order_state = request.args
if updated_order_state.get('state'):
state = updated_order_state.get('state')
else:
state = order['state']
#Обновление данных поставщика в базе данных
if update_order_in_db(conn, id, state):
return ('', 200)
else:
order = get_order_from_db(conn, id)
if order:
return render_template('update_order.html', order=order)
else:
return f'Заказ с ID {id} не найден'
else:
print("У вас нет доступа к этой странице")
response = make_response(render_template('login.html'), 403)
return response
else:
print("Неверный токен")
response = make_response(render_template('login.html'), 403)
return response
else:
print("Требуется авторизация")
response = make_response(render_template('login.html'), 401)
return response
except Exception as e:
traceback.print_exc()
return f'Ошибка при изменении поставщика: {e}'
finally:
conn.close()
# Запуск приложения
if __name__ == '__main__':
app.run() Чтобы не исчерпать лимит тунелей, последней ячейкой в Colabe закроем открытые тунель, убив ngrok-процессы. Эту ячейку следует запускать только после остановки предыдущей, с кодом серверного Flask-приложения:
#############################новая ячейка в Google Colab########################################################### # Kill tunnel !pkill -f ngrok
После запуска серверного Flask-приложения и манипуляции с ним в GUI через элементы HTML-страниц в области вывода Colab можно посмотреть HTTP-запросы и отладочную информацию.
Благодаря тунелированию с ngrok можно отправлять клиентские запросы к этому серверному приложению, получив ссылку внешнего URL-адреса. В данном случае это http://71a5-34-141-243-159.ngrok-free.app, однако, этот адрес меняется после каждого запуска. При открытии этого адреса в браузере отобразится приветственная HTML-страница с навигацией по сайту.
Полный набор артефактов этого веб-приложения доступен здесь, в моем Github-репозитории.
Справедливости ради стоит отметить, что данная реализация довольно примитивна. Во-первых, нет многих полезных функций, в частности, поиска и фильтрации данных. Также не реализована пагинация страниц, т.е. постраничный вывод. С точки зрения профессионального разработчика код выглядит не самым изящным, лаконичным и оптимальным, что я подчеркиваю в новой статье. Тем не менее, этот простой пример вполне отвечает своей исходной задаче и вполне подходит для демонстрации принципов работы REST API и архитектуры веб-приложений.
Основы архитектуры и интеграции информационных систем
Код курса
OAIS
Ближайшая дата курса
20 января, 2025
Продолжительность
16 ак.часов
Стоимость обучения
36 000 руб.
Подробнее узнать о об архитектуре и интеграции информационных систем вы сможете на моих курса в Школе прикладного бизнес-анализа на базе нашего лицензированного учебного центра обучения и повышения квалификации системных и бизнес-аналитиков в Москве:
