A lot of unplanned clean up and created some new/missing methods and routes. Mainly working on order and order_cart related stuff.

This commit is contained in:
Scott Idem
2021-08-07 18:15:59 -04:00
parent 9915c7d9ed
commit 106a186ea6
12 changed files with 565 additions and 278 deletions

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_cart_line_models import Order_Cart_Line_Base
# ### BEGIN ### API Order Cart Line Methods ### create_order_cart_line_obj() ###
def create_order_cart_line_obj(order_cart_line_obj_new:Order_Cart_Line_Base):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not order_cart_line_obj_new:
return False
order_cart_line_obj_data = order_cart_line_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'user', 'created_on', 'updated_on'})
if order_cart_line_obj_in_result := sql_insert(data=order_cart_line_obj_data, table_name='order_cart_line', rm_id_random=True, id_random_length=8): pass
else:
return False
log.debug(order_cart_line_obj_in_result)
order_cart_line_id = order_cart_line_obj_in_result
log.debug(f'Returning the new order_cart_line_id: {order_cart_line_id}')
return order_cart_line_id
# ### END ### API Order Cart Line Methods ### create_order_cart_line_obj() ###
# ### BEGIN ### API Order Cart Line Methods ### load_order_cart_line_obj() ###
def load_order_cart_line_obj(
order_cart_line_id: int|str,
limit: int = 1000,
by_alias: bool = True,
exclude_unset: bool = True,
model_as_dict: bool = False,
) -> Order_Cart_Line_Base|dict|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if order_cart_line_id := redis_lookup_id_random(record_id_random=order_cart_line_id, table_name='order_cart_line'): pass
else: return False
if order_cart_line_rec := sql_select(table_name='v_order_cart_line', record_id=order_cart_line_id): pass
else: return False
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_cart_line_rec)
try:
order_cart_line_obj = Order_Cart_Line_Base(**order_cart_line_rec)
log.debug(order_cart_line_obj)
except ValidationError as e:
log.error(e.json())
return False
if model_as_dict:
return order_cart_line_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
else:
return order_cart_line_obj
# ### END ### API Order Cart Line Methods ### load_order_cart_line_obj() ###
# ### BEGIN ### API Order Cart Line Methods ### update_order_cart_line_obj() ###
def update_order_cart_line_obj(
order_cart_line_id: int|str, # Ideally the int ID should be passed. This allows for updating of the id_random value.
order_cart_line_obj_up: Order_Cart_Line_Base,
create_missing_obj: bool = False,
) -> bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if order_cart_line_id := redis_lookup_id_random(record_id_random=order_cart_line_id, table_name='order_cart_line'): pass
else: return False
order_cart_line_obj_up.id = order_cart_line_id
log.debug(order_cart_line_obj_up)
# log.debug(order_cart_line_obj_up.dict(by_alias=True, exclude_unset=True))
log.debug(order_cart_line_obj_up.dict(by_alias=False, exclude_unset=True))
# log.debug(order_cart_line_obj_up.dict(by_alias=False, exclude_unset=False))
#order_cart_line_dict_up = order_cart_line_obj_up.dict(by_alias=False, exclude_unset=True)
# if order_cart_line_obj_up.person_id and order_cart_line_obj_up.person:
# person_id = order_cart_line_obj_up.person_id
# person_obj_up = order_cart_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_cart_line_obj_up.user_id and order_cart_line_obj_up.user:
# user_id = order_cart_line_obj_up.user_id
# user_obj_up = order_cart_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_cart_line_dict_up = order_cart_line_obj_up.dict(by_alias=False, exclude_unset=True, exclude={'user'})
log.debug(order_cart_line_dict_up)
if order_cart_line_obj_up_result := sql_update(data=order_cart_line_dict_up, table_name='order_cart_line', rm_id_random=True):
log.debug(order_cart_line_obj_up_result)
return True
else:
log.debug(order_cart_line_obj_up_result)
return False
# ### END ### API Order Cart Line Methods ### update_order_cart_line_obj() ###
# ### BEGIN ### API Order Cart Line Methods ### get_order_cart_line_rec_list() ###
def get_order_cart_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_cart_line_id', `tbl`.id_random AS 'order_cart_line_id_random'
FROM `order_cart_line` AS `tbl`
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if order_cart_line_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
order_cart_line_rec_li = order_cart_line_rec_li_result
else:
order_cart_line_rec_li = []
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_cart_line_rec_li_result)
return order_cart_line_rec_li
# ### END ### API Order Cart Line Methods ### get_order_cart_line_rec_list() ###

