Scott.Idem/OSIT-AE-API-FastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Working on all module routes, methods, and models

Browse Source
This commit is contained in:
Scott Idem
2021-06-22 18:00:34 -04:00
parent b6ca0eb3f2
commit c1d23d15f4
20 changed files with 653 additions and 281 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
app/methods/account_methods.py
View File

@@ -117,7 +117,7 @@ def load_account_obj(
inc_user_list: bool = False, # Priority l2
inc_user_role_list: bool = False,
) -> Account_Base|dict|bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
@@ -291,7 +291,7 @@ def load_account_obj(
inc_journal_entry_list = inc_journal_entry_list,
)
)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(journal_dict_list)
@@ -366,6 +366,7 @@ def load_account_obj(
load_membership_type_obj(
membership_type_id = membership_type_rec.get('membership_type_id', None),
limit = limit,
enabled = enabled,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,

2
app/methods/event_location_methods.py
View File

@@ -14,8 +14,6 @@ from app.lib_general import log, logging
# from app.methods.event_presenter_methods import load_event_presenter_obj
# from app.methods.event_session_methods import load_event_session_obj
# from app.methods.event_track_methods import load_event_track_obj
# from app.methods.person_methods import load_person_obj
# from app.methods.user_methods import load_user_obj
from app.models.event_location_models import Event_Location_Base

6
app/methods/membership_member_methods.py
View File

@@ -281,8 +281,8 @@ def load_membership_member_obj(
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
inc_address = inc_address,
inc_contact = inc_contact,
# inc_address = inc_address,
# inc_contact = inc_contact,
# inc_organization = inc_organization,
# inc_person = inc_person,
):
@@ -304,7 +304,7 @@ def get_membership_member_rec_list(
limit: int = 1000,
enabled: str = 'enabled', # enabled, disabled, all
) -> list|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if for_obj_id := redis_lookup_id_random(record_id_random=for_obj_id, table_name='for_obj_type'): pass

44
app/methods/order_cfg_methods.py Normal file
View File

@@ -0,0 +1,44 @@
from __future__ import annotations
import datetime
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from app.db_sql import redis_lookup_id_random, sql_select
from app.lib_general import log, logging
from app.models.order_cfg_models import Order_Cfg_Base
# ### BEGIN ### API Order Cfg Methods ### load_order_cfg_obj() ###
def load_order_cfg_obj(
account_id: int|str,
by_alias: bool = True,
exclude_unset: bool = True,
model_as_dict: bool = False,
) -> Order_Cfg_Base|dict|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return False
if order_cfg_rec := sql_select(
table_name = 'v_account_cfg_detail',
field_name = 'account_id',
field_value = account_id,
): pass
else: return False
log.debug(order_cfg_rec)
try:
order_cfg_obj = Order_Cfg_Base(**order_cfg_rec)
log.debug(order_cfg_obj)
except ValidationError as e:
log.error(e.json())
if model_as_dict:
return order_cfg_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
else:
return order_cfg_obj
# ### END ### API Order Cfg Methods ### load_order_cfg_obj() ###

185
app/methods/order_line_methods.py Normal file
View File

@@ -0,0 +1,185 @@
from __future__ import annotations
import datetime
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from app.db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
from app.lib_general import log, logging
from app.models.order_line_models import Order_Line_Base
# ### BEGIN ### API Order Line Methods ### create_order_line_obj() ###
def create_order_line_obj(order_line_obj_new:Order_Line_Base):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not order_line_obj_new:
return False
order_line_obj_data = order_line_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'user', 'created_on', 'updated_on'})
if order_line_obj_in_result := sql_insert(data=order_line_obj_data, table_name='order_line', rm_id_random=True, id_random_length=8): pass
else:
return False
log.debug(order_line_obj_in_result)
order_line_id = order_line_obj_in_result
log.debug(f'Returning the new order_line_id: {order_line_id}')
return order_line_id
# ### END ### API Order Line Methods ### create_order_line_obj() ###
# ### BEGIN ### API Order Line Methods ### load_order_line_obj() ###
def load_order_line_obj(
order_line_id: int|str,
limit: int = 1000,
by_alias: bool = True,
exclude_unset: bool = True,
model_as_dict: bool = False,
) -> Order_Line_Base|dict|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if order_line_id := redis_lookup_id_random(record_id_random=order_line_id, table_name='order_line'): pass
else: return False
if order_line_rec := sql_select(table_name='v_order_line', record_id=order_line_id): pass
else: return False
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_line_rec)
try:
order_line_obj = Order_Line_Base(**order_line_rec)
log.debug(order_line_obj)
except ValidationError as e:
log.error(e.json())
return False
if model_as_dict:
return order_line_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
else:
return order_line_obj
# ### END ### API Order Line Methods ### load_order_line_obj() ###
# ### BEGIN ### API Order Line Methods ### update_order_line_obj() ###
def update_order_line_obj(
order_line_id: int|str, # Ideally the int ID should be passed. This allows for updating of the id_random value.
order_line_obj_up: Order_Line_Base,
create_missing_obj: bool = False,
) -> bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if order_line_id := redis_lookup_id_random(record_id_random=order_line_id, table_name='order_line'): pass
else: return False
order_line_obj_up.id = order_line_id
log.debug(order_line_obj_up)
# log.debug(order_line_obj_up.dict(by_alias=True, exclude_unset=True))
log.debug(order_line_obj_up.dict(by_alias=False, exclude_unset=True))
# log.debug(order_line_obj_up.dict(by_alias=False, exclude_unset=False))
#order_line_dict_up = order_line_obj_up.dict(by_alias=False, exclude_unset=True)
# if order_line_obj_up.person_id and order_line_obj_up.person:
# person_id = order_line_obj_up.person_id
# person_obj_up = order_line_obj_up.person
# log.debug(person_id)
# log.debug(person_obj_up)
# if person_obj_up_result := update_person_obj(
# person_id=person_id,
# person_obj_up=person_obj_up,
# create_missing_obj=create_missing_obj,
# ):
# log.debug(person_obj_up_result)
# else:
# log.debug(person_obj_up_result)
# return False
# if order_line_obj_up.user_id and order_line_obj_up.user:
# user_id = order_line_obj_up.user_id
# user_obj_up = order_line_obj_up.user
# log.debug(user_id)
# log.debug(user_obj_up)
# if user_obj_up_result := update_user_obj(
# user_id=user_id,
# user_obj_up=user_obj_up,
# create_missing_obj=create_missing_obj,
# ):
# log.debug(user_obj_up_result)
# else:
# log.debug(user_obj_up_result)
# return False
order_line_dict_up = order_line_obj_up.dict(by_alias=False, exclude_unset=True, exclude={'user'})
log.debug(order_line_dict_up)
if order_line_obj_up_result := sql_update(data=order_line_dict_up, table_name='order_line', rm_id_random=True):
log.debug(order_line_obj_up_result)
return True
else:
log.debug(order_line_obj_up_result)
return False
# ### END ### API Order Line Methods ### update_order_line_obj() ###
# ### BEGIN ### API Order Line Methods ### get_order_line_rec_list() ###
def get_order_line_rec_list(
for_obj_type: str,
for_obj_id: str,
limit: int = 1000,
) -> list|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if for_obj_id := redis_lookup_id_random(record_id_random=for_obj_id, table_name='for_obj_type'): pass
else: return False
data = {}
data[f'{for_obj_type}_id'] = for_obj_id
# data['for_obj_type'] = for_obj_type
sql_obj_type_id = f'`tbl`.{for_obj_type}_id = :{for_obj_type}_id'
# if enabled in ['enabled', 'disabled', 'all']:
# if enabled == 'enabled':
# data['enable'] = True
# sql_enabled = f'AND `tbl`.enable = :enable'
# elif enabled == 'disabled':
# data['enable'] = False
# sql_enabled = f'AND `tbl`.enable = :enable'
# elif enabled == 'all':
# sql_enabled = ''
sql_enabled = ''
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql = f"""
SELECT `tbl`.id AS 'order_line_id', `tbl`.id_random AS 'order_line_id_random'
FROM `order_line` AS `tbl`
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if order_line_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
order_line_rec_li = order_line_rec_li_result
else:
order_line_rec_li = []
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_line_rec_li_result)
return order_line_rec_li
# ### END ### API Order Line Methods ### get_order_line_rec_list() ###

119
app/methods/order_methods.py
View File

@@ -7,10 +7,15 @@ from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, v
from app.db_sql import redis_lookup_id_random, sql_delete, sql_insert_or_update, sql_select
from app.lib_general import log, logging
from app.methods.order_cfg_methods import load_order_cfg_obj
from app.methods.order_line_methods import get_order_line_rec_list, load_order_line_obj
# from app.methods.person_methods import load_person_obj
# from app.methods.user_methods import load_user_obj
from app.models.order_models import Order_Base
from app.models.order_line_models import Order_Line_Base # This should go away later.
#from app.models.person_models import Person_Base
#from app.models.user_models import User_Base
# from app.models.person_models import Person_Base
# from app.models.user_models import User_Base
# ### BEGIN ### API Order Methods ### save_order_obj() ###
@@ -165,46 +170,22 @@ def load_order_obj(
exclude_unset: bool = True,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_order_line_list: bool = False,
inc_order_cfg: bool = False,
inc_person_obj: bool = False,
inc_user_obj: bool = False,
inc_order_line_list: bool = False,
inc_person: bool = False,
inc_user: bool = False,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if order_id := redis_lookup_id_random(record_id_random=order_id, table_name='order'): pass
else:
return False
else: return False
if order_rec := sql_select(table_name='v_order', record_id=order_id):
#log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_rec)
if order_rec := sql_select(table_name='v_order', record_id=order_id): pass
else: return False
if inc_order_line_list:
order_line_data = {}
order_line_data['order_id'] = order_id
if order_line_rec_li := sql_select(table_name='v_order_line', data=order_line_data, as_list=True):#, field_name='order_id', field_value=order_id):
log.debug(order_line_rec_li)
order_rec['order_line_li'] = order_line_rec_li
if inc_order_cfg:
if order_cfg_rec := sql_select(table_name='v_account_cfg_detail', field_name='account_id', field_value=order_rec.get('account_id', None)):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_cfg_rec)
order_rec['cfg'] = order_cfg_rec
if inc_person_obj:
if person_rec := sql_select(table_name='v_person', field_name='person_id', field_value=order_rec.get('person_id', None)):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_rec)
order_rec['person'] = person_rec
log.debug(order_rec)
else:
return False
#log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_rec)
try:
order_obj = Order_Base(**order_rec)
@@ -212,6 +193,76 @@ def load_order_obj(
except ValidationError as e:
log.error(e.json())
# Updated 2021-06-18
if inc_order_line_list:
if order_line_rec_list_result := get_order_line_rec_list(
for_obj_type = 'order',
for_obj_id = order_id,
limit = limit,
):
order_line_result_list = []
for order_line_rec in order_line_rec_list_result:
order_line_result_list.append(
load_order_line_obj(
order_line_id = order_line_rec.get('order_line_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
)
)
order_obj.order_line_list = order_line_result_list
else: order_obj.order_line_list = []
if inc_order_line_list:
order_line_data = {}
order_line_data['order_id'] = order_id
if order_line_rec_li := sql_select(table_name='v_order_line', data=order_line_data, as_list=True):#, field_name='order_id', field_value=order_id):
log.debug(order_line_rec_li)
order_rec['order_line_li'] = order_line_rec_li
if inc_order_cfg:
if order_cfg_result := load_order_cfg_obj(
order_cfg_id = order_rec.get('order_cfg_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
enabled = enabled,
):
order_obj.order_cfg = order_cfg_result
else: order_obj.order_cfg = None
# Updated 2021-06-22
if inc_person:
from app.methods.person_methods import load_person_obj
if person_result := load_person_obj(
person_id = order_rec.get('person_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
enabled = enabled,
):
order_obj.person = person_result
else: order_obj.person = None
pass
# Updated 2021-06-22
if inc_user:
from app.methods.user_methods import load_user_obj
if user_result := load_user_obj(
user_id = order_rec.get('user_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
enabled = enabled,
):
order_obj.user = user_result
else: order_obj.user = None
pass
if model_as_dict:
return order_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
else:

76
app/methods/person_methods.py
View File

@@ -9,8 +9,9 @@ from app.lib_general import log, logging
from app.methods.address_methods import load_address_obj
from app.methods.contact_methods import create_contact_obj, load_contact_obj, update_contact_obj
from app.methods.order_methods import load_order_obj, get_order_rec_list
from app.methods.organization_methods import load_organization_obj, update_organization_obj
from app.methods.user_methods import create_user_obj, load_user_obj, update_user_obj
# from app.methods.user_methods import create_user_obj, load_user_obj, update_user_obj
from app.models.person_models import Person_Base
@@ -69,7 +70,17 @@ def load_person_obj(
enabled: str = 'enabled', # enabled, disabled, all
inc_address: bool = False, # Under contact
inc_contact: bool = False,
inc_event_list: bool = False,
inc_journal_list: bool = False,
inc_journal_entry_list: bool = False,
inc_membership_member: bool = False,
inc_order_cfg: bool = False,
inc_order_line_list: bool = False,
inc_order_list: bool = False,
inc_order_cart_list: bool = False,
inc_organization: bool = False,
inc_post_list: bool = False,
inc_post_comment_list: bool = False,
inc_user: bool = False,
) -> Person_Base|bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
@@ -103,6 +114,32 @@ def load_person_obj(
person_obj.contact = contact_result
else: person_obj.contact = None
# Updated 2021-06-18
if inc_order_list:
if order_rec_list_result := get_order_rec_list(
for_obj_type = 'person',
for_obj_id = person_id,
limit = limit,
enabled = enabled,
):
order_result_list = []
for order_rec in order_rec_list_result:
order_result_list.append(
load_order_obj(
order_id = order_rec.get('order_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
enabled = enabled,
inc_order_cfg = inc_order_cfg,
inc_order_line_list = inc_order_line_list,
inc_user = inc_user,
)
)
person_obj.order_list = order_result_list
else: person_obj.order_list = []
# Updated 2021-06-18
if inc_organization:
organization_id = person_rec.get('organization_id', None)
@@ -122,18 +159,19 @@ def load_person_obj(
# Updated 2021-06-18
if inc_user:
user_id = person_rec.get('user_id', None)
log.debug(user_id)
if user_result := load_user_obj(
user_id = user_id,
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
enabled = enabled,
):
person_obj.user = user_result
else: person_obj.user = None
# user_id = person_rec.get('user_id', None)
# log.debug(user_id)
# if user_result := load_user_obj(
# user_id = user_id,
# limit = limit,
# by_alias = by_alias,
# exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
# enabled = enabled,
# ):
# person_obj.user = user_result
# else: person_obj.user = None
pass
if model_as_dict:
return person_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
@@ -167,9 +205,9 @@ def update_person_obj(
log.debug(contact_id)
log.debug(contact_obj_up)
if contact_obj_up_result := update_contact_obj(
contact_id=contact_id,
contact_obj_up=contact_obj_up,
create_missing_obj=create_missing_obj,
contact_id = contact_id,
contact_obj_up = contact_obj_up,
create_missing_obj = create_missing_obj,
):
log.debug(contact_obj_up_result)
else:
@@ -194,9 +232,9 @@ def update_person_obj(
log.debug(organization_id)
log.debug(organization_obj_up)
if organization_obj_up_result := update_organization_obj(
organization_id=organization_id,
organization_obj_up=organization_obj_up,
create_missing_obj=create_missing_obj,
organization_id = organization_id,
organization_obj_up = organization_obj_up,
create_missing_obj = create_missing_obj,
):
log.debug(organization_obj_up_result)
else:

50
app/methods/post_comment_methods.py
View File

@@ -7,7 +7,7 @@ from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, v
from app.db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
from app.lib_general import log, logging
# from app.methods.person_methods import load_person_obj
from app.methods.person_methods import load_person_obj
# from app.methods.user_methods import load_user_obj
from app.models.post_comment_models import Post_Comment_Base
@@ -66,33 +66,35 @@ def load_post_comment_obj(
log.error(e.json())
return False
# Updated 2021-06-18
# Updated 2021-06-22
if inc_person:
from app.methods.person_methods import load_person_obj
person_result = load_person_obj(
person_id = post_comment_rec.get('person_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
enabled = enabled,
)
post_comment_obj.person = person_result
else: post_comment_obj.person = None
# from app.methods.person_methods import load_person_obj
if person_result := load_person_obj(
person_id = post_comment_rec.get('person_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
enabled = enabled,
):
post_comment_obj.person = person_result
else: post_comment_obj.person = None
# Updated 2021-06-18
if inc_user:
from app.methods.user_methods import load_user_obj
user_result = load_user_obj(
user_id = post_comment_rec.get('user_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
enabled = enabled,
)
post_comment_obj.user = user_result
else: post_comment_obj.user = None
# from app.methods.user_methods import load_user_obj
# user_result = load_user_obj(
# user_id = post_comment_rec.get('user_id', None),
# limit = limit,
# by_alias = by_alias,
# exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
# enabled = enabled,
# )
# post_comment_obj.user = user_result
pass
if model_as_dict:
return post_comment_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member

132
app/methods/post_methods.py
View File

@@ -7,7 +7,7 @@ from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, v
from app.db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
from app.lib_general import log, logging
# from app.methods.person_methods import load_person_obj
from app.methods.person_methods import load_person_obj
# from app.methods.post_comment_methods import create_post_comment_obj, update_post_comment_obj
from app.methods.post_comment_methods import get_post_comment_rec_list, load_post_comment_obj
# from app.methods.user_methods import load_user_obj
@@ -70,19 +70,19 @@ def load_post_obj(
except ValidationError as e:
log.error(e.json())
# Updated 2021-06-18
# Updated 2021-06-22
if inc_person:
from app.methods.person_methods import load_person_obj
person_result = load_person_obj(
person_id = post_rec.get('person_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
enabled = enabled,
)
post_obj.person = person_result
else: post_obj.person = None
# from app.methods.person_methods import load_person_obj
if person_result := load_person_obj(
person_id = post_rec.get('person_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
enabled = enabled,
):
post_obj.person = person_result
else: post_obj.person = None
# Updated 2021-06-17
if inc_post_comment_list:
@@ -111,17 +111,17 @@ def load_post_obj(
# Updated 2021-06-18
if inc_user:
from app.methods.user_methods import load_user_obj
user_result = load_user_obj(
user_id = post_rec.get('user_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
enabled = enabled,
)
post_obj.user = user_result
else: post_obj.user = None
# from app.methods.user_methods import load_user_obj
# user_result = load_user_obj(
# user_id = post_rec.get('user_id', None),
# limit = limit,
# by_alias = by_alias,
# exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
# enabled = enabled,
# )
# post_obj.user = user_result
pass
if model_as_dict:
return post_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
@@ -130,90 +130,6 @@ def load_post_obj(
# ### END ### API Post Methods ### load_post_obj() ###
# ### BEGIN ### API Post Methods ### load_post_obj_list() ###
def load_post_obj_list(
account_id: int|str|None = None,
user_id: int|str|None = None,
limit: int = 1000,
by_alias: bool = True,
exclude_unset: bool = True,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_post_comment_list: bool = False,
inc_person: bool = False,
inc_user: bool = False,
) -> list|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
data = {}
if account_id:
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return False
data['account_id'] = account_id
sql_obj_type_id = f'`tbl`.account_id = :account_id'
elif user_id:
if user_id := redis_lookup_id_random(record_id_random=user_id, table_name='user'): pass
else: return False
data['user_id'] = user_id
sql_obj_type_id = f'`tbl`.user_id = :user_id'
if enabled in ['enabled', 'disabled', 'all']:
if enabled == 'enabled':
data['enable'] = True
sql_enabled = f'AND `tbl`.enable = :enable'
elif enabled == 'disabled':
data['enable'] = False
sql_enabled = f'AND `tbl`.enable = :enable'
elif enabled == 'all':
sql_enabled = ''
# else: tbl_obj['account'] = None
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql = f"""
SELECT `tbl`.id AS 'post_id', `tbl`.id_random AS 'post_id_random'
FROM `post` AS `tbl`
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if post_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(post_rec_li_result)
post_result_li = []
for post_rec in post_rec_li_result:
post_id = post_rec.get('post_id', None)
if post_result := load_post_obj(
post_id = post_id,
model_as_dict = model_as_dict,
enabled = enabled,
inc_post_comment_list = inc_post_comment_list,
inc_person = inc_person,
inc_user = inc_user,
):
log.debug(post_result)
post_result_li.append(post_result)
else:
log.debug(post_result)
post_result_li.append(None)
log.debug(post_result_li)
else:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(post_rec_li_result)
post_result_li = []
return post_result_li
# ### END ### API Post Methods ### load_post_obj_list() ###
# ### BEGIN ### API Post Methods ### update_post_obj() ###
def update_post_obj(
post_id: int|str, # Ideally the int ID should be passed. This allows for updating of the id_random value.

5
app/methods/site_methods.py
View File

@@ -7,9 +7,7 @@ from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, v
from app.db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
from app.lib_general import log, logging
# from app.methods.person_methods import load_person_obj
from app.methods.site_domain_methods import get_site_domain_rec_list, load_site_domain_obj
# from app.methods.user_methods import load_user_obj
from app.models.site_models import Site_Base
@@ -57,8 +55,7 @@ def load_site_obj(
if site_id := redis_lookup_id_random(record_id_random=site_id, table_name='site'): pass
else: return False
if site_rec := sql_select(table_name='v_site', record_id=site_id):
log.debug(site_rec)
if site_rec := sql_select(table_name='v_site', record_id=site_id): pass
else: return False
log.debug(site_rec)

69
app/methods/user_methods.py
View File

@@ -9,8 +9,9 @@ from app.lib_general import log, logging
from app.methods.contact_methods import load_contact_obj, update_contact_obj
# from app.methods.event_methods import get_event_rec_list
from app.methods.order_methods import load_order_obj, get_order_rec_list
from app.methods.organization_methods import load_organization_obj, update_organization_obj
# from app.methods.person_methods import load_person_obj, update_person_obj
from app.methods.person_methods import load_person_obj, update_person_obj
from app.methods.post_methods import get_post_rec_list, load_post_obj
from app.models.user_models import User_Base, User_New_Base, User_Out_Base
@@ -80,14 +81,20 @@ def load_user_obj(
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_address: bool = False,
# inc_archive_list: bool = False,
inc_contact: bool = False,
inc_event_list: bool = False,
# inc_hosted_file_list: bool = False,
inc_journal_list: bool = False,
inc_membership: bool = False,
inc_journal_entry_list: bool = False,
inc_membership_member: bool = False,
inc_order_cfg: bool = False,
inc_order_line_list: bool = False,
inc_order_list: bool = False,
inc_order_cart_list: bool = False,
inc_organization: bool = False,
inc_person: bool = False,
# inc_person_list: bool = False,
inc_post_list: bool = False,
inc_post_comment_list: bool = False,
inc_user_role_list: bool = False,
@@ -112,20 +119,20 @@ def load_user_obj(
return False
# Updated 2021-06-18
if inc_contact:
contact_id = user_rec.get('contact_id', None)
log.debug(contact_id)
if contact_result := load_contact_obj(
contact_id = contact_id,
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
enabled = enabled,
inc_address = inc_address,
):
user_obj.contact = contact_result
else: user_obj.contact = None
# if inc_contact:
# contact_id = user_rec.get('contact_id', None)
# log.debug(contact_id)
# if contact_result := load_contact_obj(
# contact_id = contact_id,
# limit = limit,
# by_alias = by_alias,
# exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
# enabled = enabled,
# inc_address = inc_address,
# ):
# user_obj.contact = contact_result
# else: user_obj.contact = None
if inc_event_list:
from app.methods.event_methods import load_event_obj_list
@@ -138,6 +145,33 @@ def load_user_obj(
user_obj.event_list = event_dict_list
else: user_obj.event_list = []
# Updated 2021-06-18
if inc_order_list:
if order_rec_list_result := get_order_rec_list(
for_obj_type = 'user',
for_obj_id = user_id,
limit = limit,
enabled = enabled,
):
order_result_list = []
for order_rec in order_rec_list_result:
order_result_list.append(
load_order_obj(
order_id = order_rec.get('order_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
enabled = enabled,
inc_order_cfg = inc_order_cfg,
inc_order_line_list = inc_order_line_list,
inc_person = inc_person,
# inc_user = inc_user,
)
)
user_obj.order_list = order_result_list
else: user_obj.order_list = []
# Updated 2021-06-18
if inc_organization:
organization_id = user_rec.get('organization_id', None)
@@ -166,8 +200,7 @@ def load_user_obj(
model_as_dict = model_as_dict,
inc_address = inc_address,
inc_contact = inc_contact,
# inc_organization = inc_organization,
# inc_user = inc_user,
inc_organization = inc_organization,
):
user_obj.person = person_result
else: user_obj.person = None