134 lines
4.6 KiB
Python
134 lines
4.6 KiB
Python
"""
|
|
Project: Magic Tracker
|
|
Author: Edward Middleton-Smith
|
|
Shuffle & Skirmish
|
|
|
|
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.tcg.user import User, Parameters_User
|
|
from business_objects.tcg.user_relationship import User_Relationship, Parameters_User_Relationship
|
|
from datastores.datastore_base import DataStore_Base
|
|
from helpers.helper_app import Helper_App
|
|
from helpers.helper_db_sql import Helper_DB_SQL
|
|
# 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__()
|
|
|
|
@classmethod
|
|
def get_many_user(cls, user_filters):
|
|
_m = f'{cls.__qualname__}.get_many_user'
|
|
# Helper_App.console_log(_m)
|
|
# Helper_App.console_log(f'user_filters: {user_filters}')
|
|
av.val_instance(user_filters, 'user_filters', _m, Parameters_User)
|
|
|
|
argument_dict = user_filters.to_json()
|
|
# Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
|
|
users = []
|
|
errors = []
|
|
try:
|
|
user_result = cls.db_function_execute('tcg.public.FN_TCG_User_Get_Many', argument_dict, Parameters_User.get_type_hints())
|
|
user_result_set = user_result.fetchall()
|
|
# Helper_App.console_log(f'raw users: {user_result_set}')
|
|
|
|
for row in user_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}')
|
|
except Exception as e:
|
|
# Helper_App.console_log(f'Error: {str(e)}')
|
|
error = SQL_Error()
|
|
error.msg = str(e)
|
|
errors.append(error)
|
|
|
|
return users, errors
|
|
|
|
@classmethod
|
|
def get_many_user_relationship(cls, relationship_filters):
|
|
_m = f'{cls.__qualname__}.get_many_user_relationship'
|
|
av.val_instance(relationship_filters, 'relationship_filters', _m, Parameters_User_Relationship)
|
|
|
|
argument_dict = relationship_filters.to_json()
|
|
|
|
relationships = []
|
|
errors = []
|
|
try:
|
|
result = cls.db_function_execute('tcg.public.FN_TCG_User_Relationship_Get_Many', argument_dict, Parameters_User_Relationship.get_type_hints())
|
|
result_set = result.fetchall()
|
|
for row in result_set:
|
|
relationship = User_Relationship.from_db_user_relationship(row)
|
|
relationships.append(relationship)
|
|
except Exception as e:
|
|
error = SQL_Error()
|
|
error.msg = str(e)
|
|
errors.append(error)
|
|
|
|
return relationships, errors
|
|
|
|
@classmethod
|
|
def login_user(cls, user):
|
|
_m = f'{cls}.login_user'
|
|
# av.val_str(comment, 'comment', _m)
|
|
|
|
guid = Helper_DB_SQL.create_guid_str()
|
|
|
|
# Helper_App.console_log(f'login user: {user}')
|
|
|
|
success = False
|
|
user_id = None
|
|
try:
|
|
argument_dict_list = {
|
|
'a_user_auth0_id': user.user_auth0_id
|
|
, 'a_email': user.email
|
|
, 'a_guid': guid
|
|
, 'a_is_email_verified': user.is_email_verified
|
|
, 'o_success': success
|
|
, 'o_user_id': user_id
|
|
}
|
|
rows = cls.db_procedure_execute('tcg.public.USP_TCG_User_Login', argument_dict_list)
|
|
row = rows[0] if rows else None
|
|
success = row['o_success'] if row else False
|
|
user_id = row['o_user_id'] if row else None
|
|
# Helper_App.console_log('User logged in')
|
|
# Helper_App.console_log(f'Success: {success}\nUser ID: {user_id}')
|
|
|
|
user.user_id = user_id
|
|
user_filters = Parameters_User.from_user(
|
|
user = user
|
|
, user_id_session = user_id
|
|
)
|
|
users, errors = cls.get_many_user(user_filters = user_filters)
|
|
user = users[0] if len(users) > 0 else None
|
|
if user is not None:
|
|
user.is_logged_in = True
|
|
return user, errors
|
|
except Exception as e:
|
|
return None, SQL_Error.from_exception(e)
|
|
|