1430 lines
53 KiB
Python
1430 lines
53 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.assessment import Assessment, Assessment_Temp
|
|
from business_objects.dog.assessment_command_modality_link import Assessment_Command_Modality_Link, Assessment_Command_Modality_Link_Temp
|
|
from business_objects.dog.assessment_response import Assessment_Response, Assessment_Response_Temp
|
|
from business_objects.dog.bribe import Bribe, Bribe_Temp
|
|
from business_objects.dog.button_icon import Button_Icon, Button_Icon_Temp
|
|
from business_objects.dog.button_shape import Button_Shape, Button_Shape_Temp
|
|
from business_objects.dog.colour import Colour, Colour_Temp
|
|
from business_objects.dog.command import Command, Command_Temp
|
|
from business_objects.dog.command_button_link import Command_Button_Link, Command_Button_Link_Temp
|
|
from business_objects.dog.command_category import Command_Category, Command_Category_Temp
|
|
from business_objects.dog.command_modality import Command_Modality, Command_Modality_Temp
|
|
from business_objects.dog.distraction import Distraction, Distraction_Temp
|
|
from business_objects.dog.distraction_intensity_level import Distraction_Intensity_Level, Distraction_Intensity_Level_Temp
|
|
from business_objects.dog.distraction_type import Distraction_Type, Distraction_Type_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.dog.lighting_level import Lighting_Level, Lighting_Level_Temp
|
|
from business_objects.dog.location import Location, Location_Temp
|
|
from business_objects.dog.obedience_level import Obedience_Level, Obedience_Level_Temp
|
|
from business_objects.dog.response_quality_metric import Response_Quality_Metric, Response_Quality_Metric_Temp
|
|
from business_objects.dog.weather import Weather, Weather_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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return command_categories, commands, 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('Command Categories 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('Command Categories 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
return 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
return errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_button_shape(cls, filters_button_shape):
|
|
_m = f'{cls.__qualname__}.get_many_button_shape'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_button_shape.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_button_shape', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Button_Shapes
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw button_shapes: {result_set_1}')
|
|
button_shapes = []
|
|
button_shape_indexes = {}
|
|
for row in result_set_1:
|
|
new_button_shape = Button_Shape.from_db_button_shape(row)
|
|
button_shape_indexes[new_button_shape.id_button_shape] = len(button_shapes)
|
|
button_shapes.append(new_button_shape)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return button_shapes, errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_colour(cls, filters_colour):
|
|
_m = f'{cls.__qualname__}.get_many_colour'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_colour.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_colour', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Colours
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw colours: {result_set_1}')
|
|
colours = []
|
|
colour_indexes = {}
|
|
for row in result_set_1:
|
|
new_colour = Colour.from_db_colour(row)
|
|
colour_indexes[new_colour.id_colour] = len(colours)
|
|
colours.append(new_colour)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return colours, errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_location(cls, filters_location):
|
|
_m = f'{cls.__qualname__}.get_many_location'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_location.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_location', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Locations
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw locations: {result_set_1}')
|
|
locations = []
|
|
location_indexes = {}
|
|
for row in result_set_1:
|
|
new_location = Location.from_db_location(row)
|
|
location_indexes[new_location.id_location] = len(locations)
|
|
locations.append(new_location)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return locations, errors
|
|
|
|
@classmethod
|
|
def save_locations(cls, comment, locations):
|
|
_m = f'{cls}.save_locations'
|
|
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 locations: {locations}')
|
|
|
|
rows = []
|
|
for link in locations:
|
|
row = Location_Temp.from_location(link)
|
|
row.guid = guid
|
|
rows.append(row)
|
|
|
|
Helper_App.console_log(f'rows: {rows}')
|
|
|
|
cls.upload_bulk(Location_Temp.__tablename__, rows, 1000)
|
|
|
|
Helper_App.console_log('locations 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_location', argument_dict_list)
|
|
|
|
Helper_App.console_log('Locations saved')
|
|
|
|
# Errors
|
|
cursor = result.cursor
|
|
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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return [] # errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_button_icon(cls, filters_button_icon):
|
|
_m = f'{cls.__qualname__}.get_many_button_icon'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_button_icon.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_button_icon', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Button_Icons
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw button_icons: {result_set_1}')
|
|
button_icons = []
|
|
button_icon_indexes = {}
|
|
for row in result_set_1:
|
|
new_button_icon = Button_Icon.from_db_button_icon(row)
|
|
button_icon_indexes[new_button_icon.id_button_icon] = len(button_icons)
|
|
button_icons.append(new_button_icon)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return button_icons, errors
|
|
|
|
@classmethod
|
|
def save_button_icons(cls, comment, button_icons):
|
|
_m = f'{cls}.save_button_icons'
|
|
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 button_icons: {button_icons}')
|
|
|
|
rows = []
|
|
for link in button_icons:
|
|
row = Button_Icon_Temp.from_button_icon(link)
|
|
row.guid = guid
|
|
rows.append(row)
|
|
|
|
Helper_App.console_log(f'rows: {rows}')
|
|
|
|
cls.upload_bulk(Button_Icon_Temp.__tablename__, rows, 1000)
|
|
|
|
Helper_App.console_log('button_icons 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_button_icon', argument_dict_list)
|
|
|
|
Helper_App.console_log('Button_Icons saved')
|
|
|
|
# Errors
|
|
cursor = result.cursor
|
|
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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_command_button_link(cls, filters_command_button_link):
|
|
_m = f'{cls.__qualname__}.get_many_command_button_link'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_command_button_link.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_command_button_link', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Command Button Links
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw command button links: {result_set_1}')
|
|
command_button_links = []
|
|
command_button_link_indexes = {}
|
|
for row in result_set_1:
|
|
# Helper_App.console_log(f'Raw dog command link: {row}')
|
|
new_command_button_link = Command_Button_Link.from_db_command_button_link(row)
|
|
command_button_link_indexes[new_command_button_link.id_link] = len(command_button_links)
|
|
command_button_links.append(new_command_button_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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return command_button_links, errors
|
|
|
|
@classmethod
|
|
def save_command_button_links(cls, comment, links):
|
|
_m = f'{cls}.save_command_button_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 = Command_Button_Link_Temp.from_command_button_link(link)
|
|
row.guid = guid
|
|
rows.append(row)
|
|
|
|
cls.upload_bulk(Command_Button_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_command_button_link', argument_dict_list)
|
|
|
|
Helper_App.console_log('Command Button 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
return errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_assessment(cls, filters_assessment):
|
|
_m = f'{cls.__qualname__}.get_many_assessment'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_assessment.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_assessment', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Assessments
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw assessments: {result_set_1}')
|
|
assessments = []
|
|
assessment_indexes = {}
|
|
for row in result_set_1:
|
|
new_assessment = Assessment.from_db_assessment(row)
|
|
assessment_indexes[new_assessment.id_assessment] = len(assessments)
|
|
assessments.append(new_assessment)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return assessments, errors
|
|
|
|
@classmethod
|
|
def save_assessments(cls, comment, assessments):
|
|
_m = f'{cls}.save_assessments'
|
|
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 assessments: {assessments}')
|
|
|
|
rows = []
|
|
for link in assessments:
|
|
row = Assessment_Temp.from_assessment(link)
|
|
row.guid = guid
|
|
rows.append(row)
|
|
|
|
Helper_App.console_log(f'rows: {rows}')
|
|
|
|
cls.upload_bulk(Assessment_Temp.__tablename__, rows, 1000)
|
|
|
|
Helper_App.console_log('assessments 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_assessment', argument_dict_list)
|
|
|
|
Helper_App.console_log('Assessments saved')
|
|
|
|
# Errors
|
|
cursor = result.cursor
|
|
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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_weather(cls, filters_weather):
|
|
_m = f'{cls.__qualname__}.get_many_weather'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_weather.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_weather', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Weathers
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw weathers: {result_set_1}')
|
|
weathers = []
|
|
weather_indexes = {}
|
|
for row in result_set_1:
|
|
new_weather = Weather.from_db_weather(row)
|
|
weather_indexes[new_weather.id_weather] = len(weathers)
|
|
weathers.append(new_weather)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return weathers, errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_lighting_level(cls, filters_lighting_level):
|
|
_m = f'{cls.__qualname__}.get_many_lighting_level'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_lighting_level.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_lighting_level', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Lighting_Levels
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw lighting_levels: {result_set_1}')
|
|
lighting_levels = []
|
|
lighting_level_indexes = {}
|
|
for row in result_set_1:
|
|
new_lighting_level = Lighting_Level.from_db_lighting_level(row)
|
|
lighting_level_indexes[new_lighting_level.id_lighting_level] = len(lighting_levels)
|
|
lighting_levels.append(new_lighting_level)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return lighting_levels, errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_distraction_type(cls, filters_distraction_type):
|
|
_m = f'{cls.__qualname__}.get_many_distraction_type'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_distraction_type.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_distraction_type', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Distraction_Types
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw distraction_types: {result_set_1}')
|
|
distraction_types = []
|
|
distraction_type_indexes = {}
|
|
for row in result_set_1:
|
|
new_distraction_type = Distraction_Type.from_db_distraction_type(row)
|
|
distraction_type_indexes[new_distraction_type.id_distraction_type] = len(distraction_types)
|
|
distraction_types.append(new_distraction_type)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return distraction_types, errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_distraction_intensity_level(cls, filters_distraction_intensity_level):
|
|
_m = f'{cls.__qualname__}.get_many_distraction_intensity_level'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_distraction_intensity_level.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_distraction_intensity_level', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Distraction_Intensity_Levels
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw distraction_intensity_levels: {result_set_1}')
|
|
distraction_intensity_levels = []
|
|
distraction_intensity_level_indexes = {}
|
|
for row in result_set_1:
|
|
new_distraction_intensity_level = Distraction_Intensity_Level.from_db_distraction_intensity_level(row)
|
|
distraction_intensity_level_indexes[new_distraction_intensity_level.id_intensity_level] = len(distraction_intensity_levels)
|
|
distraction_intensity_levels.append(new_distraction_intensity_level)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return distraction_intensity_levels, errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_distraction(cls, filters_distraction):
|
|
_m = f'{cls.__qualname__}.get_many_distraction'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_distraction.to_json()
|
|
, 'a_show_errors': 1
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_distraction', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Distractions
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw distractions: {result_set_1}')
|
|
distractions = []
|
|
distraction_indexes = {}
|
|
for row in result_set_1:
|
|
new_distraction = Distraction.from_db_distraction(row)
|
|
distraction_indexes[new_distraction.id_distraction] = len(distractions)
|
|
distractions.append(new_distraction)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return distractions, errors
|
|
|
|
@classmethod
|
|
def save_distractions(cls, comment, distractions):
|
|
_m = f'{cls}.save_distractions'
|
|
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 distractions: {distractions}')
|
|
|
|
rows = []
|
|
for distraction in distractions:
|
|
row = Distraction_Temp.from_distraction(distraction)
|
|
row.guid = guid
|
|
rows.append(row)
|
|
|
|
Helper_App.console_log(f'rows: {rows}')
|
|
|
|
cls.upload_bulk(Distraction_Temp.__tablename__, rows, 1000)
|
|
|
|
Helper_App.console_log('distractions 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_distraction', argument_dict_list)
|
|
|
|
Helper_App.console_log('Distractions saved')
|
|
|
|
# Errors
|
|
cursor = result.cursor
|
|
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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_command_modality(cls, filters_command_modality):
|
|
_m = f'{cls.__qualname__}.get_many_command_modality'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_command_modality.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_command_modality', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Command_Modalities
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw command_modalities: {result_set_1}')
|
|
command_modalities = []
|
|
command_modality_indexes = {}
|
|
for row in result_set_1:
|
|
new_command_modality = Command_Modality.from_db_command_modality(row)
|
|
command_modality_indexes[new_command_modality.id_command_modality] = len(command_modalities)
|
|
command_modalities.append(new_command_modality)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return command_modalities, errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_bribe(cls, filters_bribe):
|
|
_m = f'{cls.__qualname__}.get_many_bribe'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_bribe.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_bribe', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Bribes
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw bribes: {result_set_1}')
|
|
bribes = []
|
|
bribe_indexes = {}
|
|
for row in result_set_1:
|
|
new_bribe = Bribe.from_db_bribe(row)
|
|
bribe_indexes[new_bribe.id_bribe] = len(bribes)
|
|
bribes.append(new_bribe)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return bribes, errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_assessment_command_modality_link(cls, filters_assessment_command_modality_link):
|
|
_m = f'{cls.__qualname__}.get_many_assessment_command_modality_link'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_assessment_command_modality_link.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_assessment_command_modality_link', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Assessment_Command_Modality_Links
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw assessment_command_modality_links: {result_set_1}')
|
|
assessment_command_modality_links = []
|
|
assessment_command_modality_link_indexes = {}
|
|
for row in result_set_1:
|
|
new_assessment_command_modality_link = Assessment_Command_Modality_Link.from_db_assessment_command_modality_link(row)
|
|
assessment_command_modality_link_indexes[new_assessment_command_modality_link.id_link] = len(assessment_command_modality_links)
|
|
assessment_command_modality_links.append(new_assessment_command_modality_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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return assessment_command_modality_links, errors
|
|
|
|
@classmethod
|
|
def save_assessment_command_modality_links(cls, comment, assessment_command_modality_links):
|
|
_m = f'{cls}.save_assessment_command_modality_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 assessment_command_modality_links: {assessment_command_modality_links}')
|
|
|
|
rows = []
|
|
for link in assessment_command_modality_links:
|
|
row = Assessment_Command_Modality_Link_Temp.from_assessment_command_modality_link(link)
|
|
row.guid = guid
|
|
rows.append(row)
|
|
|
|
Helper_App.console_log(f'rows: {rows}')
|
|
|
|
cls.upload_bulk(Assessment_Command_Modality_Link_Temp.__tablename__, rows, 1000)
|
|
|
|
Helper_App.console_log('assessment_command_modality_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_assessment_command_modality_link', argument_dict_list)
|
|
|
|
Helper_App.console_log('Assessment_Command_Modality_Links saved')
|
|
|
|
# Errors
|
|
cursor = result.cursor
|
|
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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_response_quality_metric(cls, filters_response_quality_metric):
|
|
_m = f'{cls.__qualname__}.get_many_response_quality_metric'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_response_quality_metric.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_response_quality_metric', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Command_Modalities
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw command_modalities: {result_set_1}')
|
|
command_modalities = []
|
|
response_quality_metric_indexes = {}
|
|
for row in result_set_1:
|
|
new_response_quality_metric = Response_Quality_Metric.from_db_response_quality_metric(row)
|
|
response_quality_metric_indexes[new_response_quality_metric.id_response_quality_metric] = len(command_modalities)
|
|
command_modalities.append(new_response_quality_metric)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return command_modalities, errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_obedience_level(cls, filters_obedience_level):
|
|
_m = f'{cls.__qualname__}.get_many_obedience_level'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_obedience_level.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_obedience_level', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Command_Modalities
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw command_modalities: {result_set_1}')
|
|
command_modalities = []
|
|
obedience_level_indexes = {}
|
|
for row in result_set_1:
|
|
new_obedience_level = Obedience_Level.from_db_obedience_level(row)
|
|
obedience_level_indexes[new_obedience_level.id_obedience_level] = len(command_modalities)
|
|
command_modalities.append(new_obedience_level)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return command_modalities, errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_assessment_response(cls, filters_assessment_response):
|
|
_m = f'{cls.__qualname__}.get_many_assessment_response'
|
|
user = cls.get_user_session()
|
|
argument_dict = {
|
|
'a_id_user': user.id_user
|
|
, **filters_assessment_response.to_json()
|
|
, 'a_debug': 0
|
|
}
|
|
Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
result = cls.db_procedure_execute('p_dog_get_many_assessment_response', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Assessment_Responses
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw assessment_responses: {result_set_1}')
|
|
assessment_responses = []
|
|
assessment_response_indexes = {}
|
|
for row in result_set_1:
|
|
new_assessment_response = Assessment_Response.from_db_assessment_response(row)
|
|
assessment_response_indexes[new_assessment_response.id_response] = len(assessment_responses)
|
|
assessment_responses.append(new_assessment_response)
|
|
|
|
# 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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return assessment_responses, errors
|
|
|
|
@classmethod
|
|
def save_assessment_responses(cls, comment, assessment_responses):
|
|
_m = f'{cls}.save_assessment_responses'
|
|
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 assessment_responses: {assessment_responses}')
|
|
|
|
rows = []
|
|
for response in assessment_responses:
|
|
row = Assessment_Response_Temp.from_assessment_response(response)
|
|
row.guid = guid
|
|
rows.append(row)
|
|
|
|
Helper_App.console_log(f'rows: {rows}')
|
|
|
|
cls.upload_bulk(Assessment_Response_Temp.__tablename__, rows, 1000)
|
|
|
|
Helper_App.console_log('assessment_responses 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_assessment_response', argument_dict_list)
|
|
|
|
Helper_App.console_log('Assessment_Responses saved')
|
|
|
|
# Errors
|
|
cursor = result.cursor
|
|
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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return errors
|
|
|
|
|
|
@classmethod
|
|
def save_assessments_distactions_and_responses(cls, comment, assessments, distractions, assessment_command_modality_links, assessment_responses):
|
|
_m = f'{cls}.save_assessments_distactions_and_responses'
|
|
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 assessments, distractions, and responses:\nAssessments: {assessments}\nDistractions: {distractions}\nAssessment Command Modality Links: {assessment_command_modality_links}\nResponses: {assessment_responses}')
|
|
|
|
rows_assessment = []
|
|
for assessment in assessments:
|
|
row = Assessment_Temp.from_assessment(assessment)
|
|
row.guid = guid
|
|
rows_assessment.append(row)
|
|
Helper_App.console_log(f'rows_assessment: {rows_assessment}')
|
|
cls.upload_bulk(Assessment_Temp.__tablename__, rows_assessment, 1000)
|
|
Helper_App.console_log('assessments uploaded')
|
|
|
|
rows_distraction = []
|
|
for distraction in distractions:
|
|
row = Distraction_Temp.from_distraction(distraction)
|
|
row.guid = guid
|
|
rows_distraction.append(row)
|
|
Helper_App.console_log(f'rows_distraction: {rows_distraction}')
|
|
cls.upload_bulk(Distraction_Temp.__tablename__, rows_distraction, 1000)
|
|
Helper_App.console_log('distractions uploaded')
|
|
|
|
rows_link = []
|
|
for link in assessment_command_modality_links:
|
|
row = Assessment_Command_Modality_Link_Temp.from_assessment_command_modality_link(link)
|
|
row.guid = guid
|
|
rows_link.append(row)
|
|
Helper_App.console_log(f'rows_link: {rows_link}')
|
|
cls.upload_bulk(Assessment_Command_Modality_Link_Temp.__tablename__, rows_link, 1000)
|
|
Helper_App.console_log('links uploaded')
|
|
|
|
rows_response = []
|
|
for response in assessment_responses:
|
|
row = Assessment_Response_Temp.from_assessment_response(response)
|
|
row.guid = guid
|
|
rows_response.append(row)
|
|
Helper_App.console_log(f'rows_response: {rows_response}')
|
|
cls.upload_bulk(Assessment_Response_Temp.__tablename__, rows_response, 1000)
|
|
Helper_App.console_log('assessment responses 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_assessment_distraction_and_response', argument_dict_list)
|
|
|
|
Helper_App.console_log('All saved')
|
|
|
|
# Errors
|
|
cursor = result.cursor
|
|
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)
|
|
while cursor.nextset():
|
|
Helper_App.console_log(f'unexpected result set: {cursor.fetchall()}')
|
|
|
|
return errors
|