ParkSuMin/RIS
Archived
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

pre-commit changes

Browse Source
This commit is contained in:
Anton Kamalov
2024-12-08 12:17:19 +03:00
parent 2852ce1f96
commit 87e9029b09
15 changed files with 400 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
App/Auth/__init__.py
View File

@@ -1,23 +1,28 @@
from flask import Blueprint, render_template, redirect, url_for, request, session from flask import (Blueprint, redirect, render_template, request, session,
url_for)
from .auth_model import auth_model from .auth_model import auth_model
auth_bp = Blueprint('auth_bp', __name__, template_folder='templates') auth_bp = Blueprint("auth_bp", __name__, template_folder="templates")
@auth_bp.route('/', methods=['GET', 'POST'])
@auth_bp.route("/", methods=["GET", "POST"])
def auth(): def auth():
if request.method == 'GET': if request.method == "GET":
return render_template('auth.html') return render_template("auth.html")
else: else:
data = request.form.to_dict() data = request.form.to_dict()
auth_data = auth_model(data) auth_data = auth_model(data)
if auth_data.status: if auth_data.status:
session.update({ session.update(
'user_id': auth_data.result[0]['user_ID'], {
'login': auth_data.result[0]['login'], "user_id": auth_data.result[0]["user_ID"],
'access_user': data['access'], "login": auth_data.result[0]["login"],
'role': auth_data.result[0]['user_role'], "access_user": data["access"],
'permanent': True "role": auth_data.result[0]["user_role"],
}) "permanent": True,
return redirect(url_for('index')) }
)
return redirect(url_for("index"))
else: else:
return render_template('error.html', error_message=auth_data.error_message) return render_template("error.html", error_message=auth_data.error_message)

28
App/Auth/auth_model.py
View File

@@ -1,29 +1,29 @@
import os
from dataclasses import dataclass from dataclasses import dataclass
from Database.work import select_list
from Database.sql_provider import SQLProvider from Database.sql_provider import SQLProvider
from Database.work import select_list
from flask import current_app from flask import current_app
import os
sql_provider = SQLProvider(os.path.join(os.path.dirname(__file__), 'sql')) sql_provider = SQLProvider(os.path.join(os.path.dirname(__file__), "sql"))
@dataclass @dataclass
class InfoRespronse: class InfoRespronse:
result: tuple result: tuple
error_message: str error_message: str
status: bool status: bool
def auth_model(input_data) -> InfoRespronse:
db_config = current_app.config['db_config']
_sql = sql_provider.get('auth.sql', input_data) def auth_model(input_data) -> InfoRespronse:
db_config = current_app.config["db_config"]
_sql = sql_provider.get("auth.sql", input_data)
user = select_list(db_config, _sql) user = select_list(db_config, _sql)
if user is None: if user is None:
return InfoRespronse((), return InfoRespronse(
error_message = 'Ошибка при подключении к БД', (), error_message="Ошибка при подключении к БД", status=False
status=False) )
elif len(user) == 0: elif len(user) == 0:
return InfoRespronse((), return InfoRespronse((), error_message="Пользователь не найден", status=False)
error_message = 'Пользователь не найден', return InfoRespronse(user, error_message="", status=True)
status=False)
return InfoRespronse(user, error_message='', status=True)

17
App/Database/DBconnect.py
View File

@@ -1,5 +1,7 @@
import pymysql import pymysql
from pymysql.err import * from pymysql.err import *
class DBContextManager: class DBContextManager:
def __init__(self, db_config: dict): def __init__(self, db_config: dict):
self.db_config = db_config self.db_config = db_config
@@ -9,12 +11,12 @@ class DBContextManager:
def __enter__(self): def __enter__(self):
try: try:
self.connection = pymysql.connect( self.connection = pymysql.connect(
host=self.db_config['host'], host=self.db_config["host"],
port=self.db_config['port'], port=self.db_config["port"],
user=self.db_config['user'], user=self.db_config["user"],
password=self.db_config['password'], password=self.db_config["password"],
db=self.db_config['db'], db=self.db_config["db"],
charset=self.db_config['charset'] charset=self.db_config["charset"],
) )
self.cursor = self.connection.cursor() self.cursor = self.connection.cursor()
return self.cursor return self.cursor
@@ -25,11 +27,10 @@ class DBContextManager:
def __exit__(self, exc_type, exc_val, exc_tb): def __exit__(self, exc_type, exc_val, exc_tb):
if self.connection and self.cursor: if self.connection and self.cursor:
if exc_type: if exc_type:
print(exc_type, '\n', exc_val) print(exc_type, "\n", exc_val)
self.connection.rollback() self.connection.rollback()
else: else:
self.connection.commit() self.connection.commit()
self.cursor.close() self.cursor.close()
self.connection.close() self.connection.close()
return True return True

