| from pathlib import Path | |
| from fastapi import HTTPException, UploadFile, status | |
| ALLOWED_DATASET_EXTENSIONS = {".csv", ".xlsx", ".xls"} | |
| def get_file_extension(filename: str) -> str: | |
| return Path(filename).suffix.lower() | |
| def validate_supported_dataset_file(file: UploadFile) -> str: | |
| if not file.filename: | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail="Uploaded file must include a filename.", | |
| ) | |
| extension = get_file_extension(file.filename) | |
| if extension not in ALLOWED_DATASET_EXTENSIONS: | |
| allowed = ", ".join(sorted(ALLOWED_DATASET_EXTENSIONS)) | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail=f"Unsupported file type. Allowed file types: {allowed}.", | |
| ) | |
| return extension | |
| def validate_file_size(file_size_bytes: int, max_size_bytes: int) -> None: | |
| if file_size_bytes <= 0: | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail="Uploaded file is empty.", | |
| ) | |
| if file_size_bytes > max_size_bytes: | |
| max_size_display = ( | |
| f"{max_size_bytes / (1024 * 1024):.0f} MB" | |
| if max_size_bytes >= 1024 * 1024 | |
| else f"{max_size_bytes} bytes" | |
| ) | |
| raise HTTPException( | |
| status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, | |
| detail=f"Uploaded file is too large. Maximum size is {max_size_display}.", | |
| ) | |