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

Slowly getting things back to normal after FastAPI upgrade

Browse Source
This commit is contained in:
Scott Idem
2024-04-26 17:43:56 -04:00
parent e38b3cfe7a
commit d3f5f51458
4 changed files with 771 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/models/response_models.py
View File

@@ -22,7 +22,7 @@ class Resp_Body_Base(BaseModel):
# alias = 'test_prop_alias'
# )
data: Union[list, dict]
data: Union[None, list, dict]
meta: Optional[dict]
# ### END ### API Response Model ### Resp_Body_Base() ###

537
app/routers/api_crud.py
View File

@@ -193,15 +193,13 @@ obj_type_li['stripe_log'] = {'table_name': 'stripe_log', 'tbl_name_update': 'str
router = APIRouter()
# Working on the basic API CRUD - STI 2021-03-08
# Updated 2023-07-06
# Split API CRUD functions for FastAPI 0.95.1 - STI 2024-04-26
# ### BEGIN ### API CRUD ### get_obj_li_lx() ###
# Updated 2024-04-26
@router.get('/{obj_type_l1}/list')
@router.get('/{obj_type_l1}/{obj_type_l2}/list')
@router.get('/{obj_type_l1}/{obj_type_l2}/{obj_type_l3}/list')
async def get_obj_li(
async def get_obj_li_l1(
obj_type_l1: str = Path(min_length=2, max_length=50),
obj_type_l2: Optional[str] = Path(min_length=2, max_length=50),
obj_type_l3: Optional[str] = Path(min_length=2, max_length=50),
for_obj_type: Optional[str] = Query(None, max_length=50),
for_obj_id: Optional[str] = Query(None, max_length=22),
@@ -209,26 +207,10 @@ async def get_obj_li(
use_alt_table: bool = False, # NOTE: This will use table_name_alt if they exist. -2023-11-17
use_alt_base: bool = False, # NOTE: This will use base_name_alt if they exist. -2023-11-17
# field_qry_li: str = Query(None, max_length=150), # JSON formatted key value pair list of fields to search.
# fulltext_qry_li: str = Query(None, max_length=150), # JSON formatted key value pair list of fields to search.
# fulltext_qry_field_li: str = Header(None), # Json formatted string list of fields to search. It is not ideal that this is in the header. Need a better option, but this is currently a GET request.
# fulltext_qry_str: str = Query(None, max_length=150),
hidden: str = 'not_hidden', # hidden, not_hidden, all,
# order_by_li: dict = None,
order_by_li: str = Header(None), # JSON formatted string in a key value format. It is not ideal that this is in the header. Need a better option, but this is currently a GET request.
# dh_order_by_li: str = Header(None),
# dh_testing: str = Header(None),
# h_order_by_li: str = Header(None),
# h_testing: str = Header(None),
# include: Optional[list] = [],
# exclude: Optional[list] = [],
# exclude_none: Optional[bool] = True,
# Get the "json" param from the query string. This is a JSON formatted string of the data to be inserted.
jp: Optional[Union[str, None]] = None,
@@ -240,6 +222,149 @@ async def get_obj_li(
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Call generic function to get the list of objects
return handle_get_obj_li(
obj_type_l1=obj_type_l1,
for_obj_type=for_obj_type,
for_obj_id=for_obj_id,
use_alt_table=use_alt_table,
use_alt_base=use_alt_base,
hidden=hidden,
order_by_li=order_by_li,
jp=jp,
file_type=file_type,
return_file=return_file,
commons=commons,
)
@router.get('/{obj_type_l1}/{obj_type_l2}/list')
async def get_obj_li_l2(
obj_type_l1: str = Path(min_length=2, max_length=50),
obj_type_l2: str = Path(min_length=2, max_length=50),
for_obj_type: Optional[str] = Query(None, max_length=50),
for_obj_id: Optional[str] = Query(None, max_length=22),
use_alt_table: bool = False, # NOTE: This will use table_name_alt if they exist. -2023-11-17
use_alt_base: bool = False, # NOTE: This will use base_name_alt if they exist. -2023-11-17
hidden: str = 'not_hidden', # hidden, not_hidden, all,
# order_by_li: dict = None,
order_by_li: str = Header(None), # JSON formatted string in a key value format. It is not ideal that this is in the header. Need a better option, but this is currently a GET request.
# Get the "json" param from the query string. This is a JSON formatted string of the data to be inserted.
jp: Optional[Union[str, None]] = None,
file_type: str = 'CSV', # CSV, Excel
return_file: Optional[bool] = False,
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Call generic function to get the list of objects
return handle_get_obj_li(
obj_type_l1=obj_type_l1,
obj_type_l2=obj_type_l2,
for_obj_type=for_obj_type,
for_obj_id=for_obj_id,
use_alt_table=use_alt_table,
use_alt_base=use_alt_base,
hidden=hidden,
order_by_li=order_by_li,
jp=jp,
file_type=file_type,
return_file=return_file,
commons=commons,
)
@router.get('/{obj_type_l1}/{obj_type_l2}/{obj_type_l3}/list')
async def get_obj_li_l3(
obj_type_l1: str = Path(min_length=2, max_length=50),
obj_type_l2: str = Path(min_length=2, max_length=50),
obj_type_l3: str = Path(min_length=2, max_length=50),
for_obj_type: Optional[str] = Query(None, max_length=50),
for_obj_id: Optional[str] = Query(None, max_length=22),
use_alt_table: bool = False, # NOTE: This will use table_name_alt if they exist. -2023-11-17
use_alt_base: bool = False, # NOTE: This will use base_name_alt if they exist. -2023-11-17
hidden: str = 'not_hidden', # hidden, not_hidden, all,
# order_by_li: dict = None,
order_by_li: str = Header(None), # JSON formatted string in a key value format. It is not ideal that this is in the header. Need a better option, but this is currently a GET request.
# Get the "json" param from the query string. This is a JSON formatted string of the data to be inserted.
jp: Optional[Union[str, None]] = None,
file_type: str = 'CSV', # CSV, Excel
return_file: Optional[bool] = False,
commons: Common_Route_Params = Depends(common_route_params),
):
# ### SECTION ### Call generic function to get the list of objects
return handle_get_obj_li(
obj_type_l1=obj_type_l1,
obj_type_l2=obj_type_l2,
obj_type_l3=obj_type_l3,
for_obj_type=for_obj_type,
for_obj_id=for_obj_id,
use_alt_table=use_alt_table,
use_alt_base=use_alt_base,
hidden=hidden,
order_by_li=order_by_li,
jp=jp,
file_type=file_type,
return_file=return_file,
commons=commons,
)
def handle_get_obj_li(
obj_type_l1: str,
obj_type_l2: Optional[str] = None,
obj_type_l3: Optional[str] = None,
for_obj_type: Optional[str] = None,
for_obj_id: Optional[str] = None,
use_alt_table: bool = False,
use_alt_base: bool = False,
hidden: str = 'not_hidden',
order_by_li: Optional[Union[str, None]] = None,
jp: Optional[Union[str, None]] = None,
file_type: str = 'CSV',
return_file: Optional[bool] = False,
commons: Common_Route_Params = None,
):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
import urllib
# This should be a dict list of fields with a list of values to search for using FULLTEXT.
@@ -503,11 +628,76 @@ async def get_obj_li(
else: return mk_resp(data=None, response=commons.response, status_code=404)
# ### BEGIN ### API CRUD ### get_obj_lx() ###
# Updated 2023-11-03
@router.get('/{obj_type_l1}/{obj_id}')
async def get_obj_l1(
obj_type_l1: str=None,
obj_id: str=None,
use_alt_table: bool = False, # NOTE: This will use table_name_alt if they exist. -2023-12-01
use_alt_base: bool = False, # NOTE: This will use base_name_alt if they exist. -2023-12-01
# for_obj_type: Optional[str] = Query(None, max_length=50), # NOTE: This is not currently used. It is here for future use.
# for_obj_id: Optional[str] = Query(None, max_length=22), # NOTE: This is not currently used. It is here for future use.
# qry_str: Optional[str] = Query(None, max_length=50),
# qry_int: Optional[int] = None,
# include: Optional[list] = [],
# exclude: Optional[list] = [],
# exclude_none: Optional[bool] = True,
commons: Common_Route_Params = Depends(common_route_params),
):
# ### SECTION ### Call generic function to get the object
return handle_get_obj_id(
obj_type_l1=obj_type_l1,
obj_id=obj_id,
use_alt_table=use_alt_table,
use_alt_base=use_alt_base,
commons=commons,
)
@router.get('/{obj_type_l1}/{obj_type_l2}/{obj_id}')
async def get_obj_l2(
obj_type_l1: str=None,
obj_type_l2: str=None,
obj_id: str=None,
use_alt_table: bool = False, # NOTE: This will use table_name_alt if they exist. -2023-12-01
use_alt_base: bool = False, # NOTE: This will use base_name_alt if they exist. -2023-12-01
# for_obj_type: Optional[str] = Query(None, max_length=50), # NOTE: This is not currently used. It is here for future use.
# for_obj_id: Optional[str] = Query(None, max_length=22), # NOTE: This is not currently used. It is here for future use.
# qry_str: Optional[str] = Query(None, max_length=50),
# qry_int: Optional[int] = None,
# include: Optional[list] = [],
# exclude: Optional[list] = [],
# exclude_none: Optional[bool] = True,
commons: Common_Route_Params = Depends(common_route_params),
):
# ### SECTION ### Call generic function to get the object
return handle_get_obj_id(
obj_type_l1=obj_type_l1,
obj_type_l2=obj_type_l2,
obj_id=obj_id,
use_alt_table=use_alt_table,
use_alt_base=use_alt_base,
commons=commons,
)
@router.get('/{obj_type_l1}/{obj_type_l2}/{obj_type_l3}/{obj_id}')
async def get_obj(
async def get_obj_l3(
obj_type_l1: str=None,
obj_type_l2: str=None,
obj_type_l3: str=None,
@@ -528,6 +718,32 @@ async def get_obj(
commons: Common_Route_Params = Depends(common_route_params),
):
# ### SECTION ### Call generic function to get the object
return handle_get_obj_id(
obj_type_l1=obj_type_l1,
obj_type_l2=obj_type_l2,
obj_type_l3=obj_type_l3,
obj_id=obj_id,
use_alt_table=use_alt_table,
use_alt_base=use_alt_base,
commons=commons,
)
def handle_get_obj_id(
obj_type_l1: str,
obj_type_l2: Optional[str] = None,
obj_type_l3: Optional[str] = None,
obj_id: str = None,
use_alt_table: bool = False,
use_alt_base: bool = False,
commons: Common_Route_Params = None,
):
"""
Simple select object type with an ID:
- **obj_type_l1, obj_type_l2, obj_type_l3**:
@@ -614,17 +830,13 @@ async def get_obj(
return mk_resp(data=False, status_code=404, response=commons.response)
# ### BEGIN ### API CRUD ### patch_obj() ###
# ### BEGIN ### API CRUD ### patch_obj_lx() ###
# Updated 2024-03-08
@router.patch('/{obj_type_l1}/{obj_id}')
@router.patch('/{obj_type_l1}/{obj_type_l2}/{obj_id}')
@router.patch('/{obj_type_l1}/{obj_type_l2}/{obj_type_l3}/{obj_id}')
async def patch_obj(
async def patch_obj_l1(
crud: Api_Crud_Base,
obj_type_l1: str = Path(min_length=2, max_length=50),
obj_id: str = Path(min_length=11, max_length=22),
obj_type_l2: str = None,
obj_type_l3: str = None,
run_safety_check: bool = True,
@@ -637,6 +849,99 @@ async def patch_obj(
commons: Common_Route_Params = Depends(common_route_params),
):
# ### SECTION ### Call generic function to patch the object
return handle_patch_obj(
crud=crud,
obj_type_l1=obj_type_l1,
obj_id=obj_id,
run_safety_check=run_safety_check,
commons=commons,
)
@router.patch('/{obj_type_l1}/{obj_type_l2}/{obj_id}')
async def patch_obj_l2(
crud: Api_Crud_Base,
obj_type_l1: str = Path(min_length=2, max_length=50),
obj_type_l2: str = Path(min_length=2, max_length=50),
obj_id: str = Path(min_length=11, max_length=22),
run_safety_check: bool = True,
# for_obj_type: Optional[str] = Query(None, max_length=50),
# for_obj_id: Optional[str] = Query(None, max_length=22),
# The view name will be prefixed with "v_" and must be a valid view name based on the object type name from the URL. obj_type_l1, obj_type_l2, obj_type_l3 combined below as obj_name
return_obj: Optional[bool] = True, # I am not sure how to make this work yet. -2024-03-08
obj_v_name: Optional[str] = None, # Use view name to help return the object type. -2024-03-08
commons: Common_Route_Params = Depends(common_route_params),
):
# ### SECTION ### Call generic function to patch the object
return handle_patch_obj(
crud=crud,
obj_type_l1=obj_type_l1,
obj_id=obj_id,
obj_type_l2=obj_type_l2,
run_safety_check=run_safety_check,
commons=commons,
)
@router.patch('/{obj_type_l1}/{obj_type_l2}/{obj_type_l3}/{obj_id}')
async def patch_obj_l3(
crud: Api_Crud_Base,
obj_type_l1: str = Path(min_length=2, max_length=50),
obj_type_l2: str = Path(min_length=2, max_length=50),
obj_type_l3: str = Path(min_length=2, max_length=50),
obj_id: str = Path(min_length=11, max_length=22),
run_safety_check: bool = True,
# for_obj_type: Optional[str] = Query(None, max_length=50),
# for_obj_id: Optional[str] = Query(None, max_length=22),
# The view name will be prefixed with "v_" and must be a valid view name based on the object type name from the URL. obj_type_l1, obj_type_l2, obj_type_l3 combined below as obj_name
return_obj: Optional[bool] = True, # I am not sure how to make this work yet. -2024-03-08
obj_v_name: Optional[str] = None, # Use view name to help return the object type. -2024-03-08
commons: Common_Route_Params = Depends(common_route_params),
):
# ### SECTION ### Call generic function to patch the object
return handle_patch_obj(
crud=crud,
obj_type_l1=obj_type_l1,
obj_id=obj_id,
obj_type_l2=obj_type_l2,
obj_type_l3=obj_type_l3,
return_obj=return_obj,
obj_v_name=obj_v_name,
run_safety_check=run_safety_check,
commons=commons,
)
def handle_patch_obj(
crud: Api_Crud_Base,
obj_type_l1: str,
obj_id: str,
obj_type_l2: Optional[str] = None,
obj_type_l3: Optional[str] = None,
run_safety_check: bool = True,
return_obj: Optional[bool] = True,
obj_v_name: Optional[str] = None,
commons: Common_Route_Params = None,
):
"""
Simple patch object type with an ID:
- **obj_type_l1, obj_type_l2, obj_type_l3**:
@@ -805,13 +1110,74 @@ async def patch_obj(
# ### END ### API CRUD ### patch_obj() ###
# ### BEGIN ### API CRUD ### post_obj() ###
# ### BEGIN ### API CRUD ### post_obj_lx() ###
# Updated 2024-03-08
@router.post('/{obj_type_l1}')
async def post_obj_l1(
crud: Api_Crud_Base,
obj_type_l1: Optional[str] = Path(..., max_length=50),
# obj_id: str = Path(min_length=11, max_length=22),
run_safety_check: bool = True,
# The view name will be prefixed with "v_" and must be a valid view name based on the object type name from the URL. obj_type_l1, obj_type_l2, obj_type_l3 combined below as obj_name
# for_obj_type: Optional[str] = Query(None, max_length=50),
# for_obj_id: Optional[str] = Query(None, max_length=22),
return_obj: Optional[bool] = True,
obj_v_name: Optional[str] = None, # Use view name to help return the object type. -2024-03-08
commons: Common_Route_Params = Depends(common_route_params),
):
# ### SECTION ### Call generic function to post the object
return handle_post_obj(
crud=crud,
obj_type_l1=obj_type_l1,
run_safety_check=run_safety_check,
return_obj=return_obj,
obj_v_name=obj_v_name,
commons=commons,
)
@router.post('/{obj_type_l1}/{obj_type_l2}')
async def post_obj_l2(
crud: Api_Crud_Base,
obj_type_l1: Optional[str] = Path(..., max_length=50),
obj_type_l2: str = None,
# obj_id: str = Path(min_length=11, max_length=22),
run_safety_check: bool = True,
# The view name will be prefixed with "v_" and must be a valid view name based on the object type name from the URL. obj_type_l1, obj_type_l2, obj_type_l3 combined below as obj_name
# for_obj_type: Optional[str] = Query(None, max_length=50),
# for_obj_id: Optional[str] = Query(None, max_length=22),
return_obj: Optional[bool] = True,
obj_v_name: Optional[str] = None, # Use view name to help return the object type. -2024-03-08
commons: Common_Route_Params = Depends(common_route_params),
):
# ### SECTION ### Call generic function to post the object
return handle_post_obj(
crud=crud,
obj_type_l1=obj_type_l1,
obj_type_l2=obj_type_l2,
run_safety_check=run_safety_check,
return_obj=return_obj,
obj_v_name=obj_v_name,
commons=commons,
)
@router.post('/{obj_type_l1}/{obj_type_l2}/{obj_type_l3}')
async def post_obj(
async def post_obj_l3(
crud: Api_Crud_Base,
obj_type_l1: Optional[str] = Path(..., max_length=50),
obj_type_l2: str = None,
@@ -829,6 +1195,35 @@ async def post_obj(
commons: Common_Route_Params = Depends(common_route_params),
):
# ### SECTION ### Call generic function to post the object
return handle_post_obj(
crud=crud,
obj_type_l1=obj_type_l1,
obj_type_l2=obj_type_l2,
obj_type_l3=obj_type_l3,
run_safety_check=run_safety_check,
return_obj=return_obj,
obj_v_name=obj_v_name,
commons=commons,
)
def handle_post_obj(
crud: Api_Crud_Base,
obj_type_l1: str,
obj_type_l2: Optional[str] = None,
obj_type_l3: Optional[str] = None,
run_safety_check: bool = True,
return_obj: Optional[bool] = True,
obj_v_name: Optional[str] = None,
commons: Common_Route_Params = None,
):
"""
Simple post object type:
- **obj_type_l1, obj_type_l2, obj_type_l3**:
@@ -1014,12 +1409,58 @@ async def post_obj(
# ### END ### API CRUD ### post_obj() ###
# ### BEGIN ### API CRUD ### delete_obj() ###
# ### BEGIN ### API CRUD ### delete_obj_lx() ###
# Updated 2023-07-10
@router.delete('/{obj_type_l1}/{obj_id}')
async def delete_obj_l1(
obj_type_l1: str=None,
obj_id: str=None,
method: str = 'delete', # None, delete, disable, hide
# x_account_id: str = Header(...),
# response: Response = Response,
commons: Common_Route_Params = Depends(common_route_params),
):
# ### SECTION ### Call generic function to delete the object
return handle_delete_obj(
obj_type_l1=obj_type_l1,
obj_id=obj_id,
method=method,
commons=commons,
)
@router.delete('/{obj_type_l1}/{obj_type_l2}/{obj_id}')
async def delete_obj_l2(
obj_type_l1: str=None,
obj_type_l2: str=None,
obj_id: str=None,
method: str = 'delete', # None, delete, disable, hide
# x_account_id: str = Header(...),
# response: Response = Response,
commons: Common_Route_Params = Depends(common_route_params),
):
# ### SECTION ### Call generic function to delete the object
return handle_delete_obj(
obj_type_l1=obj_type_l1,
obj_type_l2=obj_type_l2,
obj_id=obj_id,
method=method,
commons=commons,
)
@router.delete('/{obj_type_l1}/{obj_type_l2}/{obj_type_l3}/{obj_id}')
async def delete_obj(
async def delete_obj_l3(
obj_type_l1: str=None,
obj_type_l2: str=None,
obj_type_l3: str=None,
@@ -1032,6 +1473,32 @@ async def delete_obj(
commons: Common_Route_Params = Depends(common_route_params),
):
# ### SECTION ### Call generic function to delete the object
return handle_delete_obj(
obj_type_l1=obj_type_l1,
obj_type_l2=obj_type_l2,
obj_type_l3=obj_type_l3,
obj_id=obj_id,
method=method,
commons=commons,
)
def handle_delete_obj(
obj_type_l1: str=None,
obj_type_l2: str=None,
obj_type_l3: str=None,
obj_id: str=None,
method: str = 'delete', # None, delete, disable, hide
# x_account_id: str = Header(...),
# response: Response = Response,
commons: Common_Route_Params = None,
):
"""
Simple delete object type with an ID:
- **obj_type_l1, obj_type_l2, obj_type_l3**:

224
app/routers/api_crud_v2.py
View File

@@ -18,11 +18,185 @@ router = APIRouter()
# Working on the basic API CRUD - STI 2021-03-08
# obj_type_l1: str = Path(min_length=2, max_length=50),
# obj_type_l2: str = Path(min_length=2, max_length=50),
# obj_type_l3: str = Path(min_length=2, max_length=50),
# for_obj_type: Optional[str] = Query(None, max_length=50),
# for_obj_id: Optional[str] = Query(None, max_length=22),
# tbl_alt: Optional[str] = Query('default', max_length=50), # This is used as a lookup for the real SQL database table or view name to use.
# mdl_alt: Optional[str] = Query('default', max_length=50), # This is used as a lookup for the real Python Pydantic model name to use.
# # use_alt_table: bool = False, # NOTE: This will use table_name_alt if they exist. -2023-11-17
# # use_alt_base: bool = False, # NOTE: This will use base_name_alt if they exist. -2023-11-17
# # field_qry_li: str = Query(None, max_length=150), # JSON formatted key value pair list of fields to search.
# # fulltext_qry_li: str = Query(None, max_length=150), # JSON formatted key value pair list of fields to search.
# # fulltext_qry_field_li: str = Header(None), # Json formatted string list of fields to search. It is not ideal that this is in the header. Need a better option, but this is currently a GET request.
# # fulltext_qry_str: str = Query(None, max_length=150),
# hidden: str = 'not_hidden', # hidden, not_hidden, all,
# # order_by_li: dict = None,
# order_by_li: str = Header(None), # JSON formatted string in a key value format. It is not ideal that this is in the header. Need a better option, but this is currently a GET request.
# # dh_order_by_li: str = Header(None),
# # dh_testing: str = Header(None),
# # h_order_by_li: str = Header(None),
# # h_testing: str = Header(None),
# # include: Optional[list] = [],
# # exclude: Optional[list] = [],
# # exclude_none: Optional[bool] = True,
# # Get the "json" param from the query string. This is a JSON formatted string of the data to be inserted.
# jp: Optional[Union[str, None]] = None,
# file_type: str = 'CSV', # CSV, Excel
# return_file: Optional[bool] = False,
# commons: Common_Route_Params = Depends(common_route_params),
# Updated 2023-07-06
@router.get('/{obj_type_l1}/list')
async def get_obj_li_l1(
obj_type_l1: str = Path(min_length=2, max_length=50),
for_obj_type: Optional[str] = Query(None, max_length=50),
for_obj_id: Optional[str] = Query(None, max_length=22),
tbl_alt: Optional[str] = Query('default', max_length=50), # This is used as a lookup for the real SQL database table or view name to use.
mdl_alt: Optional[str] = Query('default', max_length=50), # This is used as a lookup for the real Python Pydantic model name to use.
# use_alt_table: bool = False, # NOTE: This will use table_name_alt if they exist. -2023-11-17
# use_alt_base: bool = False, # NOTE: This will use base_name_alt if they exist. -2023-11-17
# field_qry_li: str = Query(None, max_length=150), # JSON formatted key value pair list of fields to search.
# fulltext_qry_li: str = Query(None, max_length=150), # JSON formatted key value pair list of fields to search.
# fulltext_qry_field_li: str = Header(None), # Json formatted string list of fields to search. It is not ideal that this is in the header. Need a better option, but this is currently a GET request.
# fulltext_qry_str: str = Query(None, max_length=150),
hidden: str = 'not_hidden', # hidden, not_hidden, all,
# order_by_li: dict = None,
order_by_li: str = Header(None), # JSON formatted string in a key value format. It is not ideal that this is in the header. Need a better option, but this is currently a GET request.
# dh_order_by_li: str = Header(None),
# dh_testing: str = Header(None),
# h_order_by_li: str = Header(None),
# h_testing: str = Header(None),
# include: Optional[list] = [],
# exclude: Optional[list] = [],
# exclude_none: Optional[bool] = True,
# Get the "json" param from the query string. This is a JSON formatted string of the data to be inserted.
jp: Optional[Union[str, None]] = None,
file_type: str = 'CSV', # CSV, Excel
return_file: Optional[bool] = False,
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Call generic function to get the list of objects
return handle_get_obj_li(
obj_type_l1=obj_type_l1,
for_obj_type=for_obj_type,
for_obj_id=for_obj_id,
tbl_alt=tbl_alt,
mdl_alt=mdl_alt,
hidden=hidden,
order_by_li=order_by_li,
jp=jp,
file_type=file_type,
return_file=return_file,
commons=commons,
)
@router.get('/{obj_type_l1}/{obj_type_l2}/list')
async def get_obj_li_l2(
obj_type_l1: str = Path(min_length=2, max_length=50),
obj_type_l2: str = Path(min_length=2, max_length=50),
for_obj_type: Optional[str] = Query(None, max_length=50),
for_obj_id: Optional[str] = Query(None, max_length=22),
tbl_alt: Optional[str] = Query('default', max_length=50), # This is used as a lookup for the real SQL database table or view name to use.
mdl_alt: Optional[str] = Query('default', max_length=50), # This is used as a lookup for the real Python Pydantic model name to use.
# use_alt_table: bool = False, # NOTE: This will use table_name_alt if they exist. -2023-11-17
# use_alt_base: bool = False, # NOTE: This will use base_name_alt if they exist. -2023-11-17
# field_qry_li: str = Query(None, max_length=150), # JSON formatted key value pair list of fields to search.
# fulltext_qry_li: str = Query(None, max_length=150), # JSON formatted key value pair list of fields to search.
# fulltext_qry_field_li: str = Header(None), # Json formatted string list of fields to search. It is not ideal that this is in the header. Need a better option, but this is currently a GET request.
# fulltext_qry_str: str = Query(None, max_length=150),
hidden: str = 'not_hidden', # hidden, not_hidden, all,
# order_by_li: dict = None,
order_by_li: str = Header(None), # JSON formatted string in a key value format. It is not ideal that this is in the header. Need a better option, but this is currently a GET request.
# dh_order_by_li: str = Header(None),
# dh_testing: str = Header(None),
# h_order_by_li: str = Header(None),
# h_testing: str = Header(None),
# include: Optional[list] = [],
# exclude: Optional[list] = [],
# exclude_none: Optional[bool] = True,
# Get the "json" param from the query string. This is a JSON formatted string of the data to be inserted.
jp: Optional[Union[str, None]] = None,
file_type: str = 'CSV', # CSV, Excel
return_file: Optional[bool] = False,
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Call generic function to get the list of objects
return handle_get_obj_li(
obj_type_l1=obj_type_l1,
obj_type_l2=obj_type_l2,
for_obj_type=for_obj_type,
for_obj_id=for_obj_id,
tbl_alt=tbl_alt,
mdl_alt=mdl_alt,
hidden=hidden,
order_by_li=order_by_li,
jp=jp,
file_type=file_type,
return_file=return_file,
commons=commons,
)
@router.get('/{obj_type_l1}/{obj_type_l2}/{obj_type_l3}/list')
async def get_obj_li(
async def get_obj_li_l3(
obj_type_l1: str = Path(min_length=2, max_length=50),
obj_type_l2: str = Path(min_length=2, max_length=50),
obj_type_l3: str = Path(min_length=2, max_length=50),
@@ -67,6 +241,54 @@ async def get_obj_li(
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Call generic function to get the list of objects
return handle_get_obj_li(
obj_type_l1=obj_type_l1,
obj_type_l2=obj_type_l2,
obj_type_l3=obj_type_l3,
for_obj_type=for_obj_type,
for_obj_id=for_obj_id,
tbl_alt=tbl_alt,
mdl_alt=mdl_alt,
hidden=hidden,
order_by_li=order_by_li,
jp=jp,
file_type=file_type,
return_file=return_file,
commons=commons,
)
def handle_get_obj_li(
obj_type_l1: str,
obj_type_l2: Optional[str] = None,
obj_type_l3: Optional[str] = None,
for_obj_type: Optional[str] = None,
for_obj_id: Optional[str] = None,
tbl_alt: Optional[str] = 'default',
mdl_alt: Optional[str] = 'default',
hidden: str = 'not_hidden',
order_by_li: Optional[str] = None,
jp: Optional[Union[str, None]] = None,
file_type: str = 'CSV',
return_file: Optional[bool] = False,
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
import urllib
# This should be a dict list of fields with a list of values to search for using FULLTEXT.

48
app/routers/data_store.py
View File

@@ -141,16 +141,58 @@ async def get_data_store_obj(
# Lookup using: for_type and for_id > account_id > data_store_code
# This is a nice way to have global default data along with account and object specific data.
# Updated 2023-05-22
@router.get('/data_store/code/{data_store_code}/{for_type}/{for_id}', response_model=Resp_Body_Base)
@router.get('/data_store/code/{data_store_code}', response_model=Resp_Body_Base)
async def get_data_store_obj_w_code(
async def get_data_store_obj_w_code_path(
data_store_code: str = Path(min_length=3, max_length=50),
for_type: Optional[str] = Path(min_length=1, max_length=25),
for_id: Optional[str] = Path(min_length=11, max_length=22),
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
log.debug('Using path parameters')
# ### SECTION ### Call generic function to get the data_store object
return handle_get_data_store_obj_w_code(
data_store_code = data_store_code,
for_type = for_type,
for_id = for_id,
commons = commons,
)
@router.get('/data_store/code/{data_store_code}', response_model=Resp_Body_Base)
async def get_data_store_obj_w_code_query(
data_store_code: str = Path(min_length=3, max_length=50),
for_type: Optional[str] = Query(None, min_length=1, max_length=25),
for_id: Optional[str] = Query(None, min_length=11, max_length=22),
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
log.debug('Using query parameters')
# ### SECTION ### Call generic function to get the data_store object
return handle_get_data_store_obj_w_code(
data_store_code = data_store_code,
for_type = for_type,
for_id = for_id,
commons = commons,
)
def handle_get_data_store_obj_w_code(
data_store_code: str,
for_type: Optional[str],
for_id: Optional[str],
commons: Common_Route_Params,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
log.debug(commons.x_account_id_random)