5
App/Database/sql_provider.py
View File

@@ -1,14 +1,15 @@
import os import os
from string import Template from string import Template
class SQLProvider: class SQLProvider:
def __init__(self, file_path): def __init__(self, file_path):
self.scripts = {} self.scripts = {}
for file in os.listdir(file_path): for file in os.listdir(file_path):
_sql = open(f'{file_path}/{file}').read() _sql = open(f"{file_path}/{file}").read()
self.scripts[file] = Template(_sql) self.scripts[file] = Template(_sql)
def get(self, name, params) -> dict: def get(self, name, params) -> dict:
if name not in self.scripts: if name not in self.scripts:
raise ValueError(f'SQL template {name} not found') raise ValueError(f"SQL template {name} not found")
return self.scripts[name].substitute(**params) return self.scripts[name].substitute(**params)

3
App/Database/work.py
View File

@@ -1,5 +1,6 @@
from .DBconnect import DBContextManager from .DBconnect import DBContextManager
def select_list(db_config, sql) -> list: def select_list(db_config, sql) -> list:
with DBContextManager(db_config) as cursor: with DBContextManager(db_config) as cursor:
if cursor is None: if cursor is None:
@@ -11,6 +12,7 @@ def select_list(db_config, sql) -> list:
lst = [dict(zip(schema, row)) for row in result] lst = [dict(zip(schema, row)) for row in result]
return lst return lst
def procedure(db_config, name, args: tuple) -> list: def procedure(db_config, name, args: tuple) -> list:
with DBContextManager(db_config) as cursor: with DBContextManager(db_config) as cursor:
if cursor is None: if cursor is None:
@@ -22,6 +24,7 @@ def procedure(db_config, name, args: tuple) -> list:
lst = dict(zip(schema, result)) lst = dict(zip(schema, result))
return lst return lst
def transaction(cursor, sql): def transaction(cursor, sql):
cursor.execute(sql) cursor.execute(sql)
return True return True

61
App/Report/__init__.py
View File

@@ -1,20 +1,26 @@
from flask import request, Blueprint, render_template, url_for
from checker import check_auth
from os import path
from datetime import date
from .report_model import view_report_model, create_report_model
import json import json
from datetime import date
from os import path
with open(path.join(path.dirname(__file__), 'reports.json'), encoding='utf-8') as f: from checker import check_auth
from flask import Blueprint, render_template, request, url_for
from .report_model import create_report_model, view_report_model
with open(path.join(path.dirname(__file__), "reports.json"), encoding="utf-8") as f:
report_list = json.load(f) report_list = json.load(f)
report_bp = Blueprint('report_bp', __name__, template_folder='templates', static_folder='static') report_bp = Blueprint(
"report_bp", __name__, template_folder="templates", static_folder="static"
)
@report_bp.route('/menu')
@report_bp.route("/menu")
@check_auth @check_auth
def menu(): def menu():
if request.method == 'GET': if request.method == "GET":
return render_template('report_menu.html') return render_template("report_menu.html")
# Рекомендации от ИС # Рекомендации от ИС
# @report_bp.route('/test', methods=['GET']) # @report_bp.route('/test', methods=['GET'])
@@ -30,15 +36,18 @@ def menu():
# report_response = model(request, report_list) # report_response = model(request, report_list)
# return view(report_response) # return view(report_response)
@report_bp.route('/create', methods=['GET', 'POST'])
@report_bp.route("/create", methods=["GET", "POST"])
@check_auth @check_auth
def create(): def create():
if request.method == 'GET': if request.method == "GET":
return render_template('report_basic.html', return render_template(
"report_basic.html",
is_write=True, is_write=True,
title='Создание отчетов', title="Создание отчетов",
items=report_list, items=report_list,
date_today=date.today()) date_today=date.today(),
)
else: else:
result = create_report_model(request, report_list) result = create_report_model(request, report_list)
if result.status: if result.status:
@@ -46,20 +55,26 @@ def create():
else: else:
return render_template("error.html", error_message=result.error_message) return render_template("error.html", error_message=result.error_message)
@report_bp.route('/view', methods=['GET', 'POST'])
@report_bp.route("/view", methods=["GET", "POST"])
@check_auth @check_auth
def view(): def view():
if request.method == 'GET': if request.method == "GET":
return render_template('report_basic.html', return render_template(
"report_basic.html",
is_write=False, is_write=False,
title='Просмотр отчетов', title="Просмотр отчетов",
items=report_list, items=report_list,
date_today=date.today()) date_today=date.today(),
)
else: else:
result = view_report_model(request, report_list) result = view_report_model(request, report_list)
if result.status: if result.status:
return render_template("output.html", items=result.result, return render_template(
header='Результаты отчёта', "output.html",
link = url_for('report_bp.menu')) items=result.result,
header="Результаты отчёта",
link=url_for("report_bp.menu"),
)
else: else:
return render_template("error.html", error_message=result.error_message) return render_template("error.html", error_message=result.error_message)