View File

@@ -7,7 +7,11 @@ from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, v
from app.lib_general import log, logging
from app.db_sql import redis_lookup_id_random, sql_insert_or_update, sql_insert, sql_update, sql_select
from app.methods.order_cart_line_methods import get_order_cart_line_rec_list, load_order_cart_line_obj
from app.methods.order_cfg_methods import load_order_cfg_obj
from app.models.order_cart_models import Order_Cart_Base
from app.models.order_cart_line_models import Order_Cart_Line_Base
def update_order_cart_obj(
@@ -122,9 +126,14 @@ def update_order_cart_obj(
# ### BEGIN ### API Order Cart Methods ### load_order_cart_obj() ###
# Update 2021-08-02 (partially)
# Updated 2021-08-07
def load_order_cart_obj(
order_cart_id: int|str,
limit: int = 1000,
by_alias: bool = True,
exclude_unset: bool = True,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_order_cart_line_list: bool = False,
inc_order_cfg: bool = False,
) -> Order_Cart_Base|dict|bool:
@@ -134,37 +143,106 @@ def load_order_cart_obj(
if order_cart_id := redis_lookup_id_random(record_id_random=order_cart_id, table_name='order_cart'): pass
else: return False
if order_cart_rec := sql_select(table_name='v_order_cart', record_id=order_cart_id):
#log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_cart_rec)
if order_cart_rec := sql_select(table_name='v_order_cart', record_id=order_cart_id): pass
else: return False
if inc_order_cart_line_list:
order_cart_line_data = {}
order_cart_line_data['order_cart_id'] = order_cart_id
if order_cart_line_rec_li := sql_select(table_name='v_order_cart_line', data=order_cart_line_data, as_list=True):
order_cart_rec['order_cart_line_list'] = order_cart_line_rec_li
else: order_cart_rec['order_cart_line_list'] = []
if inc_order_cfg:
if order_cfg_rec := sql_select(table_name='v_account_cfg_detail', field_name='account_id', field_value=order_cart_rec.get('account_id', None)):
order_cart_rec['cfg'] = order_cfg_rec
else: order_cart_rec['cfg'] = None
log.debug(order_cart_rec)
else:
return False
#log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_cart_rec)
try:
order_cart_obj = Order_Cart_Base(**order_cart_rec)
log.debug(order_cart_obj)
except ValidationError as e:
log.error(e.json())
return False
return order_cart_obj
# Updated 2021-08-07
if inc_order_cart_line_list:
if order_cart_line_rec_list_result := get_order_cart_line_rec_list(
for_obj_type = 'order_cart',
for_obj_id = order_cart_id,
limit = limit,
):
order_cart_line_result_list = []
for order_cart_line_rec in order_cart_line_rec_list_result:
order_cart_line_result_list.append(
load_order_cart_line_obj(
order_cart_line_id = order_cart_line_rec.get('order_cart_line_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
)
)
order_cart_obj.order_cart_line_list = order_cart_line_result_list
else: order_cart_obj.order_cart_line_list = []
if inc_order_cfg:
if order_cfg_result := load_order_cfg_obj(
account_id = order_cart_rec.get('account_id', None),
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
):
order_cart_obj.cfg = order_cfg_result
else: order_cart_obj.cfg = None
if model_as_dict:
return order_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
else:
return order_obj
# ### END ### API Order Cart Methods ### load_order_cart_obj() ###
# ### BEGIN ### API Order Cart Methods ### get_order_cart_rec_list() ###
# Updated 2021-08-07
def get_order_cart_rec_list(
for_obj_type: str,
for_obj_id: str,
limit: int = 1000,
enabled: str = 'enabled', # enabled, disabled, all
from_datetime: datetime.datetime = None,
to_datetime: datetime.datetime = None,
# 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 limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql = f"""
SELECT `tbl`.id AS 'order_cart_id', `tbl`.id_random AS 'order_cart_id_random'
FROM `order` AS `tbl`
WHERE
{sql_obj_type_id}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if order_cart_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
order_cart_rec_li = order_cart_rec_li_result
else:
order_cart_rec_li = []
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_cart_rec_li_result)
return order_cart_rec_li
# ### END ### API Order Cart Methods ### get_order_cart_rec_list() ###
# IS THIS STILL NEEDED?
# ### BEGIN ### API Order Cart Model ### save_order_cart_obj() ###

View File

@@ -15,9 +15,9 @@ def load_order_cfg_obj(
account_id: int|str,
by_alias: bool = True,
exclude_unset: bool = True,
model_as_dict: bool = False,
model_as_dict: bool = True,
) -> Order_Cfg_Base|dict|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 account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
@@ -29,16 +29,20 @@ def load_order_cfg_obj(
field_value = account_id,
): pass
else: return False
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_cfg_rec)
try:
order_cfg_obj = Order_Cfg_Base(**order_cfg_rec)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_cfg_obj)
except ValidationError as e:
log.error(e.json())
if model_as_dict:
log.debug('**** HERE dict ****')
return order_cfg_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
else:
log.debug('**** HERE obj ****')
return order_cfg_obj
# ### END ### API Order Cfg Methods ### load_order_cfg_obj() ###

View File

@@ -170,12 +170,12 @@ def load_order_obj(
exclude_unset: bool = True,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_order_cfg: bool = False,
inc_order_line_list: bool = False,
inc_order_cfg: bool = False,
inc_person: bool = False,
inc_user: bool = False,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.DEBUG) # 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
@@ -193,18 +193,6 @@ def load_order_obj(
except ValidationError as e:
log.error(e.json())
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-18
if inc_order_line_list:
if order_line_rec_list_result := get_order_line_rec_list(
@@ -226,12 +214,19 @@ def load_order_obj(
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
# Updated 2021-08-07
if inc_order_cfg:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
if order_cfg_result := load_order_cfg_obj(
account_id = order_rec.get('account_id', None),
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = True,
):
order_obj.cfg = order_cfg_result
else: order_obj.cfg = None
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_cfg_result)
# Updated 2021-06-22
if inc_person:

View File

@@ -0,0 +1,136 @@
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_Cart_Line_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['order_cart_line_id_random'],
alias = 'order_cart_line_id_random',
default_factory = lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
alias = 'order_cart_line_id'
)
order_cart_id_random: Optional[str]
order_cart_id: Optional[int]
product_id_random: str
product_id: Optional[int]
product_for_type: Optional[str] # Copied from product record
product_for_id_random: Optional[str] # Copied from product record
product_for_id: Optional[int] # Copied from product record
product_type_id: Optional[int] # Copied from product record
product_type: Optional[str] # Copied from product record; dup from look up? probably not use?
product_type_name: Optional[str] # Copied from product record; from look up
product_name: Optional[str] # Copied from product record
product_description: Optional[str] # Copied from product record
product_unit_price: Optional[int] # Copied from product record
product_recurring: Optional[bool] # Copied from product record
curr_product_id_random: str # Should be the same as product_id_random above
curr_product_id: Optional[int] # Should be the same as product_id above
curr_product_for_type: Optional[str] # Dynamic from v_order_current_line
curr_product_for_id_random: Optional[str] # Dynamic from v_order_current_line
curr_product_for_id: Optional[int] # Dynamic from v_order_current_line
curr_product_type_id: Optional[int] # Dynamic from v_order_current_line
curr_product_type: Optional[str] # Dynamic from v_order_current_line
curr_product_type_name: Optional[str] # Dynamic from v_order_current_line
curr_product_name: Optional[str] # Dynamic from v_order_current_line
curr_product_description: Optional[str] # Dynamic from v_order_current_line
curr_product_unit_price: Optional[int] # Dynamic from v_order_current_line
curr_product_max_quantity: Optional[int] # Dynamic from v_order_current_line
curr_product_recurring: Optional[bool] # Dynamic from v_order_current_line
for_person_id: Optional[int]
for_person_id_random: Optional[str]
for_person_given_name: Optional[str] # Dynamic from v_order_current_line
for_person_full_name: Optional[str] # Dynamic from v_order_current_line
name: Optional[str] # Should be the same as product_name above
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
# Including other related objects
# product: Optional[Union[Product_Base, None]] # Future use?
# for_person: Optional[Union[Person_Base, None]] # Future use?
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
#@validator('order_cart_line_id_random', always=True)
def order_cart_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_cart_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_cart_line')
return None
@validator('order_cart_id', always=True)
def order_cart_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['order_cart_id_random']:
return redis_lookup_id_random(record_id_random=values['order_cart_id_random'], table_name='order_cart')
return None
@validator('product_id', always=True)
def product_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values.get('product_id_random', None):
return redis_lookup_id_random(record_id_random=values['product_id_random'], table_name='product')
return None
@validator('product_for_id', always=True)
def product_for_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['product_for_id_random'] and values['product_for_type']:
return redis_lookup_id_random(record_id_random=values['product_for_id_random'], table_name=values['product_for_type'])
return None
@validator('curr_product_for_id', always=True)
def curr_product_for_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['curr_product_for_id_random'] and values['curr_product_for_type']:
return redis_lookup_id_random(record_id_random=values['curr_product_for_id_random'], table_name=values['curr_product_for_type'])
return None
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True
fields = base_fields

View File

@@ -5,103 +5,11 @@ 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 app.lib_general import log, logging
from .common_field_schema import base_fields, default_num_bytes
from .order_models import Order_Cfg_Base
# class Order_Cfg_Base(BaseModel):
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# log.debug(locals())
# account_name: Optional[str]
# show_cart: Optional[bool]
# cart_label: Optional[str]
class Order_Cart_Line_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['order_cart_line_id_random'],
alias = 'order_cart_line_id_random',
default_factory = lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
alias = 'order_cart_line_id'
)
#order_cart_line_id_random: Optional[str]
order_cart_id_random: Optional[str]
order_cart_id: Optional[int]
product_id_random: str
product_id: Optional[int]
product_type_id: Optional[int]
product_type: Optional[str]
product_name: Optional[str]
product_description: Optional[str]
product_unit_price: Optional[int] = Field(0, ge=0, lt=1500000)
product_max_quantity: Optional[int] = Field(0, ge=0, lt=150)
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_cart_line_id_random', always=True)
def order_cart_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_cart_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_cart_line')
return None
@validator('order_cart_id', always=True)
def order_cart_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['order_cart_id_random']:
return redis_lookup_id_random(record_id_random=values['order_cart_id_random'], table_name='order_cart')
return None
@validator('product_id', always=True)
def product_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values.get('product_id_random', None):
return redis_lookup_id_random(record_id_random=values['product_id_random'], table_name='product')
return None
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True
fields = base_fields
from app.models.common_field_schema import base_fields, default_num_bytes
from app.models.order_cart_line_models import Order_Cart_Line_Base
from app.models.order_cfg_models import Order_Cfg_Base
class Order_Cart_DB_Base(BaseModel):
@@ -230,8 +138,8 @@ class Order_Cart_Base(BaseModel):
updated_on: Optional[datetime.datetime] = None
# Including other related objects
order_cart_line_list: Optional[list[Order_Cart_Line_Base]] # Order_Line_Base() # List[Order_Cart_Line_Base] = []
cfg: Optional[Order_Cfg_Base] = Order_Cfg_Base() # Should this be renamed to order_cfg?
order_cart_line_list: Optional[list[Order_Cart_Line_Base]] # Order_Line_Base() # List[Order_Cart_Line_Base] = []
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)

View File

@@ -17,6 +17,21 @@ class Order_Cfg_Base(BaseModel):
account_id_random: Optional[str]
account_id: Optional[int]
account_name: Optional[str]
default_no_reply_email: Optional[str]
default_no_reply_name: Optional[str]
confirm_email: Optional[str]
confirm_name: Optional[str]
order_header: Optional[str]
order_thanks: Optional[str]
order_message: Optional[str]
order_footer: Optional[str]
order_fundraising_thanks: Optional[str]
order_fundraising_message: Optional[str]
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
@validator('account_id', always=True)
@@ -27,5 +42,3 @@ class Order_Cfg_Base(BaseModel):
if values['account_id_random']:
return redis_lookup_id_random(record_id_random=values['account_id_random'], table_name='account')
return None
Order_Cfg_Base.update_forward_refs()

View File

@@ -20,35 +20,59 @@ class Order_Line_Base(BaseModel):
default_factory = lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
#alias = 'order_line_id'
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]
product_for_type: Optional[str] # Copied from product record
product_for_id_random: Optional[str] # Copied from product record
product_for_id: Optional[int] # Copied from product record
product_type_id: Optional[int] # Copied from product record
product_type: Optional[str] # WARNING: Copied from product record; dup from look up? probably not use?
product_type_name: Optional[str] # Copied from product record; from look up
product_name: Optional[str] # Copied from product record
product_description: Optional[str] # Copied from product record
product_unit_price: Optional[int] # Copied from product record
product_recurring: Optional[bool] # Copied from product record
curr_product_id_random: str # Should be the same as product_id_random above
curr_product_id: Optional[int] # Should be the same as product_id above
curr_product_for_type: Optional[str] # Dynamic from v_order_line
curr_product_for_id_random: Optional[str] # Dynamic from v_order_line
curr_product_for_id: Optional[int] # Dynamic from v_order_line
curr_product_type_id: Optional[int] # Dynamic from v_order_line
curr_product_type: Optional[str] # Dynamic from v_order_line
curr_product_type_name: Optional[str] # Dynamic from v_order_line
curr_product_name: Optional[str] # Dynamic from v_order_line
curr_product_description: Optional[str] # Dynamic from v_order_line
curr_product_unit_price: Optional[int] # Dynamic from v_order_line
curr_product_max_quantity: Optional[int] # Dynamic from v_order_line
curr_product_recurring: Optional[bool] # Dynamic from v_order_line
# NOTE NOTE NOTE NOTE NOTE NOTE NOTE NOTE
# NOTE need to check @validator NOTE NOTE NOTE NOTE
for_type: Optional[str] # Copied from the current value in product.for_type
for_id_random: Optional[str] # Copied from the current value in product.for_id_random
for_id: Optional[int] # Copied from the current value in product.for_id
# NOTE!!! Need ot add similar to order_cart_line???
# NOTE!!! Need to add to the actual order_line (and order_cart_line???) tables!
for_person_id: Optional[int]
for_person_id_random: Optional[str]
for_person_given_name: Optional[str] # Dynamic from v_order_line
for_person_full_name: Optional[str] # Dynamic from v_order_line
name: Optional[str]
description: Optional[str]
name: Optional[str] # Should be the same as product_name above
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
# Including other related objects
# product: Optional[Union[Product_Base, None]] # Future use?
# for_person: Optional[Union[Person_Base, None]] # Future use?
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
#@validator('order_line_id_random', always=True)
@@ -84,19 +108,29 @@ class Order_Line_Base(BaseModel):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['product_id_random']:
if values.get('product_id_random', None):
return redis_lookup_id_random(record_id_random=values['product_id_random'], table_name='product')
return None
@validator('for_id', always=True)
def for_id_lookup(cls, v, values, **kwargs):
@validator('product_for_id', always=True)
def product_for_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['for_id_random'] and values['for_type']:
return redis_lookup_id_random(record_id_random=values['for_id_random'], table_name=values['for_type'])
if values['product_for_id_random'] and values['product_for_type']:
return redis_lookup_id_random(record_id_random=values['product_for_id_random'], table_name=values['product_for_type'])
return None
@validator('curr_product_for_id', always=True)
def curr_product_for_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['curr_product_for_id_random'] and values['curr_product_for_type']:
return redis_lookup_id_random(record_id_random=values['curr_product_for_id_random'], table_name=values['curr_product_for_type'])
return None
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True
fields = base_fields

