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

Working on continuing education certs

Browse Source
This commit is contained in:
Scott Idem
2021-07-28 12:33:07 -04:00
parent 5f66780522
commit e43cf35ec4
12 changed files with 930 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
app/main.py
View File

@@ -18,7 +18,7 @@ from app.lib_general import log, logging
from app.log import log
# Import the routers here first:
from app.routers import api_crud, api, account, address, archive, archive_content, contact, event, event_exhibit, event_file, event_person, event_person_detail, event_presentation, event_presenter, event_registration, event_session, flask_cfg, hosted_file, journal, journal_entry, log_client_viewing, lookup, membership_cfg, membership_group, membership_group_person, membership_person, membership_person_profile, membership_type, membership_type_person, order, order_cart, organization, page, person, post, post_comment, product, site, site_domain, user, user_person, websockets # , items, journals
from app.routers import api_crud, api, account, address, archive, archive_content, contact, cont_edu_cert_person, event, event_exhibit, event_file, event_person, event_person_detail, event_presentation, event_presenter, event_registration, event_session, flask_cfg, hosted_file, journal, journal_entry, log_client_viewing, lookup, membership_cfg, membership_group, membership_group_person, membership_person, membership_person_profile, membership_type, membership_type_person, order, order_cart, organization, page, person, post, post_comment, product, site, site_domain, user, user_person, websockets # , items, journals
from app.db_sql import db
@@ -94,6 +94,14 @@ app.include_router(
prefix='/contact',
tags=['Contact'],
)
# app.include_router(
# cont_edu_cert.router,
# tags=['Cont Edu Cert'],
# )
app.include_router(
cont_edu_cert_person.router,
tags=['Cont Edu Cert Person'],
)
app.include_router(
event.router,
# prefix='/event',

2
app/methods/address_methods.py
View File

@@ -19,7 +19,7 @@ def load_address_obj(
exclude_unset: bool = True,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all # Probably not needed for the address
) -> Address_Base|bool:
) -> Address_Base|dict|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())

96
app/methods/cont_edu_cert_methods.py Normal file
View File

