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

Work on lots of methods and models.

Browse Source
This commit is contained in:
Scott Idem
2021-06-17 17:53:55 -04:00
parent c17724cea1
commit 47ce2380be
13 changed files with 535 additions and 359 deletions
Download Patch File Download Diff File Expand all files Collapse all files

165
app/methods/account_methods.py
View File

@@ -10,12 +10,14 @@ from app.lib_general import log, logging
from app.methods.account_cfg_methods import load_account_cfg_obj
from app.methods.address_methods import get_address_rec_list, load_address_obj
from app.methods.archive_methods import get_archive_rec_list, load_archive_obj
# from app.methods.contact_methods import load_contact_obj_list
from app.methods.event_methods import load_event_obj_list
from app.methods.contact_methods import get_contact_rec_list, load_contact_obj
from app.methods.event_methods import get_event_rec_list, load_event_obj
from app.methods.order_methods import get_order_rec_list, load_order_obj
# from app.methods.order_cart_methods import get_order_cart_rec_list, load_order_cart_obj
from app.methods.organization_methods import get_organization_rec_list, load_organization_obj
from app.methods.person_methods import get_person_rec_list, load_person_obj
from app.methods.product_methods import get_product_rec_list, load_product_obj
from app.methods.post_methods import load_post_obj_list
from app.methods.post_methods import get_post_rec_list, load_post_obj
from app.methods.user_methods import get_user_rec_list, load_user_obj
from app.models.account_models import Account_Base
@@ -74,9 +76,14 @@ def load_account_obj(
# inc_membership: bool = False,
inc_membership_cfg: bool = False,
inc_membership_list: bool = False,
inc_membership_member: bool = False,
inc_membership_member_list: bool = False, # Priority l1
inc_order: bool = False,
inc_order_list: bool = False, # Priority l2
inc_order_cfg: bool = False,
inc_order_list: bool = False, # Priority l1
inc_order_line_list: bool = False, # Priority l2
inc_order_cart: bool = False,
inc_order_cart_line_list: bool = False, # Priority l2
inc_order_cart_list: bool = False,
inc_organization: bool = False,
inc_organization_list: bool = False, # Priority l3
@@ -125,16 +132,6 @@ def load_account_obj(
account_obj.account_cfg = account_cfg_dict
else: account_obj.account_cfg = None
# if inc_address_list:
# if address_dict_list := load_address_obj_list(
# account_id = account_id,
# limit = limit,
# model_as_dict = model_as_dict,
# enabled = enabled,
# ):
# account_obj.address_list = address_dict_list
# else: account_obj.address_list = []
# Updated 2021-06-17
if inc_address_list:
if address_rec_list_result := get_address_rec_list(
@@ -151,23 +148,11 @@ def load_account_obj(
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_address_content_list = inc_address_content_list,
)
)
account_obj.address_list = address_dict_list
else: account_obj.address_list = []
# if inc_archive_list:
# if archive_dict_list := load_archive_obj_list(
# account_id = account_id,
# limit = limit,
# model_as_dict = model_as_dict,
# enabled = enabled,
# inc_archive_content_list = inc_archive_content_list,
# ):
# account_obj.archive_list = archive_dict_list
# else: account_obj.archive_list = []
# Updated 2021-06-17
if inc_archive_list:
if archive_rec_list_result := get_archive_rec_list(
@@ -190,28 +175,39 @@ def load_account_obj(
account_obj.archive_list = archive_dict_list
else: account_obj.archive_list = []
# if inc_contact_list:
# if contact_dict_list := load_contact_obj_list(
# account_id = account_id,
# limit = limit,
# model_as_dict = model_as_dict,
# enabled = enabled,
# inc_address = inc_address,
# ):
# account_obj.contact_list = contact_dict_list
# else: account_obj.contact_list = []
# Updated 2021-06-17
if inc_contact_list:
if contact_dict_list := load_contact_obj_list(
account_id = account_id,
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_address = inc_address,
):
if contact_rec_list_result := get_contact_rec_list(
for_obj_type = 'account',
for_obj_id = account_id,
limit = limit,
enabled = enabled,
):
contact_dict_list = []
for contact_rec in contact_rec_list_result:
contact_dict_list.append(
load_contact_obj(
contact_id = contact_rec.get('contact_id', None),
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_address = inc_address,
)
)
account_obj.contact_list = contact_dict_list
else: account_obj.contact_list = []
# if inc_organization:
# organization_id = account_rec.get('organization_id', None)
# if organization_obj_result := load_organization_obj(organization_id=organization_id):
# organization_obj = organization_obj_result
# # account_rec['organization'] = organization_obj
# # log.debug(account_rec)
# #account_obj.organization = organization_obj.dict(by_alias=True, exclude_unset=True)
# account_obj.organization = organization_obj
# else: account_obj.organization = None
# Updated 2021-06-17
if inc_event_list:
if event_rec_list_result := get_event_rec_list(
@@ -251,33 +247,30 @@ def load_account_obj(
account_obj.event_list = event_dict_list
else: account_obj.event_list = []
# if inc_event_list:
# if event_dict_list := load_event_obj_list(
# account_id = account_id,
# limit = limit,
# model_as_dict = model_as_dict,
# enabled = enabled,
# # inc_location_address = inc_address,
# # inc_contact_1 = inc_contact,
# # inc_contact_2 = inc_contact,
# # inc_contact_3 = inc_contact,
# # inc_event_abstract_list = inc_event_abstract_list,
# # inc_event_badge_list = inc_event_badge_list,
# # inc_event_device_list = inc_event_device_list,
# inc_event_exhibit_list = inc_event_exhibit_list,
# inc_event_file_list = inc_event_file_list,
# inc_event_location_list = inc_event_location_list,
# inc_event_person_list = inc_event_person_list,
# inc_event_presentation_list = inc_event_presentation_list,
# inc_event_presenter_list = inc_event_presenter_list,
# inc_event_registration_list = inc_event_registration_list,
# inc_event_session_list = inc_event_session_list,
# inc_event_track_list = inc_event_track_list,
# # inc_person = inc_person,
# # inc_user = inc_user,
# ):
# account_obj.event_list = event_dict_list
# else: account_obj.event_list = []
# Updated 2021-06-17
if inc_order_list:
if order_rec_list_result := get_order_rec_list(
for_obj_type = 'account',
for_obj_id = account_id,
limit = limit,
enabled = enabled,
):
order_dict_list = []
for order_rec in order_rec_list_result:
order_dict_list.append(
load_order_obj(
order_id = order_rec.get('order_id', None),
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_order_line_list = inc_order_line_list,
inc_order_cfg = inc_order_cfg,
inc_person = inc_person,
inc_user = inc_user,
)
)
account_obj.order_list = order_dict_list
else: account_obj.order_list = []
# Updated 2021-06-17
if inc_organization_list:
@@ -304,17 +297,27 @@ def load_account_obj(
account_obj.organization_list = organization_dict_list
else: account_obj.organization_list = []
# Updated 2021-06-17
if inc_post_list:
if post_dict_list := load_post_obj_list(
account_id = account_id,
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_post_comment_list = inc_post_comment_list,
inc_person = inc_person,
inc_user = inc_user,
):
if post_rec_list_result := get_post_rec_list(
for_obj_type = 'account',
for_obj_id = account_id,
limit = limit,
enabled = enabled,
):
post_dict_list = []
for post_rec in post_rec_list_result:
post_dict_list.append(
load_post_obj(
post_id = post_rec.get('post_id', None),
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_post_comment_list = inc_post_comment_list,
inc_person = inc_person,
inc_user = inc_user,
)
)
account_obj.post_list = post_dict_list
else: account_obj.post_list = []

19
app/methods/address_methods.py
View File

@@ -110,15 +110,16 @@ def get_address_rec_list(
# 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 = ''
# 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

55
app/methods/contact_methods.py
View File

@@ -155,3 +155,58 @@ def update_contact_obj(
log.debug(contact_obj_up_result)
return False
# ### END ### API Contact Methods ### update_contact_obj() ###
# ### BEGIN ### API Contact Methods ### get_contact_rec_list() ###
def get_contact_rec_list(
for_obj_type: str,
for_obj_id: str,
limit: int = 1000,
enabled: str = 'enabled', # enabled, disabled, all
) -> 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 'contact_id', `tbl`.id_random AS 'contact_id_random'
FROM `contact` AS `tbl`
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if contact_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
contact_rec_li = contact_rec_li_result
else:
contact_rec_li = []
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(contact_rec_li_result)
return contact_rec_li
# ### END ### API Contact Methods ### get_contact_rec_list() ###

75
app/methods/order_methods.py
View File

@@ -4,10 +4,11 @@ import datetime
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from app.lib_general import *
from app.db_sql import redis_lookup_id_random, sql_select
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.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
@@ -157,7 +158,16 @@ def save_order_obj(order_obj_new:Order_Base, repl_order_line_li:bool=False):
# ### BEGIN ### API Order Methods ### load_order_obj() ###
def load_order_obj(order_id:int|str, inc_order_line_li:bool=False, inc_order_cfg:bool=False, inc_person_obj:bool=False, inc_user_obj:bool=False):
def load_order_obj(
order_id: int|str,
limit: int = 1000,
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,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
@@ -169,7 +179,7 @@ def load_order_obj(order_id:int|str, inc_order_line_li:bool=False, inc_order_cfg
#log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_rec)
if inc_order_line_li:
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):
@@ -202,3 +212,60 @@ def load_order_obj(order_id:int|str, inc_order_line_li:bool=False, inc_order_cfg
return order_obj
# ### END ### API Order Methods ### load_order_obj() ###
# ### BEGIN ### API Order Methods ### get_order_rec_list() ###
def get_order_rec_list(
for_obj_type: str,
for_obj_id: str,
limit: int = 1000,
enabled: str = 'enabled', # enabled, disabled, all
status: str = 'complete', # started, in progress, complete, all
balance_gt: int = 0, # $0 to $99999
) -> 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_id', `tbl`.id_random AS 'order_id_random'
FROM `order` AS `tbl`
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if order_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
order_rec_li = order_rec_li_result
else:
order_rec_li = []
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_rec_li_result)
return order_rec_li
# ### END ### API Order Methods ### get_order_rec_list() ###

23
app/methods/organization_methods.py
View File

@@ -4,7 +4,7 @@ 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
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.address_methods import load_address_obj
@@ -45,7 +45,7 @@ def load_organization_obj(
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
contact_id = organization_rec.get('contact_id', None)
log.debug(contact_id)
from app.methods.contact_methods import load_contact_obj
# from app.methods.contact_methods import load_contact_obj
if contact_dict := load_contact_obj(
contact_id = contact_id,
model_as_dict = model_as_dict,
@@ -126,15 +126,16 @@ def get_organization_rec_list(
# 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 = ''
# 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

19
app/methods/person_methods.py
View File

@@ -259,15 +259,16 @@ def get_person_rec_list(
# 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 = ''
# 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

83
app/methods/post_comment_list_methods.py
View File

@@ -1,83 +0,0 @@
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.methods.post_comment_methods import load_post_comment_obj
# ### BEGIN ### API Post Comment Methods ### load_post_comment_obj_list() ###
def load_post_comment_obj_list(
post_id: int|str,
limit: int = 1000,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_person: bool = False,
inc_user: bool = False,
) -> list|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if post_id := redis_lookup_id_random(record_id_random=post_id, table_name='post'): pass
else: return False
data = {}
data['post_id'] = post_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['post'] = None
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql = f"""
SELECT `tbl`.id AS 'post_comment_id', `tbl`.id_random AS 'post_comment_id_random'
FROM `post_comment` AS `tbl`
WHERE `tbl`.post_id = :post_id
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if post_comment_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_comment_rec_li_result)
post_comment_result_li = []
for post_comment_rec in post_comment_rec_li_result:
post_comment_id = post_comment_rec.get('post_comment_id', None)
if post_comment_result := load_post_comment_obj(
post_comment_id = post_comment_id,
model_as_dict = model_as_dict,
# enabled = enabled,
inc_person = inc_person,
inc_user = inc_user,
):
log.debug(post_comment_result)
post_comment_result_li.append(post_comment_result)
else:
log.debug(post_comment_result)
post_comment_result_li.append(None)
log.debug(post_comment_result_li)
else:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(post_comment_rec_li_result)
post_comment_result_li = []
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
return post_comment_result_li
# ### END ### API Post Comment Methods ### load_post_comment_obj_list() ###

55
app/methods/post_comment_methods.py
View File

@@ -227,3 +227,58 @@ def update_post_comment_obj(
log.debug(post_comment_obj_up_result)
return False
# ### END ### API Post Comment Methods ### update_post_comment_obj() ###
# ### BEGIN ### API Post Comment Methods ### get_post_comment_rec_list() ###
def get_post_comment_rec_list(
for_obj_type: str,
for_obj_id: str,
limit: int = 1000,
enabled: str = 'enabled', # enabled, disabled, all
) -> 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 'post_comment_id', `tbl`.id_random AS 'post_comment_id_random'
FROM `post_comment` AS `tbl`
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if post_comment_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
post_comment_rec_li = post_comment_rec_li_result
else:
post_comment_rec_li = []
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(post_comment_rec_li_result)
return post_comment_rec_li
# ### END ### API Post Comment Methods ### get_post_comment_rec_list() ###

85
app/methods/post_list_methods.py
View File

@@ -1,85 +0,0 @@
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.methods.post_methods import load_post_obj
# ### BEGIN ### API Post Methods ### load_post_obj_list() ###
def load_post_obj_list(
account_id: int|str,
limit: int = 1000,
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())
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return False
data = {}
data['account_id'] = account_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 `tbl`.account_id = :account_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 = []
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
return post_result_li
# ### END ### API Post Methods ### load_post_obj_list() ###

97
app/methods/post_methods.py
View File

@@ -9,7 +9,7 @@ from app.lib_general import log, logging
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 load_post_comment_obj_list
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
from app.models.post_models import Post_Base
@@ -78,17 +78,40 @@ def load_post_obj(
post_obj.person = person_obj
else: post_obj.person = None
# Updated 2021-06-17
if inc_post_comment_list:
if post_comment_dict_list := load_post_comment_obj_list(
post_id = post_id,
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_person = inc_person,
inc_user = inc_user,
):
if post_comment_rec_list_result := get_post_comment_rec_list(
for_obj_type = 'post',
for_obj_id = post_id,
limit = limit,
enabled = enabled,
):
post_comment_dict_list = []
for post_comment_rec in post_comment_rec_list_result:
post_comment_dict_list.append(
load_post_comment_obj(
post_comment_id = post_comment_rec.get('post_comment_id', None),
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_person = inc_person,
inc_user = inc_user,
)
)
post_obj.post_comment_list = post_comment_dict_list
else: post_obj.post_comment_list = None
else: post_obj.post_comment_list = []
# if inc_post_comment_list:
# if post_comment_dict_list := load_post_comment_obj_list(
# post_id = post_id,
# limit = limit,
# model_as_dict = model_as_dict,
# enabled = enabled,
# inc_person = inc_person,
# inc_user = inc_user,
# ):
# post_obj.post_comment_list = post_comment_dict_list
# else: post_obj.post_comment_list = None
if inc_user:
user_id = post_rec.get('user_id', None)
@@ -250,3 +273,57 @@ def update_post_obj(
log.debug(post_obj_up_result)
return False
# ### END ### API Post Methods ### update_post_obj() ###
# ### BEGIN ### API Post Methods ### get_post_rec_list() ###
def get_post_rec_list(
for_obj_type: str,
for_obj_id: str,
limit: int = 1000,
enabled: str = 'enabled', # enabled, disabled, all
) -> 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 = ''
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):
post_rec_li = post_rec_li_result
else:
post_rec_li = []
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(post_rec_li_result)
return post_rec_li
# ### END ### API Post Methods ### get_post_rec_list() ###

83
app/models/order_line_models.py Normal file
View File

@@ -0,0 +1,83 @@
from __future__ import annotations
import datetime, hashlib, logging, os, pytz, redis, secrets
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator
from app.db_sql import redis_lookup_id_random
from app.lib_general import *
from .common_field_schema import base_fields, default_num_bytes
class Order_Line_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['order_line_id_random'],
alias='order_line_id_random',
default_factory=lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
#alias='order_line_id'
)
#order_line_id_random: Optional[str]
order_id_random: Optional[str]
order_id: Optional[int]
product_id_random: str
product_id: Optional[int]
product_type_id: Optional[int]
product_type: Optional[str]
name: Optional[str]
description: Optional[str]
quantity: int = Field(0, ge=0, lt=150)
amount: int = Field(0, ge=0, lt=1500000)
recurring: Optional[bool] = False
message: Optional[str]
notes: Optional[str]
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
#@validator('order_line_id_random', always=True)
def order_line_id_random_copy(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['id_random']:
return values['id_random']
return None
@validator('id', always=True)
def order_line_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['id_random']:
log.debug(values['id_random'])
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='order_line')
return None
@validator('order_id', always=True)
def order_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['order_id_random']:
return redis_lookup_id_random(record_id_random=values['order_id_random'], table_name='order')
return None
@validator('product_id', always=True)
def product_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['product_id_random']:
return redis_lookup_id_random(record_id_random=values['product_id_random'], table_name='product')
return None
class Config:
underscore_attrs_are_private = True
fields = base_fields

133
app/models/order_models.py
View File

@@ -8,7 +8,7 @@ from app.db_sql import redis_lookup_id_random
from app.lib_general import *
from .common_field_schema import base_fields, default_num_bytes
#from .supporting_core_models import *
from .order_line_models import Order_Line_Base
from .person_models import Person_Base
from .user_models import User_Base
@@ -32,77 +32,77 @@ class Order_Cfg_Base(BaseModel):
order_fundraising_message: Optional[str]
class Order_Line_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# class Order_Line_Base(BaseModel):
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['order_line_id_random'],
alias='order_line_id_random',
default_factory=lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
#alias='order_line_id'
)
#order_line_id_random: Optional[str]
order_id_random: Optional[str]
order_id: Optional[int]
product_id_random: str
product_id: Optional[int]
product_type_id: Optional[int]
product_type: Optional[str]
name: Optional[str]
description: Optional[str]
quantity: int = Field(0, ge=0, lt=150)
amount: int = Field(0, ge=0, lt=1500000)
recurring: Optional[bool] = False
message: Optional[str]
notes: Optional[str]
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
# id_random: Optional[str] = Field(
# **base_fields['order_line_id_random'],
# alias='order_line_id_random',
# default_factory=lambda:secrets.token_urlsafe(default_num_bytes),
# )
# id: Optional[int] = Field(
# #alias='order_line_id'
# )
# #order_line_id_random: Optional[str]
# order_id_random: Optional[str]
# order_id: Optional[int]
# product_id_random: str
# product_id: Optional[int]
# product_type_id: Optional[int]
# product_type: Optional[str]
# name: Optional[str]
# description: Optional[str]
# quantity: int = Field(0, ge=0, lt=150)
# amount: int = Field(0, ge=0, lt=1500000)
# recurring: Optional[bool] = False
# message: Optional[str]
# notes: Optional[str]
# created_on: Optional[datetime.datetime] = None
# updated_on: Optional[datetime.datetime] = None
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
# _processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
#@validator('order_line_id_random', always=True)
def order_line_id_random_copy(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
# #@validator('order_line_id_random', always=True)
# def order_line_id_random_copy(cls, v, values, **kwargs):
# log.setLevel(logging.WARNING)
# log.debug(locals())
if values['id_random']:
return values['id_random']
return None
# if values['id_random']:
# return values['id_random']
# return None
@validator('id', always=True)
def order_line_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
# @validator('id', always=True)
# def order_line_id_lookup(cls, v, values, **kwargs):
# log.setLevel(logging.WARNING)
# log.debug(locals())
if values['id_random']:
log.debug(values['id_random'])
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='order_line')
return None
# if values['id_random']:
# log.debug(values['id_random'])
# return redis_lookup_id_random(record_id_random=values['id_random'], table_name='order_line')
# return None
@validator('order_id', always=True)
def order_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
# @validator('order_id', always=True)
# def order_id_lookup(cls, v, values, **kwargs):
# log.setLevel(logging.WARNING)
# log.debug(locals())
if values['order_id_random']:
return redis_lookup_id_random(record_id_random=values['order_id_random'], table_name='order')
return None
# if values['order_id_random']:
# return redis_lookup_id_random(record_id_random=values['order_id_random'], table_name='order')
# return None
@validator('product_id', always=True)
def product_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
# @validator('product_id', always=True)
# def product_id_lookup(cls, v, values, **kwargs):
# log.setLevel(logging.WARNING)
# log.debug(locals())
if values['product_id_random']:
return redis_lookup_id_random(record_id_random=values['product_id_random'], table_name='product')
return None
# if values['product_id_random']:
# return redis_lookup_id_random(record_id_random=values['product_id_random'], table_name='product')
# return None
class Config:
underscore_attrs_are_private = True
fields = base_fields
# class Config:
# underscore_attrs_are_private = True
# fields = base_fields
class Order_Base(BaseModel):
@@ -138,12 +138,11 @@ class Order_Base(BaseModel):
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
order_line_li: List[Order_Line_Base] = []
cfg: Optional[Order_Cfg_Base] = Order_Cfg_Base()
person: Optional[Person_Base] = Person_Base()
user: Optional[User_Base] = User_Base()
# Including other related objects
cfg: Optional[Order_Cfg_Base]
order_line_list: Optional[list] # Order_Line_Base()
person: Optional[Person_Base]
user: Optional[User_Base]
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)

2
app/routers/account.py
View File

@@ -217,6 +217,8 @@ async def get_account_obj_new(
inc_person_list = inc_person_list,
# inc_post = inc_post,
inc_post_list = inc_post_list,
# inc_post_comment = inc_post_comment,
inc_post_comment_list = inc_post_comment_list,
# inc_product = inc_product,
inc_product_list = inc_product_list,
# inc_site = inc_site,