View File

@@ -5,136 +5,45 @@ 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 app.lib_general import log, logging
from .common_field_schema import base_fields, default_num_bytes
from .order_line_models import Order_Line_Base
from .person_models import Person_Base
from .user_models import User_Base
class Order_Cfg_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
account_name: Optional[str]
default_no_reply_email: Optional[str]
default_no_reply_name: Optional[str]
confirm_email: Optional[str]
confirm_name: Optional[str]
order_header: Optional[str]
order_thanks: Optional[str]
order_message: Optional[str]
order_footer: Optional[str]
order_fundraising_thanks: Optional[str]
order_fundraising_message: Optional[str]
# 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
from app.models.common_field_schema import base_fields, default_num_bytes
from app.models.order_cfg_models import Order_Cfg_Base
from app.models.order_line_models import Order_Line_Base
from app.models.person_models import Person_Base
from app.models.user_models import User_Base
class Order_Base(BaseModel):
log.setLevel(logging.WARNING)
log.debug(locals())
# from .person_models import Person_Base
# from .user_models import User_Base
id_random: Optional[str] = Field(
**base_fields['order_id_random'],
alias = 'order_id_random',
default_factory = lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
#alias = 'order_id'
alias = 'order_id'
)
#order_id_random: Optional[str]
#order_id: Optional[int]
account_id_random: Optional[str]
account_id: Optional[int] # NOTE: This is not really optional
person_id_random: Optional[str]
person_id: Optional[int]
user_id_random: Optional[str]
user_id: Optional[int]
total_quantity: Optional[int] = Field(0, ge=0, lt=150)
total_bill: Optional[int] = Field(0, ge=0, lt=1500000) # NOTE: This is total_amount in the order_cart
total_paid: Optional[int] = Field(0, ge=0, lt=1500000)
balance: Optional[int] = Field(0, ge=-1500000, lt=1500000) # Balance needs to be calculated
status: Optional[str]
notes: Optional[str]
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None

