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

Working finally getting this to interact with the actual front end.

Browse Source
This commit is contained in:
Scott Idem
2021-03-15 20:08:28 +00:00
parent 4ba34d594d
commit bccb8370af
9 changed files with 284 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
app/routers/flask_cfg.py Normal file
View File

@@ -0,0 +1,134 @@
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 app.config import settings
from app.db_sql import *
from .api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
from ..models.response_model import *
from ..models.flask_cfg_model import Flask_Cfg_Base
router = APIRouter()
@router.post('', response_model=Resp_Body_Base)
async def post_flask_cfg_obj(
obj: Flask_Cfg_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
include: Optional[list] = [],
exclude: Optional[list] = [],
exclude_unset: Optional[bool] = True,
exclude_none: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'flask_cfg'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
result = post_obj_template(
obj_type=obj_type,
data=obj_data_dict,
return_obj=True,
by_alias=True,
exclude_unset=True,
)
return result
@router.patch('/{obj_id}', response_model=Resp_Body_Base)
async def patch_flask_cfg_obj(
obj_id: str = Query(..., min_length=1, max_length=22),
obj: Flask_Cfg_Base = None,
x_account_id: Optional[str] = Header(..., ),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
include: Optional[list] = [],
exclude: Optional[list] = [],
exclude_unset: Optional[bool] = True,
exclude_none: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'flask_cfg'
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=obj_type)
obj_data_dict['id_random'] = obj_id
result = patch_obj_template(
obj_type=obj_type,
data=obj_data_dict,
obj_id=obj_id,
return_obj=True,
by_alias=True,
exclude_unset=True,
)
return result
@router.get('/list', response_model=Resp_Body_Base)
async def get_flask_cfg_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,
include: Optional[list] = [],
exclude: Optional[list] = [],
exclude_unset: Optional[bool] = True,
exclude_none: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'flask_cfg'
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('/{flask_cfg_id}', response_model=Resp_Body_Base)
async def get_flask_cfg_obj(
flask_cfg_id: int = None,
#flask_cfg_id: str = Query(..., min_length=1, max_length=22),
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
include: Optional[list] = [],
exclude: Optional[list] = [],
exclude_unset: Optional[bool] = True,
exclude_none: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if sql_select_result := sql_select(table_name='flask_cfg', record_id=flask_cfg_id):
return mk_resp(data=sql_select_result)
else:
return mk_resp(data=None, status_code=404)
@router.delete('/{obj_id}', response_model=Resp_Body_Base)
async def delete_flask_cfg_obj(
obj_id: str = Query(..., 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 = 'flask_cfg'
result = delete_obj_template(
obj_type=obj_type,
obj_id=obj_id,
)
return result

70
app/routers/user.py
View File

@@ -1,4 +1,4 @@
import datetime
import datetime, pytz, time
#from datetime import datetime, time, timedelta
from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, status
from pydantic import BaseModel, EmailStr, Field
@@ -68,6 +68,74 @@ async def patch_user_obj(
return result
# This will look up a user based on the auth key given
# This can only be done once per key. It will be deleted if found
# A new one will need to be requested for a particular user each time
@router.get('/authenticate/auth_key/{auth_key}', response_model=Resp_Body_Base)
async def auth_key_get_user_obj(
auth_key: str = Query(..., min_length=11, max_length=22),
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
exclude_none: Optional[bool] = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if sql_select_result := sql_select(table_name='user', field_name='auth_key', field_value=auth_key):
log.debug(sql_select_result)
resp_data = {}
resp_data['user_id_random'] = sql_select_result.get('id_random')
resp_data['username'] = sql_select_result.get('username')
resp_data['enable'] = sql_select_result.get('enable')
resp_data['enable_from'] = sql_select_result.get('enable_from')
resp_data['enable_to'] = sql_select_result.get('enable_to')
try:
user_obj = User_Base(**sql_select_result).dict(by_alias=by_alias, exclude_unset=exclude_unset, exclude_none=exclude_none)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(user_obj)
log.debug(user_obj.get('enable_from', None))
except ValidationError as e:
log.error(e.json())
current_utc_datetime = datetime.datetime.now(datetime.timezone.utc)
log.debug(user_obj.get('enable_from', None).astimezone(pytz.UTC))
user_enable_from = user_obj.get('enable_from', None).astimezone(pytz.UTC)
log.debug(user_enable_from)
log.debug(user_obj.get('enable_to', None))
user_enable_to = user_obj.get('enable_to', None).astimezone(pytz.UTC)
log.debug(user_enable_to)
if resp_data['enable']: pass
else:
log.info('The user account has been disabled')
if user_enable_from <= current_utc_datetime:
log.info('Enable from datetime is valid')
else:
log.info('Enable from datetime is in the future. Please wait.')
if user_enable_to >= current_utc_datetime:
log.info('Enable to datetime is valid')
else:
log.info('Enable to datetime is in the past. Your user account has been disabled.')
update_data = {}
update_data['id'] = sql_select_result.get('id')
update_data['auth_key'] = None
if sql_update_resp := sql_update(table_name='user', data=update_data):
log.info('The user record was updated with a NULL auth_key')
else:
log.info('The user record was not updated with a NULL auth_key')
log.debug(sql_update_resp)
return mk_resp(data=user_obj)
else:
return mk_resp(data=None, status_code=404)
@router.get('/list', response_model=Resp_Body_Base)
async def get_user_obj_li(
for_obj_type: Optional[str] = Query(None, min_length=2, max_length=50),