Feat(SQL): Location Get Many and Calc Stored Procedures created with logic using Location Link table.
This commit is contained in:
@@ -17,6 +17,7 @@ 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.dog.location import Location
|
||||
from business_objects.sql_error import SQL_Error
|
||||
from datastores.datastore_base import DataStore_Base
|
||||
from helpers.helper_app import Helper_App
|
||||
@@ -72,6 +73,7 @@ class DataStore_Dog(DataStore_Base):
|
||||
|
||||
return dogs, errors
|
||||
|
||||
|
||||
@classmethod
|
||||
def get_many_command(cls, filters_command):
|
||||
_m = f'{cls.__qualname__}.get_many_command'
|
||||
@@ -124,6 +126,95 @@ class DataStore_Dog(DataStore_Base):
|
||||
|
||||
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('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
|
||||
|
||||
@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 get_many_dog_command_link(cls, filters_dog_command_link):
|
||||
_m = f'{cls.__qualname__}.get_many_dog_command_link'
|
||||
@@ -205,94 +296,41 @@ class DataStore_Dog(DataStore_Base):
|
||||
|
||||
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()
|
||||
def get_many_location(cls, filters_location):
|
||||
_m = f'{cls.__qualname__}.get_many_location'
|
||||
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
|
||||
argument_dict = {
|
||||
'a_id_user': user.id_user
|
||||
, **filters_location.to_json()
|
||||
, 'a_debug': 0
|
||||
}
|
||||
result = cls.db_procedure_execute('p_dog_save_command', argument_dict_list)
|
||||
|
||||
Helper_App.console_log('Commands saved')
|
||||
|
||||
# Errors
|
||||
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)
|
||||
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
|
||||
|
||||
|
||||
|
||||
return locations, errors
|
||||
|
||||
Reference in New Issue
Block a user