94
App/Report/report_model.py
View File

@@ -1,81 +1,87 @@
from dataclasses import dataclass from dataclasses import dataclass
from Database.work import select_list, procedure
from Database.sql_provider import SQLProvider
from flask import current_app, session
from os import path from os import path
sql_provider = SQLProvider(path.join(path.dirname(__file__), 'sql')) from Database.sql_provider import SQLProvider
from Database.work import procedure, select_list
from flask import current_app, session
sql_provider = SQLProvider(path.join(path.dirname(__file__), "sql"))
@dataclass @dataclass
class InfoRespronse: class InfoRespronse:
result: tuple result: tuple
error_message: str error_message: str
status: bool status: bool
def check_report(input_data: dict) -> bool:
db_config = current_app.config['db_config']
result = procedure(db_config, 'check_report', tuple(input_data.values())) def check_report(input_data: dict) -> bool:
if result is None or result['exist'] == 0: db_config = current_app.config["db_config"]
result = procedure(db_config, "check_report", tuple(input_data.values()))
if result is None or result["exist"] == 0:
return False return False
return True return True
def view_report_model(request, report_list: dict) -> InfoRespronse: def view_report_model(request, report_list: dict) -> InfoRespronse:
id = request.form.get('category') id = request.form.get("category")
month = request.form.get('month') month = request.form.get("month")
year = request.form.get('year') year = request.form.get("year")
data = dict(id=id, month=month, year=year) data = dict(id=id, month=month, year=year)
if session['role'] in report_list[id]['data']['read']: if session["role"] in report_list[id]["data"]["read"]:
status = check_report(data) status = check_report(data)
if not status: if not status:
return InfoRespronse((), return InfoRespronse((), error_message="Отчет не найден", status=False)
error_message = 'Отчет не найден',
status=False)
db_config = current_app.config['db_config'] db_config = current_app.config["db_config"]
view_script = report_list[id]['data']['view'] view_script = report_list[id]["data"]["view"]
_sql = sql_provider.get(f'{view_script}.sql', data) _sql = sql_provider.get(f"{view_script}.sql", data)
result = select_list(db_config, _sql) result = select_list(db_config, _sql)
if result is None: if result is None:
return InfoRespronse((), return InfoRespronse(
error_message = 'Ошибка в подключении к базе данных. Свяжитесь с администратором', (),
status=False) error_message="Ошибка в подключении к базе данных. Свяжитесь с администратором",
return InfoRespronse(result, error_message='', status=True) status=False,
)
return InfoRespronse(result, error_message="", status=True)
else: else:
return InfoRespronse((), return InfoRespronse(
error_message='Недостаточно прав для чтения данного отчета!', (),
status=False) error_message="Недостаточно прав для чтения данного отчета!",
status=False,
)
def create_report_model(request, report_list: dict) -> InfoRespronse: def create_report_model(request, report_list: dict) -> InfoRespronse:
id = request.form.get('category') id = request.form.get("category")
month = request.form.get('month') month = request.form.get("month")
year = request.form.get('year') year = request.form.get("year")
data = dict(id=id, month=month, year=year) data = dict(id=id, month=month, year=year)
if session['role'] in report_list[id]['data']['write']: if session["role"] in report_list[id]["data"]["write"]:
status = check_report(data) status = check_report(data)
if status: if status:
return InfoRespronse((), return InfoRespronse((), error_message="Отчет уже существует", status=False)
error_message = 'Отчет уже существует',
status=False)
db_config = current_app.config['db_config'] db_config = current_app.config["db_config"]
proc_name = report_list[id]['data']['procedure'] proc_name = report_list[id]["data"]["procedure"]
args = tuple(data.values()) args = tuple(data.values())
result = procedure(db_config, proc_name, args) result = procedure(db_config, proc_name, args)
if result is None: if result is None:
return InfoRespronse((), return InfoRespronse(
error_message = 'Ошибка в подключении к базе данных. Свяжитесь с администратором', (),
status=False) error_message="Ошибка в подключении к базе данных. Свяжитесь с администратором",
return InfoRespronse((), error_message='', status=True) status=False,
)
return InfoRespronse((), error_message="", status=True)
else: else:
return InfoRespronse((), return InfoRespronse(
error_message='Недостаточно прав для создания данного отчета!', (),
status=False) error_message="Недостаточно прав для создания данного отчета!",
status=False,
)

