(session: SessionDep, trans, file)
| 85 | return StreamingResponse(result, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") |
| 86 | |
| 87 | async def batchUpload(session: SessionDep, trans, file) -> UploadResultDTO: |
| 88 | ALLOWED_EXTENSIONS = {"xlsx", "xls"} |
| 89 | if not file.filename.lower().endswith(tuple(ALLOWED_EXTENSIONS)): |
| 90 | raise HTTPException(400, "Only support .xlsx/.xls") |
| 91 | |
| 92 | # Support FastAPI UploadFile (async read) and file-like objects. |
| 93 | NA_VALUES = ['', 'NA', 'N/A', 'NULL'] |
| 94 | df = None |
| 95 | # If file provides an async read (UploadFile), read bytes first |
| 96 | if hasattr(file, 'read') and asyncio.iscoroutinefunction(getattr(file, 'read')): |
| 97 | content = await file.read() |
| 98 | df = pd.read_excel(io.BytesIO(content), sheet_name=0, na_values=NA_VALUES) |
| 99 | else: |
| 100 | # If it's a Starlette UploadFile-like with a .file attribute, use that |
| 101 | if hasattr(file, 'file'): |
| 102 | fobj = file.file |
| 103 | try: |
| 104 | fobj.seek(0) |
| 105 | except Exception: |
| 106 | pass |
| 107 | df = pd.read_excel(fobj, sheet_name=0, na_values=NA_VALUES) |
| 108 | else: |
| 109 | # fallback: assume a path or file-like object |
| 110 | try: |
| 111 | file.seek(0) |
| 112 | except Exception: |
| 113 | pass |
| 114 | df = pd.read_excel(file, sheet_name=0, na_values=NA_VALUES) |
| 115 | head_list = list(df.columns) |
| 116 | i18n_head_list = get_i18n_head_list() |
| 117 | if not validate_head(trans=trans, head_i18n_list=i18n_head_list, head_list=head_list): |
| 118 | raise HTTPException(400, "Excel header validation failed") |
| 119 | success_list = [] |
| 120 | error_list = [] |
| 121 | for row in df.itertuples(): |
| 122 | row_validator = validate_row(trans=trans, head_i18n_list=i18n_head_list, row=row) |
| 123 | if row_validator.success: |
| 124 | success_list.append(row_validator.dict_data) |
| 125 | else: |
| 126 | error_list.append(row_validator) |
| 127 | error_file_id = None |
| 128 | if error_list: |
| 129 | error_file_id = generate_error_file(error_list, head_list) |
| 130 | result = UploadResultDTO(successCount=len(success_list), errorCount=len(error_list), dataKey=error_file_id) |
| 131 | if success_list: |
| 132 | user_po_list = [UserModel.model_validate(row) for row in success_list] |
| 133 | session.add_all(user_po_list) |
| 134 | session.commit() |
| 135 | return result |
| 136 | |
| 137 | def get_i18n_head_list(): |
| 138 | return [ |
no test coverage detected