@@ -0,0 +1,96 @@
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.common_field_schema import default_num_bytes
from app.models.cont_edu_cert_models import Cont_Edu_Cert_Base
# ### BEGIN ### API Cont Edu Cert Methods ### load_cont_edu_cert_obj() ###
def load_cont_edu_cert_obj(
cont_edu_cert_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
) -> Cont_Edu_Cert_Base|dict|bool:
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if cont_edu_cert_id := redis_lookup_id_random(record_id_random=cont_edu_cert_id, table_name='cont_edu_cert'): pass
else: return False
if cont_edu_cert_rec := sql_select(table_name='v_cont_edu_cert', record_id=cont_edu_cert_id): pass
else: return False
try:
cont_edu_cert_obj = Cont_Edu_Cert_Base(**cont_edu_cert_rec)
log.debug(cont_edu_cert_obj)
except ValidationError as e:
log.error(e.json())
return False
if model_as_dict:
return cont_edu_cert_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
else:
return cont_edu_cert_obj
# ### END ### API Cont Edu Cert Methods ### load_cont_edu_cert_obj() ###
# ### BEGIN ### API Cont Edu Cert Methods ### get_cont_edu_cert_rec_list() ###
def get_cont_edu_cert_rec_list(
account_id: str,
limit: int = 1000,
enabled: str = 'enabled', # enabled, disabled, all
) -> list|bool:
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return False
data = {}
data['account_id'] = account_id
sql_account_id = f'`tbl`.account_id = :account_id'
if enabled in ['enabled', 'disabled', 'all']:
if enabled == 'enabled':
data['enable'] = True
sql_enabled = f'AND `tbl`.enable = :enable'
elif enabled == 'disabled':
data['enable'] = False
sql_enabled = f'AND `tbl`.enable = :enable'
elif enabled == 'all':
sql_enabled = ''
# sql_enabled = ''
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql = f"""
SELECT `tbl`.id AS 'cont_edu_cert_id', `tbl`.id_random AS 'cont_edu_cert_id_random'
FROM `cont_edu_cert` AS `tbl`
WHERE
{sql_account_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if cont_edu_cert_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
cont_edu_cert_rec_li = cont_edu_cert_rec_li_result
else:
cont_edu_cert_rec_li = []
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(cont_edu_cert_rec_li_result)
return cont_edu_cert_rec_li
# ### END ### API Cont Edu Cert Methods ### get_cont_edu_cert_rec_list() ###

113
app/methods/cont_edu_cert_person_methods.py Normal file
View File

@@ -0,0 +1,113 @@
from __future__ import annotations
import datetime
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from app.db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
from app.lib_general import log, logging
from app.methods.cont_edu_cert_methods import load_cont_edu_cert_obj
from app.models.common_field_schema import default_num_bytes
from app.models.cont_edu_cert_person_models import Cont_Edu_Cert_Person_Base
# ### BEGIN ### API Cont Edu Cert Person Methods ### load_cont_edu_cert_person_obj() ###
def load_cont_edu_cert_person_obj(
cont_edu_cert_person_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_cont_edu_cert:bool=False
) -> Cont_Edu_Cert_Person_Base|dict|bool:
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if cont_edu_cert_person_id := redis_lookup_id_random(record_id_random=cont_edu_cert_person_id, table_name='cont_edu_cert_person'): pass
else: return False
if cont_edu_cert_person_rec := sql_select(table_name='v_cont_edu_cert_person', record_id=cont_edu_cert_person_id): pass
else: return False
try:
cont_edu_cert_person_obj = Cont_Edu_Cert_Person_Base(**cont_edu_cert_person_rec)
log.debug(cont_edu_cert_person_obj)
except ValidationError as e:
log.error(e.json())
return False
if inc_cont_edu_cert:
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
cont_edu_cert_id = cont_edu_cert_person_rec.get('cont_edu_cert_id', None)
log.debug(cont_edu_cert_id)
if cont_edu_cert_result := load_cont_edu_cert_obj(
cont_edu_cert_id = cont_edu_cert_id,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
):
cont_edu_cert_person_obj.cont_edu_cert = cont_edu_cert_result
else: cont_edu_cert_person_obj.cont_edu_cert = None
if model_as_dict:
return cont_edu_cert_person_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
else:
return cont_edu_cert_person_obj
# ### END ### API Cont Edu Cert Person Methods ### load_cont_edu_cert_person_obj() ###
# ### BEGIN ### API Cont Edu Cert Person Methods ### get_cont_edu_cert_person_rec_list() ###
def get_cont_edu_cert_person_rec_list(
cont_edu_cert_id: str,
limit: int = 1000,
enabled: str = 'enabled', # enabled, disabled, all
) -> list|bool:
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if cont_edu_cert_id := redis_lookup_id_random(record_id_random=cont_edu_cert_id, table_name='cont_edu_cert'): pass
else: return False
data = {}
data['cont_edu_cert_id'] = cont_edu_cert_id
sql_cont_edu_cert_id = f'`tbl`.cont_edu_cert_id = :cont_edu_cert_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 'cont_edu_cert_person_id', `tbl`.id_random AS 'cont_edu_cert_person_id_random'
FROM `cont_edu_cert_person` AS `tbl`
WHERE
{sql_cont_edu_cert_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if cont_edu_cert_person_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
cont_edu_cert_person_rec_li = cont_edu_cert_person_rec_li_result
else:
cont_edu_cert_person_rec_li = []
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(cont_edu_cert_person_rec_li_result)
return cont_edu_cert_person_rec_li
# ### END ### API Cont Edu Cert Person Methods ### get_cont_edu_cert_person_rec_list() ###

2
app/methods/contact_methods.py
View File

@@ -22,7 +22,7 @@ def load_contact_obj(
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_address:bool=False
) -> Contact_Base|bool:
) -> Contact_Base|dict|bool:
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())

2
app/methods/person_methods.py
View File

@@ -52,7 +52,7 @@ def load_person_obj(
inc_post_comment_list: bool = False,
inc_product: bool = False,
inc_user: bool = False,
) -> Person_Base|bool:
) -> Person_Base|dict|bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())

2
app/methods/user_methods.py
View File

@@ -98,7 +98,7 @@ def load_user_obj(
inc_post_list: bool = False,
inc_post_comment_list: bool = False,
inc_user_role_list: bool = False,
) -> User_Out_Base|bool:
) -> User_Out_Base|dict|bool:
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())

2
app/models/common_field_schema.py
View File

@@ -27,6 +27,8 @@ base_fields['account_cfg_id_random'] = xxx_id_random_field_schema
base_fields['address_id_random'] = xxx_id_random_field_schema
base_fields['archive_id_random'] = xxx_id_random_field_schema
base_fields['contact_id_random'] = xxx_id_random_field_schema
base_fields['cont_edu_cert_id_random'] = xxx_id_random_field_schema
base_fields['cont_edu_cert_person_id_random'] = xxx_id_random_field_schema
base_fields['event_exhibit_id_random'] = xxx_id_random_field_schema
base_fields['event_file_id_random'] = xxx_id_random_field_schema
base_fields['event_id_random'] = xxx_id_random_field_schema

82
app/models/cont_edu_cert_models.py Normal file
View File

@@ -0,0 +1,82 @@
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 log, logging
from app.models.common_field_schema import base_fields, default_num_bytes
from app.models.product_models import Product_Base
class Cont_Edu_Cert_Base(BaseModel):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['cont_edu_cert_id_random'],
alias = 'cont_edu_cert_id_random',
default_factory = lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
alias = 'cont_edu_cert_id'
)
account_id_random: Optional[str]
account_id: Optional[int]
product_id_random: Optional[str]
product_id: Optional[int]
name: Optional[str]
description: Optional[str]
title: Optional[str]
datetime_start: Optional[datetime.datetime]
datetime_end: Optional[datetime.datetime]
location: Optional[str]
enable: Optional[bool] = False
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]]
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
@validator('id', always=True)
def cont_edu_cert_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values.get('id_random', None): # 'id_random' in values and values['id_random']:
log.debug(values['id_random'])
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='cont_edu_cert')
return None
@validator('account_id', always=True)
def account_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['account_id_random']:
return redis_lookup_id_random(record_id_random=values['account_id_random'], table_name='account')
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): # 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
allow_population_by_field_name = True
fields = base_fields

117
app/models/cont_edu_cert_person_models.py Normal file
View File

@@ -0,0 +1,117 @@
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 log, logging
from app.models.common_field_schema import base_fields, default_num_bytes
from app.models.cont_edu_cert_models import Cont_Edu_Cert_Base
from app.models.person_models import Person_Base
from app.models.user_models import User_Base
class Cont_Edu_Cert_Person_Base(BaseModel):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['cont_edu_cert_person_id_random'],
alias = 'cont_edu_cert_person_id_random',
default_factory = lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
alias = 'cont_edu_cert_person_id'
)
account_id_random: Optional[str]
account_id: Optional[int]
cont_edu_cert_id_random: Optional[str]
cont_edu_cert_id: Optional[int]
person_id_random: Optional[str]
person_id: Optional[int]
user_id_random: Optional[str]
user_id: Optional[int]
external_id: Optional[str]
informal_name: Optional[str]
given_name: Optional[str]
family_name: Optional[str]
middle_name: Optional[str]
# prefix: Optional[str]
# suffix: Optional[str]
full_name: Optional[str]
display_name: Optional[str]
title: Optional[str]
organization_name: Optional[str]
enable: Optional[bool] = False
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
# Including other related objects
cont_edu_cert: Optional[Union[Cont_Edu_Cert_Base, None]]
person: Optional[Union[Person_Base, None]]
user: Optional[Union[User_Base, None]]
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
@validator('id', always=True)
def cont_edu_cert_person_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values.get('id_random', None): # 'id_random' in values and values['id_random']:
log.debug(values['id_random'])
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='cont_edu_cert_person')
return None
@validator('account_id', always=True)
def account_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['account_id_random']:
return redis_lookup_id_random(record_id_random=values['account_id_random'], table_name='account')
return None
@validator('cont_edu_cert_id', always=True)
def cont_edu_cert_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values.get('cont_edu_cert_id_random', None): # values['cont_edu_cert_id_random']:
return redis_lookup_id_random(record_id_random=values['cont_edu_cert_id_random'], table_name='cont_edu_cert')
return None
@validator('person_id', always=True)
def person_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['person_id_random']:
return redis_lookup_id_random(record_id_random=values['person_id_random'], table_name='person')
return None
@validator('user_id', always=True)
def user_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['user_id_random']:
return redis_lookup_id_random(record_id_random=values['user_id_random'], table_name='user')
return None
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True
fields = base_fields

253
app/routers/cont_edu_cert.py Normal file
View File

@@ -0,0 +1,253 @@
import datetime
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 app.lib_general import log, logging
from app.config import settings
from app.db_sql import sql_insert, sql_update, sql_insert_or_update, sql_select, sql_delete, redis_lookup_id_random
from app.routers.api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
from app.methods.cont_edu_cert_methods import get_cont_edu_cert_rec_list, load_cont_edu_cert_obj
from app.models.cont_edu_cert_models import Cont_Edu_Cert_Base
from app.models.response_models import *
router = APIRouter()
@router.post('/cont_edu/cert', response_model=Resp_Body_Base)
async def post_cont_edu_cert_obj(
obj: Cont_Edu_Cert_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.debug(locals())
obj_type = 'cont_edu_cert'
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('/cont_edu/cert/{obj_id}', response_model=Resp_Body_Base)
async def patch_cont_edu_cert_obj(
obj: Cont_Edu_Cert_Base,
obj_id: str = Query(..., min_length=1, max_length=22),
x_account_id: Optional[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())
obj_type = 'cont_edu_cert'
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
# ### BEGIN ### API Cont Edu Cert ### patch_cont_edu_cert_json() ###
@router.patch('/cont_edu/cert/{cont_edu_cert_id}/json', response_model=Resp_Body_Base)
async def patch_cont_edu_cert_json(
cont_edu_cert_obj: Cont_Edu_Cert_Base,
cont_edu_cert_id: str = Query(..., min_length=1, max_length=22),
create_missing_obj: bool = False,
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())
if cont_edu_cert_id := redis_lookup_id_random(record_id_random=cont_edu_cert_id, table_name='cont_edu_cert'): pass
else:
return mk_resp(data=None, status_code=404)
if cont_edu_cert_obj_up_result := update_cont_edu_cert_obj(
cont_edu_cert_id=cont_edu_cert_id,
cont_edu_cert_obj_up=cont_edu_cert_obj,
create_missing_obj=create_missing_obj,
):
log.debug(cont_edu_cert_obj_up_result)
if return_obj:
cont_edu_cert_obj = load_cont_edu_cert_obj(cont_edu_cert_id=cont_edu_cert_id)
cont_edu_cert_dict = cont_edu_cert_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset)
return mk_resp(data=cont_edu_cert_dict)
else:
return mk_resp(data=cont_edu_cert_obj_up_result)
else:
return mk_resp(data=False, status_code=400) # Bad Request
# ### END ### API Cont Edu Cert ### patch_cont_edu_cert_json() ###
@router.get('/cont_edu/cert/list', response_model=Resp_Body_Base)
async def get_cont_edu_cert_obj_li(
for_obj_type: str = Query(None, min_length=2, max_length=50),
for_obj_id: str = Query(None, min_length=1, max_length=22),
group: str = Query(None, min_length=2, max_length=50),
x_account_id: str = Header(...),
by_alias: bool = True,
exclude_unset: bool = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'cont_edu_cert'
if for_obj_type == 'event_exhibit' and for_obj_id:
#base_name = obj_type_li[obj_type]['base_name']
base_name = Cont_Edu_Cert_Base
for_obj_id_random = for_obj_id
for_obj_id = redis_lookup_id_random(record_id_random=for_obj_id_random, table_name=for_obj_type)
data = {}
data['for_obj_type'] = for_obj_type
data['for_obj_id'] = for_obj_id
data['for_obj_id_random'] = for_obj_id_random
data['group'] = group
if group:
sql = """
SELECT *
FROM `cont_edu_cert` AS cont_edu_cert
WHERE cont_edu_cert.for_type = :for_obj_type AND cont_edu_cert.for_id = :for_obj_id
AND cont_edu_cert.group = :group
"""
else:
sql = """
SELECT *
FROM `cont_edu_cert` AS cont_edu_cert
WHERE cont_edu_cert.for_type = :for_obj_type AND cont_edu_cert.for_id = :for_obj_id
"""
if sql_result := sql_select(data=data, sql=sql, as_list=True):
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)
else:
log.debug(sql_result)
return mk_resp(data=False, status_code=404)
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('/cont_edu/cert/{obj_id}', response_model=Resp_Body_Base)
async def get_cont_edu_cert_obj(
obj_id: str = Query(..., 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 = 'cont_edu_cert'
result = get_obj_template(
obj_type=obj_type,
obj_id=obj_id,
by_alias=True,
exclude_unset=True,
)
return result
# ### BEGIN ### API Cont Edu Cert ### get_account_obj_cont_edu_cert_list() ###
# Updated 2021-07-28
@router.get('/account/{account_id}/cont_edu/cert/list', response_model=Resp_Body_Base)
async def get_account_obj_cont_edu_cert_list(
account_id: str = Query(..., min_length=1, max_length=22),
limit: int = 500, # For now this covers any included objects or object lists
enabled: str = 'enabled', # For now this covers any included objects or object lists
inc_cont_edu_cert: bool = False,
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())
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return mk_resp(data=None, status_code=404)
response_data = None
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# Updated 2021-07-28
if cont_edu_cert_rec_list_result := get_cont_edu_cert_rec_list(
account_id = account_id,
limit = limit,
enabled = enabled,
):
cont_edu_cert_result_list = []
for cont_edu_cert_rec in cont_edu_cert_rec_list_result:
if load_cont_edu_cert_result := load_cont_edu_cert_obj(
cont_edu_cert_id = cont_edu_cert_rec.get('cont_edu_cert_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
enabled = enabled,
):
cont_edu_cert_result_list.append(load_cont_edu_cert_result)
else:
cont_edu_cert_result_list.append(None)
response_data = cont_edu_cert_result_list
else:
return mk_resp(data=False, status_code=400) # Bad Request
return mk_resp(data=response_data)
# ### END ### API ### get_account_obj_cont_edu_cert_list() ###
@router.delete('/cont_edu/cert/{obj_id}', response_model=Resp_Body_Base)
async def delete_cont_edu_cert_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 = 'cont_edu_cert'
result = delete_obj_template(
obj_type=obj_type,
obj_id=obj_id,
)
return result

254
app/routers/cont_edu_cert_person.py Normal file
View File

@@ -0,0 +1,254 @@
import datetime
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 app.lib_general import log, logging
from app.config import settings
from app.db_sql import sql_insert, sql_update, sql_insert_or_update, sql_select, sql_delete, redis_lookup_id_random
from app.routers.api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
from app.methods.cont_edu_cert_person_methods import get_cont_edu_cert_person_rec_list, load_cont_edu_cert_person_obj
from app.models.cont_edu_cert_person_models import Cont_Edu_Cert_Person_Base
from app.models.response_models import *
router = APIRouter()
@router.post('/cont_edu/cert/person', response_model=Resp_Body_Base)
async def post_cont_edu_cert_person_obj(
obj: Cont_Edu_Cert_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.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'cont_edu_cert_person'
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('/cont_edu/cert/person/{obj_id}', response_model=Resp_Body_Base)
async def patch_cont_edu_cert_person_obj(
obj: Cont_Edu_Cert_Person_Base,
obj_id: str = Query(..., min_length=1, max_length=22),
x_account_id: Optional[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())
obj_type = 'cont_edu_cert_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=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
# ### BEGIN ### API Cont Edu Cert Person ### patch_cont_edu_cert_person_json() ###
@router.patch('/cont_edu/cert/person/{cont_edu_cert_person_id}/json', response_model=Resp_Body_Base)
async def patch_cont_edu_cert_person_json(
cont_edu_cert_person_obj: Cont_Edu_Cert_Person_Base,
cont_edu_cert_person_id: str = Query(..., min_length=1, max_length=22),
create_missing_obj: bool = False,
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())
if cont_edu_cert_person_id := redis_lookup_id_random(record_id_random=cont_edu_cert_person_id, table_name='cont_edu_cert_person'): pass
else:
return mk_resp(data=None, status_code=404)
if cont_edu_cert_person_obj_up_result := update_cont_edu_cert_person_obj(
cont_edu_cert_person_id=cont_edu_cert_person_id,
cont_edu_cert_person_obj_up=cont_edu_cert_person_obj,
create_missing_obj=create_missing_obj,
):
log.debug(cont_edu_cert_person_obj_up_result)
if return_obj:
cont_edu_cert_person_obj = load_cont_edu_cert_person_obj(cont_edu_cert_person_id=cont_edu_cert_person_id)
cont_edu_cert_person_dict = cont_edu_cert_person_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset)
return mk_resp(data=cont_edu_cert_person_dict)
else:
return mk_resp(data=cont_edu_cert_person_obj_up_result)
else:
return mk_resp(data=False, status_code=400) # Bad Request
# ### END ### API Cont Edu Cert Person ### patch_cont_edu_cert_person_json() ###
@router.get('/cont_edu/cert/person/list', response_model=Resp_Body_Base)
async def get_cont_edu_cert_person_obj_li(
for_obj_type: str = Query(None, min_length=2, max_length=50),
for_obj_id: str = Query(None, min_length=1, max_length=22),
group: str = Query(None, min_length=2, max_length=50),
x_account_id: str = Header(...),
by_alias: bool = True,
exclude_unset: bool = True,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'cont_edu_cert_person'
if for_obj_type == 'event_exhibit' and for_obj_id:
#base_name = obj_type_li[obj_type]['base_name']
base_name = Cont_Edu_Cert_Person_Base
for_obj_id_random = for_obj_id
for_obj_id = redis_lookup_id_random(record_id_random=for_obj_id_random, table_name=for_obj_type)
data = {}
data['for_obj_type'] = for_obj_type
data['for_obj_id'] = for_obj_id
data['for_obj_id_random'] = for_obj_id_random
data['group'] = group
if group:
sql = """
SELECT *
FROM `cont_edu_cert_person` AS cont_edu_cert_person
WHERE cont_edu_cert_person.for_type = :for_obj_type AND cont_edu_cert_person.for_id = :for_obj_id
AND cont_edu_cert_person.group = :group
"""
else:
sql = """
SELECT *
FROM `cont_edu_cert_person` AS cont_edu_cert_person
WHERE cont_edu_cert_person.for_type = :for_obj_type AND cont_edu_cert_person.for_id = :for_obj_id
"""
if sql_result := sql_select(data=data, sql=sql, as_list=True):
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)
else:
log.debug(sql_result)
return mk_resp(data=False, status_code=404)
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('/cont_edu/cert/person/{obj_id}', response_model=Resp_Body_Base)
async def get_cont_edu_cert_person_obj(
obj_id: str = Query(..., 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 = 'cont_edu_cert_person'
result = get_obj_template(
obj_type=obj_type,
obj_id=obj_id,
by_alias=True,
exclude_unset=True,
)
return result
# ### BEGIN ### API Cont Edu Cert Person ### get_account_obj_cont_edu_cert_person_list() ###
# Updated 2021-07-28
@router.get('/account/{account_id}/cont_edu/cert/person/list', response_model=Resp_Body_Base)
async def get_account_obj_cont_edu_cert_person_list(
account_id: str = Query(..., min_length=1, max_length=22),
limit: int = 500, # For now this covers any included objects or object lists
enabled: str = 'enabled', # For now this covers any included objects or object lists
inc_cont_edu_cert: bool = False,
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())
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return mk_resp(data=None, status_code=404)
response_data = None
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# Updated 2021-07-28
if cont_edu_cert_person_rec_list_result := get_cont_edu_cert_person_rec_list(
account_id = account_id,
limit = limit,
enabled = enabled,
):
cont_edu_cert_person_result_list = []
for cont_edu_cert_person_rec in cont_edu_cert_person_rec_list_result:
if load_cont_edu_cert_person_result := load_cont_edu_cert_person_obj(
cont_edu_cert_person_id = cont_edu_cert_person_rec.get('cont_edu_cert_person_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
enabled = enabled,
inc_cont_edu_cert = inc_cont_edu_cert,
):
cont_edu_cert_person_result_list.append(load_cont_edu_cert_person_result)
else:
cont_edu_cert_person_result_list.append(None)
response_data = cont_edu_cert_person_result_list
else:
return mk_resp(data=False, status_code=400) # Bad Request
return mk_resp(data=response_data)
# ### END ### API Person ### get_account_obj_cont_edu_cert_person_list() ###
@router.delete('/cont_edu/cert/person/{obj_id}', response_model=Resp_Body_Base)
async def delete_cont_edu_cert_person_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 = 'cont_edu_cert_person'
result = delete_obj_template(
obj_type=obj_type,
obj_id=obj_id,
)
return result