159 lines
5.2 KiB
Python
159 lines
5.2 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.sql_error import SQL_Error
|
|
from business_objects.project_hub.apply_founding_partner_form import Apply_Founding_Partner_Form, Apply_Founding_Partner_Form_Temp
|
|
from business_objects.project_hub.contact_form import Contact_Form, Contact_Form_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_Project_Hub_Contact_Form(DataStore_Base):
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
@classmethod
|
|
def get_many_contact_form(cls):
|
|
_m = f'{cls.__qualname__}.get_many_contact_form'
|
|
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_contact_form', argument_dict)
|
|
cursor = result.cursor
|
|
|
|
# Contact_Forms
|
|
result_set_1 = cursor.fetchall()
|
|
Helper_App.console_log(f'raw contact_forms: {result_set_1}')
|
|
contact_forms = []
|
|
contact_form_indexes = {}
|
|
for row in result_set_1:
|
|
new_contact_form = Contact_Form.from_DB_contact_form(row)
|
|
contact_form_indexes[new_contact_form.id_contact_form] = len(contact_forms)
|
|
contact_forms.append(new_contact_form)
|
|
|
|
# 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 contact_forms, errors
|
|
|
|
@classmethod
|
|
def save_contact_forms(cls, comment, contact_forms):
|
|
_m = f'{cls}.save_contact_forms'
|
|
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 contact forms: {contact_forms}')
|
|
|
|
rows = []
|
|
for contact_form in contact_forms:
|
|
row = Contact_Form_Temp.from_contact_form(contact_form)
|
|
row.guid = guid
|
|
rows.append(row)
|
|
|
|
cls.upload_bulk(Contact_Form_Temp.__tablename__, rows, 1000)
|
|
|
|
Helper_App.console_log('Contact Forms 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_contact_form', argument_dict_list)
|
|
|
|
Helper_App.console_log('Contact Forms 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_apply_founding_partner_forms(cls, comment, apply_founding_partner_forms):
|
|
_m = f'{cls}.save_apply_founding_partner_forms'
|
|
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 contact forms: {apply_founding_partner_forms}')
|
|
|
|
rows = []
|
|
for apply_founding_partner_form in apply_founding_partner_forms:
|
|
row = Apply_Founding_Partner_Form_Temp.from_apply_founding_partner_form(apply_founding_partner_form)
|
|
row.guid = guid
|
|
rows.append(row)
|
|
|
|
cls.upload_bulk(Apply_Founding_Partner_Form_Temp.__tablename__, rows, 1000)
|
|
|
|
Helper_App.console_log('Contact Forms 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_apply_founding_partner_form', argument_dict_list)
|
|
|
|
Helper_App.console_log('Contact Forms 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
|