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

Added more documentation. Improved CRUD V3 endpoints and better responses.

Browse Source
This commit is contained in:
Scott Idem
2026-01-06 13:52:05 -05:00
parent 9c06b07665
commit a42f32acf4
20 changed files with 1768 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
app/models/core_object_models.py
View File

@@ -68,6 +68,10 @@ class Core_Std_Obj_Base(BaseModel):
# return redis_lookup_id_random(record_id_random=id_random, table_name=otype)
return None
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True
class Core_Object_Base(BaseModel):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
@@ -148,6 +152,10 @@ class Core_Object_Base(BaseModel):
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True
class Example_Object_Base(Core_Object_Base): # Based on Core_Object_Base
title: Optional[str] = None

1
app/models/product_models.py
View File

@@ -103,4 +103,5 @@ class Product_Base(BaseModel):
class Config:
underscore_attrs_are_private = True
fields = base_fields
allow_population_by_field_name = True
# ### END ### API Product Models ### Product_Base() ###

2
app/models/site_domain_models.py
View File

@@ -70,6 +70,7 @@ class Site_Domain_Base(BaseModel):
class Config:
underscore_attrs_are_private = True
fields = base_fields
allow_population_by_field_name = True
# ### END ### API Site Domain Models ### Site_Domain_Base() ###
@@ -170,4 +171,5 @@ class Site_Domain_FQDN_ID_Base(BaseModel):
class Config:
underscore_attrs_are_private = True
fields = base_fields
allow_population_by_field_name = True
# ### END ### API Site Domain Models ### Site_Domain_FQDN_ID_Base() ###

65
app/routers/api_crud_v3.py
View File

@@ -246,21 +246,12 @@ async def get_obj_li(
if sql_result is False:
return mk_resp(data=False, status_code=500, response=response, status_message="Database error occurred.")
elif sql_result:
resp_data_li = []
for record in sql_result:
obj_inst = base_name(**record)
obj_dict = obj_inst.dict(
by_alias=serialization.by_alias,
exclude_unset=serialization.exclude_unset,
exclude_defaults=serialization.exclude_defaults,
exclude_none=serialization.exclude_none
)
# Ensure id_random is present if it exists on the instance
id_random_alias = base_name.__fields__['id_random'].alias if 'id_random' in base_name.__fields__ else 'id_random'
if id_random_alias not in obj_dict and hasattr(obj_inst, 'id_random'):
obj_dict[id_random_alias] = obj_inst.id_random
resp_data_li.append(obj_dict)
resp_data_li = [base_name(**record).dict(
by_alias=serialization.by_alias,
exclude_unset=serialization.exclude_unset,
exclude_defaults=serialization.exclude_defaults,
exclude_none=serialization.exclude_none
) for record in sql_result]
return mk_resp(data=resp_data_li, response=response)
else:
return mk_resp(data=[], status_code=200, response=response)
@@ -345,21 +336,12 @@ async def search_obj_li(
if sql_result is False:
return mk_resp(data=False, status_code=500, response=response, status_message="Database error occurred.")
elif sql_result:
resp_data_li = []
for record in sql_result:
obj_inst = base_name(**record)
obj_dict = obj_inst.dict(
by_alias=serialization.by_alias,
exclude_unset=serialization.exclude_unset,
exclude_defaults=serialization.exclude_defaults,
exclude_none=serialization.exclude_none
)
# Ensure id_random is present if it exists on the instance
id_random_alias = base_name.__fields__['id_random'].alias if 'id_random' in base_name.__fields__ else 'id_random'
if id_random_alias not in obj_dict and hasattr(obj_inst, 'id_random'):
obj_dict[id_random_alias] = obj_inst.id_random
resp_data_li.append(obj_dict)
resp_data_li = [base_name(**record).dict(
by_alias=serialization.by_alias,
exclude_unset=serialization.exclude_unset,
exclude_defaults=serialization.exclude_defaults,
exclude_none=serialization.exclude_none
) for record in sql_result]
return mk_resp(data=resp_data_li, response=response)
else:
return mk_resp(data=[], status_code=200, response=response)
@@ -608,21 +590,12 @@ async def get_child_obj_li(
if sql_result is False:
return mk_resp(data=False, status_code=500, response=response, status_message="Database error occurred.")
elif sql_result:
resp_data_li = []
for record in sql_result:
obj_inst = base_name(**record)
obj_dict = obj_inst.dict(
by_alias=serialization.by_alias,
exclude_unset=serialization.exclude_unset,
exclude_defaults=serialization.exclude_defaults,
exclude_none=serialization.exclude_none
)
# Ensure id_random is present if it exists on the instance
id_random_alias = base_name.__fields__['id_random'].alias if 'id_random' in base_name.__fields__ else 'id_random'
if id_random_alias not in obj_dict and hasattr(obj_inst, 'id_random'):
obj_dict[id_random_alias] = obj_inst.id_random
resp_data_li.append(obj_dict)
resp_data_li = [base_name(**record).dict(
by_alias=serialization.by_alias,
exclude_unset=serialization.exclude_unset,
exclude_defaults=serialization.exclude_defaults,
exclude_none=serialization.exclude_none
) for record in sql_result]
return mk_resp(data=resp_data_li, response=response)
else:
return mk_resp(data=[], status_code=200, response=response)
@@ -819,4 +792,4 @@ async def delete_child_obj(
if success:
return mk_resp(data=True, response=response, status_message=f"Deleted successfully via {method}.")
return mk_resp(data=False, status_code=400, response=response, status_message="Deletion failed.")
return mk_resp(data=False, status_code=400, response=response, status_message="Deletion failed.")