53
App/Requests/__init__.py
View File

@@ -1,44 +1,55 @@
from flask import request, Blueprint, render_template, url_for
from os import path
from checker import check_auth
from .requests_model import sklad, materials_per_seller, sellers_names, materials_names
import json import json
from os import path
with open(path.join(path.dirname(__file__), 'zapros_menu.json'), encoding='utf-8') as f: from checker import check_auth
from flask import Blueprint, render_template, request, url_for
from .requests_model import (materials_names, materials_per_seller,
sellers_names, sklad)
with open(path.join(path.dirname(__file__), "zapros_menu.json"), encoding="utf-8") as f:
requests_list = json.load(f) requests_list = json.load(f)
requests_bp = Blueprint('requests_bp', __name__, template_folder='templates') requests_bp = Blueprint("requests_bp", __name__, template_folder="templates")
@requests_bp.route('/', methods=['GET', 'POST'])
@requests_bp.route("/", methods=["GET", "POST"])
@check_auth @check_auth
def requests(): def requests():
if request.method == 'GET': if request.method == "GET":
return render_template('zapros_menu.html', options=requests_list) return render_template("zapros_menu.html", options=requests_list)
@requests_bp.route('/sklad', methods=['GET', 'POST'])
@requests_bp.route("/sklad", methods=["GET", "POST"])
@check_auth @check_auth
def sklad_zapros(): def sklad_zapros():
if request.method == 'GET': if request.method == "GET":
return render_template('zagotovki.html') return render_template("zagotovki.html")
else: else:
zagotovki = sklad(request) zagotovki = sklad(request)
if zagotovki.status: if zagotovki.status:
material = dict(request.form) material = dict(request.form)
header = f'Заготовки на складе из материала \"{material["material"]}\"' header = f'Заготовки на складе из материала "{material["material"]}"'
return render_template('output.html', items=zagotovki.result, header=header) return render_template("output.html", items=zagotovki.result, header=header)
else: else:
return render_template('error.html', error_message=zagotovki.error_message) return render_template("error.html", error_message=zagotovki.error_message)
@requests_bp.route('/shipments', methods=['GET', 'POST'])
@requests_bp.route("/shipments", methods=["GET", "POST"])
@check_auth @check_auth
def sellers_ship(): def sellers_ship():
if request.method == 'GET': if request.method == "GET":
return render_template('sellers_ship.html') return render_template("sellers_ship.html")
else: else:
zagotovki = materials_per_seller(request) zagotovki = materials_per_seller(request)
if zagotovki.status: if zagotovki.status:
seller = dict(request.form) seller = dict(request.form)
header = f'Поставки от поставщика \"{seller["seller"]}\"' header = f'Поставки от поставщика "{seller["seller"]}"'
return render_template('output.html', items=zagotovki.result, header=header, link=url_for('requests_bp.requests')) return render_template(
"output.html",
items=zagotovki.result,
header=header,
link=url_for("requests_bp.requests"),
)
else: else:
return render_template('error.html', error_message=zagotovki.error_message) return render_template("error.html", error_message=zagotovki.error_message)

74
App/Requests/requests_model.py
View File

