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

Working on general POST, PATCH, SELECT list, SELECT, and DELETE template functions.

Browse Source
This commit is contained in:
Scott Idem
2021-03-09 17:24:29 -05:00
parent e30b5a9092
commit 768159918f
11 changed files with 736 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
app/main.py
View File

@@ -18,12 +18,12 @@ from .lib_general import *
from .log import *
# Import the routers here first:
from .routers import api_crud, address, contact, websockets # , items, journals, users
from .routers import api_crud, account, address, contact, page, person, post, post_comment, product, user, websockets # , items, journals
from .db_sql import db
print('### **** *** ** * The Aether FastAPI API app is starting... * ** *** **** ###')
print('### **** *** ** * The Aether FastAPI API app is loading... * ** *** **** ###')
#log = logging.getLogger('root')
@@ -53,6 +53,14 @@ app.include_router(
#dependencies=[Depends(get_account_header)],
#responses={404: {'description': 'Not found'}},
)
app.include_router(
account.router,
prefix='/account',
tags=['Account'],
#dependencies=[Depends(get_token_header)],
#dependencies=[Depends(get_account_header)],
#responses={404: {'description': 'Not found'}},
)
app.include_router(
address.router,
prefix='/address',
@@ -69,6 +77,54 @@ app.include_router(
#dependencies=[Depends(get_account_header)],
#responses={404: {'description': 'Not found'}},
)
app.include_router(
page.router,
prefix='/page',
tags=['Page'],
#dependencies=[Depends(get_token_header)],
#dependencies=[Depends(get_account_header)],
#responses={404: {'description': 'Not found'}},
)
app.include_router(
person.router,
prefix='/person',
tags=['Person'],
#dependencies=[Depends(get_token_header)],
#dependencies=[Depends(get_account_header)],
#responses={404: {'description': 'Not found'}},
)
app.include_router(
post.router,
prefix='/post',
tags=['Post'],
#dependencies=[Depends(get_token_header)],
#dependencies=[Depends(get_account_header)],
#responses={404: {'description': 'Not found'}},
)
app.include_router(
post_comment.router,
prefix='/post_comment',
tags=['Post_Comment'],
#dependencies=[Depends(get_token_header)],
#dependencies=[Depends(get_account_header)],
#responses={404: {'description': 'Not found'}},
)
app.include_router(
product.router,
prefix='/product',
tags=['Product'],
#dependencies=[Depends(get_token_header)],
#dependencies=[Depends(get_account_header)],
#responses={404: {'description': 'Not found'}},
)
app.include_router(
user.router,
prefix='/user',
tags=['User'],
#dependencies=[Depends(get_token_header)],
#dependencies=[Depends(get_account_header)],
#responses={404: {'description': 'Not found'}},
)
app.include_router(
websockets.router,
#prefix='/websocket',
@@ -106,7 +162,7 @@ async def startup():
log.setLevel(logging.INFO) # DEBUG, INFO, WARN, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
log.info('FastAPI app is starting up...')
log.info('The Aether FastAPI API app is starting up...')
#await database.connect()
@@ -115,7 +171,7 @@ async def shutdown():
log.setLevel(logging.INFO) # DEBUG, INFO, WARN, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
log.info('FastAPI app is shutting down...')
log.info('The Aether FastAPI API app is shutting down...')
#await database.disconnect()

70
app/routers/account.py Normal file
View File

@@ -0,0 +1,70 @@
import datetime
#from datetime import datetime, time, timedelta
from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, status
from pydantic import BaseModel, EmailStr, Field
from typing import Dict, List, Optional, Set, Union
from ..lib_general import *
from ..log import *
from app.config import settings
from app.db_sql import *
from .api_crud import post_obj_template, patch_obj_template
from ..models.account_model import Account_Base
from ..models.response_model import *
router = APIRouter()
@router.post('')
async def post_account_obj(
obj: Account_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_insert = 'account'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
result = post_obj_template(
table_name_insert=table_name_insert,
data=obj_data_dict,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result
@router.patch('/{obj_id}')
async def patch_account_obj(
obj_id: str,
obj: Account_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_update = 'account'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=table_name_update)
obj_data_dict['id_random'] = obj_id
result = patch_obj_template(
table_name_update=table_name_update,
data=obj_data_dict,
obj_id=obj_id,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result

82
app/routers/address.py
View File

@@ -9,7 +9,7 @@ from ..log import *
from app.config import settings
from app.db_sql import *
from .api_crud import post_obj_template, patch_obj_template
from .api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
from ..models.address_model import Address_Base
from ..models.response_model import *
@@ -26,25 +26,24 @@ async def post_address_obj(
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_insert = 'address'
obj_type = 'address'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
result = post_obj_template(
table_name_insert=table_name_insert,
obj_type=obj_type,
data=obj_data_dict,
return_obj=True,
by_alias=True,
exclude_unset=True
exclude_unset=True,
)
return result
@router.patch('/{obj_id}')
async def patch_address_obj(
obj_id: str,
obj_id: str = Query(None, min_length=1, max_length=22),
obj: Address_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
@@ -54,17 +53,76 @@ async def patch_address_obj(
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_update = 'address'
obj_type = 'address'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=table_name_update)
obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=obj_type)
obj_data_dict['id_random'] = obj_id
result = patch_obj_template(
table_name_update=table_name_update,
obj_type=obj_type,
data=obj_data_dict,
obj_id=obj_id,
return_obj=True,
by_alias=True,
exclude_unset=True
exclude_unset=True,
)
return result
@router.get('/list')
async def get_address_obj_li(
for_obj_type: Optional[str] = Query(None, min_length=2, max_length=50),
for_obj_id: Optional[str] = Query(None, min_length=1, max_length=22),
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'address'
result = get_obj_li_template(
obj_type=obj_type,
for_obj_type=for_obj_type,
for_obj_id=for_obj_id,
by_alias=True,
exclude_unset=True,
)
return result
@router.get('/{obj_id}')
async def get_address_obj(
obj_id: str = Query(None, min_length=1, max_length=22),
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'address'
#obj_id = redis_lookup_id_random(record_id_random=obj_id, table_name=obj_type)
result = get_obj_template(
obj_type=obj_type,
obj_id=obj_id,
by_alias=True,
exclude_unset=True,
)
return result
@router.delete('/{obj_id}')
async def delete_address_obj(
obj_id: str = Query(None, min_length=1, max_length=22),
x_account_id: str = Header(...),
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'address'
#obj_id = redis_lookup_id_random(record_id_random=obj_id, table_name=obj_type)
result = delete_obj_template(
obj_type=obj_type,
obj_id=obj_id,
)
return result

108
app/routers/api_crud.py
View File

@@ -127,7 +127,7 @@ async def get_obj_li(
obj_type_l1: str=None,
obj_type_l2: str=None,
obj_type_l3: str=None,
obj_id: str=None,
#obj_id: str=None,
for_obj_type: Optional[str] = Query(None, max_length=50),
for_obj_id: Optional[str] = Query(None, max_length=22),
x_account_id: str = Header(...),
@@ -141,7 +141,7 @@ async def get_obj_li(
debug_data['obj_type_l1'] = obj_type_l1
debug_data['obj_type_l2'] = obj_type_l2
debug_data['obj_type_l3'] = obj_type_l3
debug_data['obj_id'] = obj_id
#debug_data['obj_id'] = obj_id
debug_data['for_obj_type'] = for_obj_type
debug_data['for_obj_id'] = for_obj_id
@@ -349,7 +349,7 @@ async def delete_obj(
def post_obj_template(
table_name_insert:str=None,
obj_type:str=None,
data:dict=None,
return_obj:bool=True,
by_alias:bool=True,
@@ -360,8 +360,9 @@ def post_obj_template(
obj_data_dict = data
obj_data = lookup_id_random_pop(obj_data_dict)
table_name_select = obj_type_li[table_name_insert]['table_name']
base_name = obj_type_li[table_name_insert]['base_name']
table_name_insert = obj_type
table_name_select = obj_type_li[obj_type]['table_name']
base_name = obj_type_li[obj_type]['base_name']
if sql_insert_result := sql_insert(table_name=table_name_insert, data=obj_data):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
@@ -384,7 +385,7 @@ def post_obj_template(
def patch_obj_template(
table_name_update:str=None,
obj_type:str=None,
data:dict=None,
obj_id:str=None,
return_obj:bool=True,
@@ -395,12 +396,14 @@ def patch_obj_template(
log.debug(locals())
obj_data_dict = data
#obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=table_name_update)
#obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=obj_type)
obj_data_dict['id_random'] = obj_id # NOTE: Adding this in so the id_random is NOT updated
log.debug(obj_data_dict)
obj_data = lookup_id_random_pop(obj_data_dict)
table_name_select = obj_type_li[table_name_update]['table_name']
base_name = obj_type_li[table_name_update]['base_name']
table_name_update = obj_type
table_name_select = obj_type_li[obj_type]['table_name']
base_name = obj_type_li[obj_type]['base_name']
if sql_update_result := sql_update(table_name=table_name_update, data=obj_data):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
@@ -420,4 +423,91 @@ def patch_obj_template(
return mk_resp(data=resp_data)
else:
log.debug(sql_select_result)
return mk_resp(data=False, status_code=404)
def get_obj_li_template(
obj_type: str = Query(None, max_length=50),
for_obj_type: Optional[str] = Query(None, max_length=50),
for_obj_id: Optional[str] = Query(None, max_length=22),
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_select = obj_type_li[obj_type]['table_name']
if for_obj_type and for_obj_id:
for_obj_id = redis_lookup_id_random(record_id_random=for_obj_id, table_name=for_obj_type)
field_name = f'{for_obj_type}_id'
sql_result = sql_select(table_name=table_name_select, field_name=field_name, field_value=for_obj_id)
else:
sql_result = sql_select(table_name=table_name_select)
log.debug(sql_result)
base_name = obj_type_li[obj_type]['base_name']
resp_data_li = []
for record in sql_result:
resp_data = base_name(**record).dict(by_alias=by_alias, exclude_unset=exclude_unset)
resp_data_li.append(resp_data)
return mk_resp(data=resp_data_li)
def get_obj_template(
obj_type: str = Query(None, max_length=50),
obj_id: str = Query(None, max_length=22),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
debug_data = {}
debug_data['obj_type'] = obj_type
debug_data['obj_id'] = obj_id
log.debug(debug_data)
table_name_select = obj_type_li[obj_type]['table_name']
# NOTE: Add a check for the object ID... assuming it is a random ID string for now.
if sql_result := sql_select(table_name=table_name_select, record_id_random=obj_id):
log.debug(sql_result)
base_name = obj_type_li[obj_type]['base_name']
resp_data = base_name(**sql_result).dict(by_alias=by_alias, exclude_unset=exclude_unset)
return mk_resp(data=resp_data) #, details=debug_data)
else:
log.debug(sql_result)
return mk_resp(data=False, status_code=404)
def delete_obj_template(
obj_type: str = Query(None, max_length=50),
obj_id: str = Query(None, max_length=22),
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
debug_data = {}
debug_data['obj_type'] = obj_type
debug_data['obj_id'] = obj_id
log.debug(debug_data)
table_name_delete = obj_type # NOTE: Not using the table name from the object type list because it may be a view (v_*).
# NOTE: Add a check for the object ID... assuming it is a random ID string for now.
if sql_result := sql_delete(table_name=table_name_delete, record_id_random=obj_id):
log.debug(sql_result)
return mk_resp(data=True)
else:
log.debug(sql_result)
return mk_resp(data=False, status_code=404)

50
app/routers/contact.py
View File

@@ -9,7 +9,7 @@ from ..log import *
from app.config import settings
from app.db_sql import *
from .api_crud import patch_obj_template
from .api_crud import post_obj_template, patch_obj_template
from ..models.contact_model import Contact_Base
from ..models.response_model import *
@@ -18,60 +18,44 @@ from ..models.response_model import *
router = APIRouter()
# Working on the contact API CRUD - STI 2021-03-08
@router.post('')
async def post_contact_obj(
contact: Contact_Base,
obj: Contact_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_insert = 'contact'
obj_data_dict = contact.dict(by_alias=False, exclude_unset=True)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(obj_data_dict)
obj_data = lookup_id_random_pop(obj_data_dict)
base_name = Contact_Base
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
result = post_obj_template(
table_name_insert=table_name_insert,
data=obj_data_dict,
return_obj=True,
by_alias=True,
exclude_unset=True
)
if sql_insert_result := sql_insert(table_name=table_name_insert, data=obj_data):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(sql_insert_result)
obj_id = sql_insert_result
else:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(sql_insert_result)
return mk_resp(data=False, status_code=400)
table_name_select = 'v_contact'
sql_select_result = sql_select(table_name=table_name_select, record_id=obj_id)
log.debug(sql_select_result)
resp_data = base_name(**sql_select_result).dict(by_alias=by_alias, exclude_unset=exclude_unset)
return mk_resp(data=resp_data)
return result
@router.patch('/{contact_id}')
@router.patch('/{obj_id}')
async def patch_contact_obj(
contact_id: str,
contact: Contact_Base,
obj_id: str,
obj: Contact_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_update = 'contact'
obj_id = contact_id # actually id_random
obj_data_dict = contact.dict(by_alias=False, exclude_unset=True)
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=table_name_update)
obj_data_dict['id_random'] = obj_id
result = patch_obj_template(

70
app/routers/page.py Normal file
View File

@@ -0,0 +1,70 @@
import datetime
#from datetime import datetime, time, timedelta
from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, status
from pydantic import BaseModel, EmailStr, Field
from typing import Dict, List, Optional, Set, Union
from ..lib_general import *
from ..log import *
from app.config import settings
from app.db_sql import *
from .api_crud import post_obj_template, patch_obj_template
from ..models.page_model import Page_Base
from ..models.response_model import *
router = APIRouter()
@router.post('')
async def post_page_obj(
obj: Page_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_insert = 'page'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
result = post_obj_template(
table_name_insert=table_name_insert,
data=obj_data_dict,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result
@router.patch('/{obj_id}')
async def patch_page_obj(
obj_id: str,
obj: Page_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_update = 'page'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=table_name_update)
obj_data_dict['id_random'] = obj_id
result = patch_obj_template(
table_name_update=table_name_update,
data=obj_data_dict,
obj_id=obj_id,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result

70
app/routers/person.py Normal file
View File

@@ -0,0 +1,70 @@
import datetime
#from datetime import datetime, time, timedelta
from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, status
from pydantic import BaseModel, EmailStr, Field
from typing import Dict, List, Optional, Set, Union
from ..lib_general import *
from ..log import *
from app.config import settings
from app.db_sql import *
from .api_crud import post_obj_template, patch_obj_template
from ..models.person_model import Person_Base
from ..models.response_model import *
router = APIRouter()
@router.post('')
async def post_person_obj(
obj: Person_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_insert = 'person'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
result = post_obj_template(
table_name_insert=table_name_insert,
data=obj_data_dict,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result
@router.patch('/{obj_id}')
async def patch_person_obj(
obj_id: str,
obj: Person_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_update = 'person'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=table_name_update)
obj_data_dict['id_random'] = obj_id
result = patch_obj_template(
table_name_update=table_name_update,
data=obj_data_dict,
obj_id=obj_id,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result

70
app/routers/post.py Normal file
View File

@@ -0,0 +1,70 @@
import datetime
#from datetime import datetime, time, timedelta
from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, status
from pydantic import BaseModel, EmailStr, Field
from typing import Dict, List, Optional, Set, Union
from ..lib_general import *
from ..log import *
from app.config import settings
from app.db_sql import *
from .api_crud import post_obj_template, patch_obj_template
from ..models.post_model import Post_Base
from ..models.response_model import *
router = APIRouter()
@router.post('')
async def post_post_obj(
obj: Post_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_insert = 'post'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
result = post_obj_template(
table_name_insert=table_name_insert,
data=obj_data_dict,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result
@router.patch('/{obj_id}')
async def patch_post_obj(
obj_id: str,
obj: Post_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_update = 'post'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=table_name_update)
obj_data_dict['id_random'] = obj_id
result = patch_obj_template(
table_name_update=table_name_update,
data=obj_data_dict,
obj_id=obj_id,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result

70
app/routers/post_comment.py Normal file
View File

@@ -0,0 +1,70 @@
import datetime
#from datetime import datetime, time, timedelta
from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, status
from pydantic import BaseModel, EmailStr, Field
from typing import Dict, List, Optional, Set, Union
from ..lib_general import *
from ..log import *
from app.config import settings
from app.db_sql import *
from .api_crud import post_obj_template, patch_obj_template
from ..models.post_comment_model import Post_Comment_Base
from ..models.response_model import *
router = APIRouter()
@router.post('')
async def post_post_comment_obj(
obj: Post_Comment_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_insert = 'post_comment'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
result = post_obj_template(
table_name_insert=table_name_insert,
data=obj_data_dict,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result
@router.patch('/{obj_id}')
async def patch_post_comment_obj(
obj_id: str,
obj: Post_Comment_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_update = 'post_comment'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=table_name_update)
obj_data_dict['id_random'] = obj_id
result = patch_obj_template(
table_name_update=table_name_update,
data=obj_data_dict,
obj_id=obj_id,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result

70
app/routers/product.py Normal file
View File

@@ -0,0 +1,70 @@
import datetime
#from datetime import datetime, time, timedelta
from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, status
from pydantic import BaseModel, EmailStr, Field
from typing import Dict, List, Optional, Set, Union
from ..lib_general import *
from ..log import *
from app.config import settings
from app.db_sql import *
from .api_crud import post_obj_template, patch_obj_template
from ..models.product_model import Product_Base
from ..models.response_model import *
router = APIRouter()
@router.post('')
async def post_product_obj(
obj: Product_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_insert = 'product'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
result = post_obj_template(
table_name_insert=table_name_insert,
data=obj_data_dict,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result
@router.patch('/{obj_id}')
async def patch_product_obj(
obj_id: str,
obj: Product_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_update = 'product'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=table_name_update)
obj_data_dict['id_random'] = obj_id
result = patch_obj_template(
table_name_update=table_name_update,
data=obj_data_dict,
obj_id=obj_id,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result

70
app/routers/user.py Normal file
View File

@@ -0,0 +1,70 @@
import datetime
#from datetime import datetime, time, timedelta
from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, status
from pydantic import BaseModel, EmailStr, Field
from typing import Dict, List, Optional, Set, Union
from ..lib_general import *
from ..log import *
from app.config import settings
from app.db_sql import *
from .api_crud import post_obj_template, patch_obj_template
from ..models.user_model import User_Base
from ..models.response_model import *
router = APIRouter()
@router.post('')
async def post_user_obj(
obj: User_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_insert = 'user'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
result = post_obj_template(
table_name_insert=table_name_insert,
data=obj_data_dict,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result
@router.patch('/{obj_id}')
async def patch_user_obj(
obj_id: str,
obj: User_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
table_name_update = 'user'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=table_name_update)
obj_data_dict['id_random'] = obj_id
result = patch_obj_template(
table_name_update=table_name_update,
data=obj_data_dict,
obj_id=obj_id,
return_obj=True,
by_alias=True,
exclude_unset=True
)
return result