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

More robust and clean up

Browse Source
This commit is contained in:
Scott Idem
2021-06-14 16:23:12 -04:00
parent 2eda94baba
commit 7080b51dfc
1 changed files with 36 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
app/routers/hosted_file.py
View File

@@ -18,21 +18,25 @@ from app.models.response_models import mk_resp
router = APIRouter() router = APIRouter()
# ### BEGIN ### API Hosted File Route ### upload_files() ###
# This just needs to return the currect model for a hosted_file # This just needs to return the currect model for a hosted_file
# Everything else seems to be working well # Everything else seems to be working well
# Should this also do something with meta data and updating the DB? # Should this also do something with meta data and updating the DB?
@router.post('/upload_files/') @router.post('/upload_files/')
async def create_upload_files( async def upload_files(
file_list: List[UploadFile] = File(...), file_list: List[UploadFile] = File(...),
account_id: str = Form(..., min_length=1, max_length=22), account_id: str = Form(..., min_length=1, max_length=22),
# filename: Optional[str] = Form(...), # filename: Optional[str] = Form(...),
for_object_type: str = Form(...), for_object_type: str = Form(...),
for_object_id: str = Form(..., min_length=1, max_length=22), for_object_id: str = Form(..., min_length=1, max_length=22),
x_account_id: Optional[str] = Header(..., ), check_allowed_extension: bool = False,
return_obj: Optional[bool] = True, x_account_id: str = Header(..., ),
by_alias: Optional[bool] = True, return_obj: bool = True,
exclude_unset: Optional[bool] = True, by_alias: bool = True,
exclude_unset: bool = True,
): ):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
account_id_random = account_id # This is for the account random str ID account_id_random = account_id # This is for the account random str ID
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
@@ -49,13 +53,15 @@ async def create_upload_files(
file_info_list = [] file_info_list = []
for file_obj in file_list: for file_obj in file_list:
file_info = await save_file(file=file_obj, account_id=account_id, account_id_random=account_id_random, for_object_type=for_object_type, for_object_id=for_object_id, for_object_id_random=for_object_id_random) file_info = await save_file(file=file_obj, account_id=account_id, account_id_random=account_id_random, for_object_type=for_object_type, for_object_id=for_object_id, for_object_id_random=for_object_id_random, check_allowed_extension=check_allowed_extension)
file_info_list.append(file_info) file_info_list.append(file_info)
log.debug(file_info_list) log.debug(file_info_list)
return mk_resp(data=file_info_list) return mk_resp(data=file_info_list)
# ### END ### API Hosted File Route ### upload_files() ###
# ### BEGIN ### API Hosted File Route ### save_file() ###
async def save_file( async def save_file(
file: UploadFile, file: UploadFile,
account_id: int, account_id: int,
@@ -63,6 +69,7 @@ async def save_file(
for_object_type: str, for_object_type: str,
for_object_id: int, for_object_id: int,
for_object_id_random: str, for_object_id_random: str,
check_allowed_extension: bool = False,
): ):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals()) log.debug(locals())
@@ -75,12 +82,23 @@ async def save_file(
log.debug(f'{file.filename}') log.debug(f'{file.filename}')
file_info = {} file_info = {}
file_info['saved'] = None
file_info['for_object_type'] = for_object_type file_info['for_object_type'] = for_object_type
file_info['for_object_id'] = for_object_id file_info['for_object_id'] = for_object_id
file_info['for_object_id_random'] = for_object_id_random file_info['for_object_id_random'] = for_object_id_random
file_info['filename'] = file.filename file_info['filename'] = file.filename
file_info['extension'] = guess_file_extension(filename=file.filename) file_info['extension'] = guess_file_extension(filename=file.filename)
if check_allowed_extension:
if allowed_file_extension:
file_info['extension_allowed'] = True
else:
file_info['extension_allowed'] = False
file_info['saved'] = False
return file_info
else:
file_info['extension_allowed'] = None
# There is a difference between Content-Type and MIME type. # There is a difference between Content-Type and MIME type.
# https://stackoverflow.com/questions/3452381/whats-the-difference-of-contenttype-and-mimetype # https://stackoverflow.com/questions/3452381/whats-the-difference-of-contenttype-and-mimetype
file_info['content_type'] = file.content_type # might also include charset or other parameters file_info['content_type'] = file.content_type # might also include charset or other parameters
@@ -122,11 +140,15 @@ async def save_file(
log.debug(f'Elapsed time: {elapsed_time}') log.debug(f'Elapsed time: {elapsed_time}')
file_info['copy_timer'] = elapsed_time file_info['copy_timer'] = elapsed_time
file_info['saved'] = True
log.debug(shutil.disk_usage(hosted_file_path)) log.debug(shutil.disk_usage(hosted_file_path))
return file_info return file_info
# ### END ### API Hosted File Route ### save_file() ###
# ### BEGIN ### API Hosted File Route ### get_file_object_hash() ###
async def get_file_object_hash(file_object:File): async def get_file_object_hash(file_object:File):
#log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL #log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals()) log.debug(locals())
@@ -146,17 +168,16 @@ async def get_file_object_hash(file_object:File):
log.debug(f'Elapsed time: {elapsed_time}') log.debug(f'Elapsed time: {elapsed_time}')
return file_hash return file_hash
# ### END ### API Hosted File Route ### get_file_object_hash() ###
# def allowed_file_extension(filename): # ### BEGIN ### API Hosted File Route ### guess_file_extension() ###
# return False
# return '.' in filename and filename.rsplit('.', 1)[1].lower() in app.config['ALLOWED_EXTENSIONS']
def guess_file_extension(filename:str): def guess_file_extension(filename:str):
return filename.rsplit('.', 1)[1].lower() return filename.rsplit('.', 1)[1].lower()
# ### END ### API Hosted File Route ### guess_file_extension() ###
# def copyLargeFile(src, dest, buffer_size=16000):
# with open(src, 'rb') as fsrc: # ### BEGIN ### API Hosted File Route ### allowed_file_extension() ###
# with open(dest, 'wb') as fdest: def allowed_file_extension(extension:str):
# shutil.copyfileobj(fsrc, fdest, buffer_size) return extension.lower() in app.config['ALLOWED_EXTENSIONS']
# ### END ### API Hosted File Route ### allowed_file_extension() ###