View File

@@ -27,6 +27,7 @@ class Product_Base(BaseModel):
# NOTE NOTE NOTE NOTE NOTE NOTE NOTE NOTE
# NOTE need to check @validator NOTE NOTE NOTE NOTE
# This is essentially what the product is paying for
for_type: Optional[str]
for_id_random: Optional[str]
for_id: Optional[int]

View File

@@ -158,24 +158,39 @@ async def get_order_obj_li(
return mk_resp(data=False, status_code=404)
@router.get('/{obj_id}', response_model=Resp_Body_Base)
# ### BEGIN ### API Order Routes ### get_order_obj() ###
# Updated 2021-08-07
@router.get('/{order_id}', response_model=Resp_Body_Base)
async def get_order_obj(
obj_id: str = Query(..., min_length=1, max_length=22),
order_id: str = Query(..., min_length=11, max_length=22),
limit: int = 500,
enabled: str = 'enabled',
inc_order_line_list: bool = False,
inc_order_cfg: bool = False,
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
by_alias: bool = True,
exclude_unset: bool = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'order'
result = get_obj_template(
obj_type=obj_type,
obj_id=obj_id,
by_alias=True,
exclude_unset=True,
)
return result
if order_obj := load_order_obj(
order_id = order_id,
limit = limit,
enabled = enabled,
by_alias = by_alias,
exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
inc_order_line_list = inc_order_line_list,
inc_order_cfg = inc_order_cfg,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_obj)
data = order_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset)
return mk_resp(data=data)
else:
return mk_resp(data=False, status_code=404) # Not Found
# ### END ### API Order Routes ### get_order_obj() ###
@router.delete('/{obj_id}', response_model=Resp_Body_Base)

View File

@@ -197,10 +197,13 @@ async def lookup_order_cart_obj(
return mk_resp(data=False, status_code=404, response=response) # Not Found
# Update 2021-08-02 (partially)
# ### BEGIN ### API Order Cart Routes ### get_order_cart_obj() ###
# Updated 2021-08-07
@router.get('/{order_cart_id}', response_model=Resp_Body_Base)
async def get_order_cart_obj(
order_cart_id: str = Query(..., min_length=1, max_length=22),
order_cart_id: str = Query(..., min_length=11, max_length=22),
limit: int = 500,
enabled: str = 'enabled',
inc_order_cart_line_list: bool = False,
inc_order_cfg: bool = False,
x_account_id: str = Header(...),
@@ -212,6 +215,11 @@ async def get_order_cart_obj(
if order_cart_obj := load_order_cart_obj(
order_cart_id = order_cart_id,
limit = limit,
enabled = enabled,
by_alias = by_alias,
exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
inc_order_cart_line_list = inc_order_cart_line_list,
inc_order_cfg = inc_order_cfg,
):
@@ -219,6 +227,7 @@ async def get_order_cart_obj(
return mk_resp(data=data)
else:
return mk_resp(data=False, status_code=404) # Not Found
# ### END ### API Order Cart Routes ### get_order_cart_obj() ###
@router.delete('/{obj_id}', response_model=Resp_Body_Base)