299 lines
10 KiB
Python
299 lines
10 KiB
Python
"""
|
|
Project: PARTS Website
|
|
Author: Edward Middleton-Smith
|
|
Precision And Research Technology Systems Limited
|
|
|
|
Technology: DataStores
|
|
Feature: User DataStore
|
|
|
|
Description:
|
|
Datastore for Users
|
|
"""
|
|
|
|
# internal
|
|
# from routes import bp_home
|
|
import lib.argument_validation as av
|
|
from business_objects.dog.command import Command, Command_Temp
|
|
from business_objects.dog.command_category import Command_Category, Command_Category_Temp
|
|
from business_objects.dog.dog import Dog
|
|
from business_objects.dog.dog_command_link import Dog_Command_Link, Dog_Command_Link_Temp
|
|
from business_objects.sql_error import SQL_Error
|
|
from datastores.datastore_base import DataStore_Base
|
|
from helpers.helper_app import Helper_App
|
|
from helpers.helper_db_mysql import Helper_DB_MySQL
|
|
# from models.model_view_store_checkout import Model_View_Store_Checkout # circular!
|
|
from extensions import db
|
|
# external
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from datetime import datetime
|
|
|
|
db = SQLAlchemy()
|
|
|
|
|
|
class DataStore_Dog(DataStore_Base):
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
@classmethod
|
|
def get_many_dog(cls, filters_dog):
|
|
_m = f'{cls.__qualname__}.get_many_dog'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_dog.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_dog', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Dogs
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw dogs: {result_set_1}')
|
|
dogs = []
|
|
dog_indexes = {}
|
|
for row in result_set_1:
|
|
new_dog = Dog.from_db_dog(row)
|
|
dog_indexes[new_dog.id_dog] = len(dogs)
|
|
dogs.append(new_dog)
|
|
|
|
# Errors
|
|
cursor.nextset()
|
|
result_set_e = cursor.fetchall()
|
|
Helper_App.console_log(f'raw errors: {result_set_e}')
|
|
errors = []
|
|
if len(result_set_e) > 0:
|
|
errors = [SQL_Error.from_db_record(row) for row in result_set_e]
|
|
for error in errors:
|
|
Helper_App.console_log(f"Error [{error.code}]: {error.msg}")
|
|
|
|
cls.db_cursor_clear(cursor)
|
|
|
|
return dogs, errors
|
|
|
|
@classmethod
|
|
def get_many_command(cls, filters_command):
|
|
_m = f'{cls.__qualname__}.get_many_command'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_command.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_command', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Command Categories
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw command categories: {result_set_1}')
|
|
command_categories = []
|
|
command_category_indexes = {}
|
|
for row in result_set_1:
|
|
new_command_category = Command_Category.from_db_command(row)
|
|
command_category_indexes[new_command_category.id_command_category] = len(command_categories)
|
|
command_categories.append(new_command_category)
|
|
|
|
Helper_App.console_log(f'command_category_indexes: {command_category_indexes}')
|
|
|
|
# Commands
|
|
cursor.nextset()
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw commands: {result_set_1}')
|
|
commands = []
|
|
command_indexes = {}
|
|
for row in result_set_1:
|
|
new_command = Command.from_db_command(row)
|
|
command_indexes[new_command.id_command] = len(commands)
|
|
index_command_category = command_category_indexes[new_command.id_command_category]
|
|
new_command.command_category = command_categories[index_command_category]
|
|
commands.append(new_command)
|
|
|
|
# Errors
|
|
cursor.nextset()
|
|
result_set_e = cursor.fetchall()
|
|
Helper_App.console_log(f'raw errors: {result_set_e}')
|
|
errors = []
|
|
if len(result_set_e) > 0:
|
|
errors = [SQL_Error.from_db_record(row) for row in result_set_e]
|
|
for error in errors:
|
|
Helper_App.console_log(f"Error [{error.code}]: {error.msg}")
|
|
|
|
cls.db_cursor_clear(cursor)
|
|
|
|
return command_categories, commands, errors
|
|
|
|
@classmethod
|
|
def get_many_dog_command_link(cls, filters_dog_command_link):
|
|
_m = f'{cls.__qualname__}.get_many_dog_command_link'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_dog_command_link.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_dog_command_link', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Dog Command Links
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw dog command links: {result_set_1}')
|
|
dog_command_links = []
|
|
dog_command_link_indexes = {}
|
|
for row in result_set_1:
|
|
# Helper_App.console_log(f'Raw dog command link: {row}')
|
|
new_dog_command_link = Dog_Command_Link.from_db_dog_command_link(row)
|
|
dog_command_link_indexes[new_dog_command_link.id_link] = len(dog_command_links)
|
|
dog_command_links.append(new_dog_command_link)
|
|
|
|
# Errors
|
|
cursor.nextset()
|
|
result_set_e = cursor.fetchall()
|
|
Helper_App.console_log(f'raw errors: {result_set_e}')
|
|
errors = []
|
|
if len(result_set_e) > 0:
|
|
errors = [SQL_Error.from_db_record(row) for row in result_set_e]
|
|
for error in errors:
|
|
Helper_App.console_log(f"Error [{error.code}]: {error.msg}")
|
|
|
|
cls.db_cursor_clear(cursor)
|
|
|
|
return dog_command_links, errors
|
|
|
|
@classmethod
|
|
def save_dog_command_links(cls, comment, links):
|
|
_m = f'{cls}.save_dog_command_links'
|
|
av.val_str(comment, 'comment', _m)
|
|
|
|
guid = Helper_DB_MySQL.create_guid_str()
|
|
now = datetime.now()
|
|
user = cls.get_user_session()
|
|
|
|
Helper_App.console_log(f'saving links: {links}')
|
|
|
|
rows = []
|
|
for link in links:
|
|
row = Dog_Command_Link_Temp.from_dog_command_link(link)
|
|
row.guid = guid
|
|
rows.append(row)
|
|
|
|
cls.upload_bulk(Dog_Command_Link_Temp.__tablename__, rows, 1000)
|
|
|
|
Helper_App.console_log('Links uploaded')
|
|
|
|
argument_dict_list = {
|
|
'a_comment': comment,
|
|
'a_guid': guid,
|
|
'a_id_user': user.id_user,
|
|
'a_debug': 0
|
|
}
|
|
result = cls.db_procedure_execute('p_dog_save_dog_command_link', argument_dict_list)
|
|
|
|
Helper_App.console_log('Dog Command Links saved')
|
|
|
|
# Errors
|
|
cursor = result.cursor
|
|
cursor.nextset()
|
|
result_set_e = cursor.fetchall()
|
|
errors = []
|
|
if len(result_set_e) > 0:
|
|
errors = [SQL_Error.from_db_record(row) for row in result_set_e]
|
|
for error in errors:
|
|
Helper_App.console_log(f"Error [{error.code}]: {error.msg}")
|
|
|
|
cls.db_cursor_clear(cursor)
|
|
return errors
|
|
|
|
@classmethod
|
|
def save_commands(cls, comment, commands):
|
|
_m = f'{cls}.save_commands'
|
|
av.val_str(comment, 'comment', _m)
|
|
|
|
guid = Helper_DB_MySQL.create_guid_str()
|
|
now = datetime.now()
|
|
user = cls.get_user_session()
|
|
|
|
Helper_App.console_log(f'saving commands: {commands}')
|
|
|
|
rows = []
|
|
for command in commands:
|
|
row = Command_Temp.from_command(command)
|
|
row.guid = guid
|
|
rows.append(row)
|
|
|
|
cls.upload_bulk(Command_Temp.__tablename__, rows, 1000)
|
|
|
|
Helper_App.console_log('Commands uploaded')
|
|
|
|
argument_dict_list = {
|
|
'a_comment': comment,
|
|
'a_guid': guid,
|
|
'a_id_user': user.id_user,
|
|
'a_debug': 0
|
|
}
|
|
result = cls.db_procedure_execute('p_dog_save_command', argument_dict_list)
|
|
|
|
Helper_App.console_log('Commands saved')
|
|
|
|
# Errors
|
|
cursor = result.cursor
|
|
cursor.nextset()
|
|
result_set_e = cursor.fetchall()
|
|
errors = []
|
|
if len(result_set_e) > 0:
|
|
errors = [SQL_Error.from_db_record(row) for row in result_set_e]
|
|
for error in errors:
|
|
Helper_App.console_log(f"Error [{error.code}]: {error.msg}")
|
|
|
|
cls.db_cursor_clear(cursor)
|
|
return errors
|
|
|
|
@classmethod
|
|
def save_command_categories(cls, comment, command_categories):
|
|
_m = f'{cls}.save_command_categories'
|
|
av.val_str(comment, 'comment', _m)
|
|
|
|
guid = Helper_DB_MySQL.create_guid_str()
|
|
now = datetime.now()
|
|
user = cls.get_user_session()
|
|
|
|
Helper_App.console_log(f'saving command categories: {command_categories}')
|
|
|
|
rows = []
|
|
for command_category in command_categories:
|
|
row = Command_Category_Temp.from_command_category(command_category)
|
|
row.guid = guid
|
|
rows.append(row)
|
|
|
|
cls.upload_bulk(Command_Category_Temp.__tablename__, rows, 1000)
|
|
|
|
Helper_App.console_log('Commands uploaded')
|
|
|
|
argument_dict_list = {
|
|
'a_comment': comment,
|
|
'a_guid': guid,
|
|
'a_id_user': user.id_user,
|
|
'a_debug': 0
|
|
}
|
|
result = cls.db_procedure_execute('p_dog_save_command_category', argument_dict_list)
|
|
|
|
Helper_App.console_log('Commands saved')
|
|
|
|
# Errors
|
|
cursor = result.cursor
|
|
cursor.nextset()
|
|
result_set_e = cursor.fetchall()
|
|
errors = []
|
|
if len(result_set_e) > 0:
|
|
errors = [SQL_Error.from_db_record(row) for row in result_set_e]
|
|
for error in errors:
|
|
Helper_App.console_log(f"Error [{error.code}]: {error.msg}")
|
|
|
|
cls.db_cursor_clear(cursor)
|
|
return errors
|
|
|
|
|
|
|