@@ -1,61 +1,73 @@
from dataclasses import dataclass from dataclasses import dataclass
from Database.work import select_list
from Database.sql_provider import SQLProvider
from flask import current_app
from os import path from os import path
sql_provider = SQLProvider(path.join(path.dirname(__file__), 'sql')) from Database.sql_provider import SQLProvider
from Database.work import select_list
from flask import current_app
sql_provider = SQLProvider(path.join(path.dirname(__file__), "sql"))
@dataclass @dataclass
class InfoRespronse: class InfoRespronse:
result: tuple result: tuple
error_message: str error_message: str
status: bool status: bool
def sellers_names() -> InfoRespronse:
db_config = current_app.config['db_config']
_sql = sql_provider.get('sellers_names.sql', {}) def sellers_names() -> InfoRespronse:
db_config = current_app.config["db_config"]
_sql = sql_provider.get("sellers_names.sql", {})
result = select_list(db_config, _sql) result = select_list(db_config, _sql)
if result is None: if result is None:
return InfoRespronse((), return InfoRespronse(
error_message = 'Ошибка в подключении к базе данных. Свяжитесь с администратором', (),
status=False) error_message="Ошибка в подключении к базе данных. Свяжитесь с администратором",
return InfoRespronse(result, error_message='', status=True) status=False,
)
return InfoRespronse(result, error_message="", status=True)
def materials_names() -> InfoRespronse: def materials_names() -> InfoRespronse:
db_config = current_app.config['db_config'] db_config = current_app.config["db_config"]
_sql = sql_provider.get('materials_names.sql', {}) _sql = sql_provider.get("materials_names.sql", {})
result = select_list(db_config, _sql) result = select_list(db_config, _sql)
if result is None: if result is None:
return InfoRespronse((), return InfoRespronse(
error_message = 'Ошибка в подключении к базе данных. Свяжитесь с администратором', (),
status=False) error_message="Ошибка в подключении к базе данных. Свяжитесь с администратором",
return InfoRespronse(result, error_message='', status=True) status=False,
)
return InfoRespronse(result, error_message="", status=True)
def sklad(request) -> InfoRespronse: def sklad(request) -> InfoRespronse:
db_config = current_app.config['db_config'] db_config = current_app.config["db_config"]
material = dict(request.form) material = dict(request.form)
_sql = sql_provider.get('sklad_material.sql', material) _sql = sql_provider.get("sklad_material.sql", material)
result = select_list(db_config, _sql) result = select_list(db_config, _sql)
if result is None: if result is None:
return InfoRespronse((), return InfoRespronse(
error_message = 'Ошибка в подключении к базе данных. Свяжитесь с администратором', (),
status=False) error_message="Ошибка в подключении к базе данных. Свяжитесь с администратором",
return InfoRespronse(result, error_message='', status=True) status=False,
)
return InfoRespronse(result, error_message="", status=True)
def materials_per_seller(request) -> InfoRespronse: def materials_per_seller(request) -> InfoRespronse:
db_config = current_app.config['db_config'] db_config = current_app.config["db_config"]
seller = dict(request.form) seller = dict(request.form)
_sql = sql_provider.get('ship_seller.sql', seller) _sql = sql_provider.get("ship_seller.sql", seller)
result = select_list(db_config, _sql) result = select_list(db_config, _sql)
if result is None: if result is None:
return InfoRespronse((), return InfoRespronse(
error_message = 'Ошибка в подключении к базе данных. Свяжитесь с администратором', (),
status=False) error_message="Ошибка в подключении к базе данных. Свяжитесь с администратором",
return InfoRespronse(result, error_message='', status=True) status=False,
)
return InfoRespronse(result, error_message="", status=True)

46
App/Waybill/__init__.py
View File

