69622dbea6
- Reduced api_crud.py (1843 -> 143 lines) by extracting V1 registry and logic. - Reduced hosted_file.py (1596 -> 361 lines) by moving storage and media logic to methods. - Created lib_media.py for specialized video/image processing. - Created api_crud_methods.py for legacy template handlers. - Created legacy_v1.py for the legacy object registry. - Fixed subdirectory_path bug in Hosted File creation. - Verified full File Lifecycle via consolidated E2E suite.
111 lines
4.3 KiB
Python
111 lines
4.3 KiB
Python
from fastapi import Query, Response
|
|
from typing import Optional, Union, List
|
|
import logging
|
|
|
|
from app.db_sql import sql_insert, sql_update, sql_select, sql_delete, redis_lookup_id_random, lookup_id_random_pop
|
|
from app.models.response_models import mk_resp
|
|
from app.object_definitions.legacy_v1 import obj_type_li
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
def post_obj_template(
|
|
obj_type: str,
|
|
data: dict,
|
|
id_random_length: int = 8,
|
|
return_obj: bool = True,
|
|
by_alias: bool = True,
|
|
exclude_unset: Optional[bool] = True,
|
|
response: Response = Response,
|
|
**kwargs
|
|
):
|
|
obj_data = lookup_id_random_pop(data)
|
|
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=obj_type, data=obj_data, id_random_length=id_random_length):
|
|
obj_id = sql_insert_result
|
|
else:
|
|
return mk_resp(data=False, status_code=400, response=response)
|
|
|
|
if sql_select_result := sql_select(table_name=table_name_select, record_id=obj_id):
|
|
resp_data = base_name(**sql_select_result).dict(by_alias=by_alias, exclude_unset=exclude_unset)
|
|
return mk_resp(data=resp_data, response=response)
|
|
return mk_resp(data=False, status_code=404, response=response)
|
|
|
|
def patch_obj_template(
|
|
obj_type: str,
|
|
data: dict,
|
|
obj_id: str,
|
|
by_alias: bool=True,
|
|
exclude_unset: Optional[bool] = True,
|
|
response: Response = Response,
|
|
**kwargs
|
|
):
|
|
data['id_random'] = obj_id
|
|
obj_data = lookup_id_random_pop(data)
|
|
table_name_select = obj_type_li[obj_type]['table_name']
|
|
base_name = obj_type_li[obj_type]['base_name']
|
|
|
|
if sql_update(table_name=obj_type, data=obj_data):
|
|
obj_id_int = data['id']
|
|
else:
|
|
return mk_resp(data=False, status_code=400, response=response)
|
|
|
|
if sql_select_result := sql_select(table_name=table_name_select, record_id=obj_id_int):
|
|
resp_data = base_name(**sql_select_result).dict(by_alias=by_alias, exclude_unset=exclude_unset)
|
|
return mk_resp(data=resp_data, response=response)
|
|
return mk_resp(data=False, status_code=404, response=response)
|
|
|
|
def get_obj_li_template(
|
|
obj_type: str,
|
|
for_obj_type: Optional[str] = None,
|
|
for_obj_id: Optional[Union[int,str]] = None,
|
|
by_alias: Optional[bool] = True,
|
|
exclude_unset: Optional[bool] = True,
|
|
response: Response = Response,
|
|
**kwargs
|
|
):
|
|
if isinstance(for_obj_id, str):
|
|
for_obj_id = redis_lookup_id_random(record_id_random=for_obj_id, table_name=for_obj_type)
|
|
|
|
table_name_select = obj_type_li[obj_type]['table_name']
|
|
base_name = obj_type_li[obj_type]['base_name']
|
|
|
|
if for_obj_type and for_obj_id:
|
|
sql_result = sql_select(table_name=table_name_select, field_name=f'{for_obj_type}_id', field_value=for_obj_id)
|
|
else:
|
|
sql_result = sql_select(table_name=table_name_select)
|
|
|
|
resp_data_li = [base_name(**record).dict(by_alias=by_alias, exclude_unset=exclude_unset) for record in (sql_result or [])]
|
|
return mk_resp(data=resp_data_li, response=response)
|
|
|
|
def get_obj_template(
|
|
obj_id: Union[int,str],
|
|
obj_type: str,
|
|
by_alias: Optional[bool] = True,
|
|
exclude_unset: Optional[bool] = True,
|
|
response: Response = Response,
|
|
**kwargs
|
|
):
|
|
if isinstance(obj_id, str):
|
|
obj_id = redis_lookup_id_random(record_id_random=obj_id, table_name=obj_type)
|
|
|
|
table_name_select = obj_type_li[obj_type]['table_name']
|
|
if not obj_id: return mk_resp(data=False, status_code=404, response=response)
|
|
|
|
if sql_result := sql_select(table_name=table_name_select, record_id=obj_id):
|
|
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, response=response)
|
|
return mk_resp(data=False, status_code=404, response=response)
|
|
|
|
def delete_obj_template(
|
|
obj_type: str,
|
|
obj_id: str,
|
|
response: Response = Response,
|
|
**kwargs
|
|
):
|
|
if sql_delete(table_name=obj_type, record_id_random=obj_id):
|
|
return mk_resp(data=True, response=response)
|
|
return mk_resp(data=False, status_code=404, response=response)
|