admin/dog_training
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Feat(SQL, UI): Redesign database with much more detailed command response quality analysis and created successfully loading Dog Command Links page

Browse Source
This commit is contained in:
Teddy Middleton-Smith
2025-06-28 20:48:37 +01:00
parent ab50a81a0e
commit caeb13429a
245 changed files with 7244 additions and 2035 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
datastores/datastore_base.py
View File

@@ -12,15 +12,13 @@ Datastore for Store
# internal
# from routes import bp_home
import lib.argument_validation as av
from business_objects.sql_error import SQL_Error
from business_objects.project_hub.user import User
# 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
from forms.access_level import Filters_Access_Level
from forms.unit_measurement import Filters_Unit_Measurement
from helpers.helper_app import Helper_App
import dog_training.lib.argument_validation as av
from dog_training.business_objects.sql_error import SQL_Error
from dog_training.business_objects.dog.user import User
# from dog_training.helpers.helper_db_mysql import Helper_DB_MySQL
# from dog_training.models.model_view_store_checkout import Model_View_Store_Checkout # circular!
from dog_training.extensions import db
from dog_training.helpers.helper_app import Helper_App
# external
from sqlalchemy import text
from flask import Flask, session, current_app

201
datastores/datastore_dog.py Normal file
View File

@@ -0,0 +1,201 @@
"""
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 dog_training.lib.argument_validation as av
from dog_training.business_objects.dog.command import Command, Command_Temp
from dog_training.business_objects.dog.dog import Dog
from dog_training.business_objects.dog.dog_command_link import Dog_Command_Link
from dog_training.business_objects.sql_error import SQL_Error
from dog_training.datastores.datastore_base import DataStore_Base
from dog_training.helpers.helper_app import Helper_App
from dog_training.helpers.helper_db_mysql import Helper_DB_MySQL
# from dog_training.models.model_view_store_checkout import Model_View_Store_Checkout # circular!
from dog_training.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
, 'a_get_all_dog': filters_dog.get_all_dog
, 'a_get_inactive_dog': filters_dog.get_inactive_dog
, 'a_ids_dog': filters_dog.ids_dog
, 'a_names_dog': filters_dog.names_dog
, '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
, 'a_get_all_command': filters_command.get_all_command
, 'a_get_inactive_command': filters_command.get_inactive_command
, 'a_ids_command': filters_command.ids_command
, 'a_names_command': filters_command.names_command
, '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
# Commands
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)
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 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
, 'a_get_all_dog': filters_dog_command_link.get_all_dog
, 'a_get_inactive_dog': filters_dog_command_link.get_inactive_dog
, 'a_ids_dog': filters_dog_command_link.ids_dog
, 'a_get_all_command': filters_dog_command_link.get_all_command
, 'a_get_inactive_command': filters_dog_command_link.get_inactive_command
, 'a_ids_command': filters_dog_command_link.ids_command
, '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:
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_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

119
datastores/datastore_user.py Normal file
View File

@@ -0,0 +1,119 @@
"""
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.sql_error import SQL_Error
from business_objects.dog.user import User, Parameters_User, User_Permission_Evaluation
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 abc import ABC, abstractmethod, abstractproperty
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text
import stripe
import os
from flask import Flask, session, current_app
from pydantic import BaseModel, ConfigDict
from typing import ClassVar
from datetime import datetime
db = SQLAlchemy()
class DataStore_User(DataStore_Base):
def __init__(self):
super().__init__()
def edit_user(self):
_m = 'DataStore_User.edit_user'
# guid = Helper_DB_MySQL.create_guid()
argument_dict_list = {
'a_id_user': self.info_user.get('sub'),
'a_name': self.info_user.get('name'),
'a_email': self.info_user.get('email'),
'a_email_verified': 1 if self.info_user.get('email_verified') == 'True' else 0
}
result = self.db_procedure_execute('p_dog_save_user', argument_dict_list)
cursor = result.cursor
result_set_1 = cursor.fetchall()
Helper_App.console_log(f'raw user data: {result_set_1}')
# Errors
cursor.nextset()
result_set_e = cursor.fetchall()
Helper_App.console_log(f'raw errors: {result_set_e}')
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}")
DataStore_User.db_cursor_clear(cursor)
return (result_set_1[0][1] == b'\x01')
def get_many_user(self, user_filters, user=None):
_m = 'DataStore_User.get_many_user'
Helper_App.console_log(_m)
Helper_App.console_log(f'user_filters: {user_filters}')
Helper_App.console_log(f"valid user_filters: {av.val_instance(user_filters, 'user_filters', _m, Parameters_User)}")
# guid = Helper_DB_MySQL.create_guid()
Helper_App.console_log(f'user: {user}')
if user is None:
user = self.get_user_session()
Helper_App.console_log(f'user: {user}')
argument_dict_list = {
'a_id_user': user.id_user
, 'a_id_user_auth0': user.id_user_auth0
, **user_filters.to_json()
, 'a_debug': 0
}
Helper_App.console_log(f'argument_dict_list: {argument_dict_list}')
result = self.db_procedure_execute('p_dog_get_many_user', argument_dict_list)
cursor = result.cursor
result_set = cursor.fetchall()
Helper_App.console_log(f'raw users: {result_set}')
Helper_App.console_log(f'type result set: {str(type(result_set))}')
Helper_App.console_log(f'len result set: {len(result_set)}')
users = []
if len(result_set) > 0:
for row in result_set:
Helper_App.console_log(f'row: {row}')
user = User.from_DB_user(row)
users.append(user)
Helper_App.console_log(f'user {str(type(user))}: {user}')
Helper_App.console_log(f'type users: {str(type(users))}\n type user 0: {str(type(None if len(users) == 0 else users[0]))}')
"""
errors = []
cursor.nextset()
result_set_e = cursor.fetchall()
Helper_App.console_log(f'raw errors: {result_set_e}')
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}")
"""
errors = None
DataStore_User.db_cursor_clear(cursor)
return users, errors

113
datastores/project_hub/datastore_contact_form.py
View File

@@ -1,113 +0,0 @@
"""
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.sql_error import SQL_Error
from business_objects.project_hub.command import Command, Command_Temp
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_Command(DataStore_Base):
def __init__(self):
super().__init__()
@classmethod
def get_many_command(cls):
_m = f'{cls.__qualname__}.get_many_command'
user = cls.get_user_session()
argument_dict = {
'a_id_user': user.id_user
, 'a_debug': 0
}
Helper_App.console_log(f'argument_dict: {argument_dict}')
result = cls.db_procedure_execute('p_ph_get_many_command', argument_dict)
cursor = result.cursor
# Commands
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)
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 commands, 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_ph_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