@@ -1,42 +1,56 @@
from flask import Blueprint, render_template, redirect, url_for, session, request
from checker import check_auth
from .model import index_waybill, form_waybill, clear, button_click, transaction_order_model
from datetime import datetime from datetime import datetime
waybill_bp = Blueprint('waybill_bp', __name__, template_folder='templates', static_folder='static') from checker import check_auth
from flask import (Blueprint, redirect, render_template, request, session,
url_for)
@waybill_bp.route('/', methods=['GET']) from .model import (button_click, clear, form_waybill, index_waybill,
transaction_order_model)
waybill_bp = Blueprint(
"waybill_bp", __name__, template_folder="templates", static_folder="static"
)
@waybill_bp.route("/", methods=["GET"])
@check_auth @check_auth
def index(): def index():
lst = index_waybill() lst = index_waybill()
if lst is not None: if lst is not None:
waybill = form_waybill() waybill = form_waybill()
return render_template('waybill.html', items=lst, waybill=waybill) return render_template("waybill.html", items=lst, waybill=waybill)
else: else:
return render_template('error.html', error_message="Ошибка в подключении к СУБД") return render_template(
"error.html", error_message="Ошибка в подключении к СУБД"
)
@waybill_bp.route('/', methods=['POST'])
@waybill_bp.route("/", methods=["POST"])
@check_auth @check_auth
def waybill_main(): def waybill_main():
status = button_click(request) status = button_click(request)
if status: if status:
return redirect(url_for('waybill_bp.index')) return redirect(url_for("waybill_bp.index"))
else: else:
return render_template("error.html", error_message="Товар не был добавлен в корзину") return render_template(
"error.html", error_message="Товар не был добавлен в корзину"
)
@waybill_bp.route('/clear', methods=['GET'])
@waybill_bp.route("/clear", methods=["GET"])
@check_auth @check_auth
def clear_waybill(): def clear_waybill():
clear() clear()
return redirect(url_for('waybill_bp.index')) return redirect(url_for("waybill_bp.index"))
@waybill_bp.route('/save_order')
@waybill_bp.route("/save_order")
@check_auth @check_auth
def save_order(): def save_order():
if not session.get('waybill',{}): if not session.get("waybill", {}):
return redirect(url_for('waybill_bp.index')) return redirect(url_for("waybill_bp.index"))
user_id = session.get('user_id',"") user_id = session.get("user_id", "")
current_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S") current_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
result = transaction_order_model(user_id, current_date) result = transaction_order_model(user_id, current_date)

119
App/Waybill/model.py
View File

@@ -1,12 +1,13 @@
from Database.sql_provider import SQLProvider import os
from Database.work import select_list
from Database.DBconnect import DBContextManager
from flask import current_app, session
from dataclasses import dataclass from dataclasses import dataclass
from datetime import date from datetime import date
from cache.wrapper import fetch_from_cache from cache.wrapper import fetch_from_cache
import os from Database.DBconnect import DBContextManager
from Database.sql_provider import SQLProvider
from Database.work import select_list
from flask import current_app, session
@dataclass @dataclass
class InfoRespronse: class InfoRespronse:
@@ -14,103 +15,111 @@ class InfoRespronse:
error_message: str error_message: str
status: bool status: bool
sql_provider = SQLProvider(os.path.join(os.path.dirname(__file__), 'sql'))
sql_provider = SQLProvider(os.path.join(os.path.dirname(__file__), "sql"))
def clear(): def clear():
if session.get('waybill',{}): if session.get("waybill", {}):
session.pop('waybill') session.pop("waybill")
def form_waybill() -> list: def form_waybill() -> list:
db_config = current_app.config['db_config'] db_config = current_app.config["db_config"]
cache_config = current_app.config['cache_config'] cache_config = current_app.config["cache_config"]
current_waybill = session.get('waybill',{}) current_waybill = session.get("waybill", {})
waybill = [] waybill = []
for k, v in current_waybill.items(): for k, v in current_waybill.items():
_sql = sql_provider.get('one_good.sql', dict(work_id=k)) _sql = sql_provider.get("one_good.sql", dict(work_id=k))
cache_select = fetch_from_cache(f'product_{k}', cache_config)(select_list) cache_select = fetch_from_cache(f"product_{k}", cache_config)(select_list)
product = cache_select(db_config, _sql)[0] product = cache_select(db_config, _sql)[0]
product['amount'] = v product["amount"] = v
waybill.append(product) waybill.append(product)
return waybill return waybill
def index_waybill() -> list:
db_config = current_app.config['db_config']
cache_config = current_app.config['cache_config']
cache_select = fetch_from_cache('items_cached', cache_config)(select_list) def index_waybill() -> list:
_sql = sql_provider.get('goods.sql', {}) db_config = current_app.config["db_config"]
cache_config = current_app.config["cache_config"]
cache_select = fetch_from_cache("items_cached", cache_config)(select_list)
_sql = sql_provider.get("goods.sql", {})
products = cache_select(db_config, _sql) products = cache_select(db_config, _sql)
if products == None: if products is None:
return [] return []
return products return products
def button_click(request):
db_config = current_app.config['db_config']
data = dict(work_id=int(request.form['product_display']))
_sql = sql_provider.get('one_good.sql', data) def button_click(request):
db_config = current_app.config["db_config"]
data = dict(work_id=int(request.form["product_display"]))
_sql = sql_provider.get("one_good.sql", data)
result = select_list(db_config, _sql) result = select_list(db_config, _sql)
if result == None: if result is None:
return False return False
product = result[0] product = result[0]
if request.form.get('add'): if request.form.get("add"):
if 'waybill' not in session: if "waybill" not in session:
session['waybill'] = dict() session["waybill"] = dict()
session['total'] = '0' session["total"] = "0"
if str(product['work_id']) in session['waybill']: if str(product["work_id"]) in session["waybill"]:
pr_id = product['work_id'] pr_id = product["work_id"]
price = product['price'] price = product["price"]
amount = int(session['waybill'][str(pr_id)]) amount = int(session["waybill"][str(pr_id)])
session['waybill'][str(pr_id)] = str(amount+1) session["waybill"][str(pr_id)] = str(amount + 1)
session['total'] = str(int(session['total']) + price) session["total"] = str(int(session["total"]) + price)
session.modified = True session.modified = True
else: else:
print("NEW WORKPIECE") print("NEW WORKPIECE")
pr_id = product['work_id'] pr_id = product["work_id"]
price = product['price'] price = product["price"]
session['waybill'][str(pr_id)] = '1' session["waybill"][str(pr_id)] = "1"
session['total'] = str(int(session['total']) + price) session["total"] = str(int(session["total"]) + price)
print(session['waybill']) print(session["waybill"])
session.modified = True session.modified = True
elif request.form.get('product_display_minus'): elif request.form.get("product_display_minus"):
# decreasing count in waybill # decreasing count in waybill
amount = int(session['waybill'][str(product['work_id'])]) amount = int(session["waybill"][str(product["work_id"])])
if amount == 1: if amount == 1:
session['waybill'].pop(str(product['work_id'])) session["waybill"].pop(str(product["work_id"]))
else: else:
session['waybill'][str(product['work_id'])] = str(amount-1) session["waybill"][str(product["work_id"])] = str(amount - 1)
session['total'] = str(int(session['total']) - product['price']) session["total"] = str(int(session["total"]) - product["price"])
session.modified = True session.modified = True
return True return True
def transaction_order_model(user_id: int, current_date: date):
db_config = current_app.config['db_config'] def transaction_order_model(user_id: int, current_date: date):
waybill = session.get('waybill',{}) db_config = current_app.config["db_config"]
total = session.get('total', 0) waybill = session.get("waybill", {})
total = session.get("total", 0)
# Чтобы всё это шло как одна транзакция # Чтобы всё это шло как одна транзакция
with DBContextManager(db_config) as cursor: with DBContextManager(db_config) as cursor:
data = dict(e_user_id=user_id, e_order_date=current_date, e_total=total) data = dict(e_user_id=user_id, e_order_date=current_date, e_total=total)
try: try:
_sql = sql_provider.get('create_order.sql', data) _sql = sql_provider.get("create_order.sql", data)
cursor.execute(_sql) cursor.execute(_sql)
order_id = cursor.lastrowid order_id = cursor.lastrowid
for key, value in waybill.items(): for key, value in waybill.items():
_sql = sql_provider.get('insert_order_line.sql', _sql = sql_provider.get(
dict(e_order_id = order_id, "insert_order_line.sql",
dict(
e_order_id=order_id,
e_price=0, e_price=0,
e_prod_id=int(key), e_prod_id=int(key),
e_amount = int(value))) e_amount=int(value),
),
)
cursor.execute(_sql) cursor.execute(_sql)
except: except:
return InfoRespronse((), error_message="Заказ не был создан", status=False) return InfoRespronse((), error_message="Заказ не был создан", status=False)

54
App/app.py
View File

@@ -1,32 +1,52 @@
import json
import os
from Auth import auth_bp
from flask import Flask, render_template, session from flask import Flask, render_template, session
from Report import report_bp
from Requests import requests_bp from Requests import requests_bp
from Waybill import waybill_bp from Waybill import waybill_bp
from Auth import auth_bp
from Report import report_bp
import os, json
app = Flask(__name__) app = Flask(__name__)
app.secret_key = 'suplex' app.secret_key = "suplex"
app.config.update( app.config.update(
db_config=json.load(open(os.path.join(os.path.dirname(__file__), 'data/config.json'), encoding='utf-8')), db_config=json.load(
db_access=json.load(open(os.path.join(os.path.dirname(__file__), 'data/db_access.json'), encoding='utf-8')), open(
cache_config=json.load(open(os.path.join(os.path.dirname(__file__), 'data/redis_config.json'), encoding='utf-8')) os.path.join(os.path.dirname(__file__), "data/config.json"),
encoding="utf-8",
)
),
db_access=json.load(
open(
os.path.join(os.path.dirname(__file__), "data/db_access.json"),
encoding="utf-8",
)
),
cache_config=json.load(
open(
os.path.join(os.path.dirname(__file__), "data/redis_config.json"),
encoding="utf-8",
)
),
) )
app.register_blueprint(requests_bp, url_prefix='/requests') app.register_blueprint(requests_bp, url_prefix="/requests")
app.register_blueprint(auth_bp, url_prefix='/auth') app.register_blueprint(auth_bp, url_prefix="/auth")
app.register_blueprint(report_bp, url_prefix='/report') app.register_blueprint(report_bp, url_prefix="/report")
app.register_blueprint(waybill_bp, url_prefix='/waybill') app.register_blueprint(waybill_bp, url_prefix="/waybill")
@app.route('/')
@app.route("/")
def index(): def index():
return render_template('main_menu.html', ses=session) return render_template("main_menu.html", ses=session)
@app.route('/logout')
@app.route("/logout")
def logout(): def logout():
session.clear() session.clear()
return render_template('main_menu.html', ses=session) return render_template("main_menu.html", ses=session)
if __name__ == '__main__':
app.run(port=5002, host='0.0.0.0') if __name__ == "__main__":
app.run(port=5002, host="0.0.0.0")

4
App/cache/__init__.py vendored
View File

@@ -1,5 +1,7 @@
import json import json
from redis import Redis, ConnectionError, DataError
from redis import ConnectionError, DataError, Redis
class RedisCache: class RedisCache:
def __init__(self, config: dict): def __init__(self, config: dict):

9
App/cache/wrapper.py vendored
View File

@@ -1,15 +1,18 @@
from functools import wraps from functools import wraps
from . import RedisCache from . import RedisCache
def fetch_from_cache(cache_name: str, cache_config: dict): def fetch_from_cache(cache_name: str, cache_config: dict):
cache_conn = RedisCache(cache_config['redis']) cache_conn = RedisCache(cache_config["redis"])
ttl = cache_config['ttl'] ttl = cache_config["ttl"]
""" """
It checks if a cached value exists for a given cache_name. It checks if a cached value exists for a given cache_name.
If a cached value exists, it returns the cached value. If a cached value exists, it returns the cached value.
If no cached value exists, it calls the original function f with the provided arguments, caches the result, and then returns the result. If no cached value exists, it calls the original function f with the provided arguments, caches the result, and then returns the result.
""" """
def decorator(f): def decorator(f):
@wraps(f) @wraps(f)
def wrapper(*args, **kwargs): def wrapper(*args, **kwargs):
@@ -21,5 +24,7 @@ def fetch_from_cache(cache_name: str, cache_config: dict):
print("response=", response) print("response=", response)
cache_conn.set_value(cache_name, response, ttl) cache_conn.set_value(cache_name, response, ttl)
return response return response
return wrapper return wrapper
return decorator return decorator

19
App/checker.py
View File

@@ -1,19 +1,22 @@
from flask import redirect, url_for, session, request, current_app, render_template
from functools import wraps from functools import wraps
from flask import (current_app, redirect, render_template, request, session,
url_for)
def check_auth(func): def check_auth(func):
@wraps(func) @wraps(func)
def wrapper(*args, **kwargs): def wrapper(*args, **kwargs):
if 'login' not in session: if "login" not in session:
return redirect(url_for('auth_bp.auth')) return redirect(url_for("auth_bp.auth"))
user_role = session.get('role') user_role = session.get("role")
user_request = request.endpoint user_request = request.endpoint
print('request_endpoint=', user_request) print("request_endpoint=", user_request)
user_bp = user_request.split('.')[0] user_bp = user_request.split(".")[0]
access = current_app.config['db_access'] access = current_app.config["db_access"]
if user_role in access and user_bp in access[user_role]: if user_role in access and user_bp in access[user_role]:
return func(*args, **kwargs) return func(*args, **kwargs)
else: else:
return render_template('error.html', error_message='Недостаточно прав') return render_template("error.html", error_message="Недостаточно прав")
return wrapper return wrapper