(self, token: JWTPayload = Depends(get_jwt_token))
| 78 | client: boto3.client |
| 79 | |
| 80 | async def __call__(self, token: JWTPayload = Depends(get_jwt_token)) -> ObjectUploadResponse: |
| 81 | assert self.bucket_name is not None, "`bucket_name` must be provided" |
| 82 | |
| 83 | self.token = token |
| 84 | self.client = get_s3_client() |
| 85 | |
| 86 | body = BytesIO() |
| 87 | total_size = 0 |
| 88 | |
| 89 | # read the body in chunks so we don't ever load an entire oversized file into memory |
| 90 | async for chunk in self.request.stream(): |
| 91 | total_size += len(chunk) |
| 92 | |
| 93 | if total_size > self.max_size: |
| 94 | logger.error("Uploaded file exceeds maximum size limit") |
| 95 | raise HTTPException( |
| 96 | status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, |
| 97 | detail=f"File size exceeds the maximum limit of {self.max_size} bytes", |
| 98 | ) |
| 99 | |
| 100 | body.write(chunk) |
| 101 | |
| 102 | body.seek(0) |
| 103 | await self.upload_body(body) |
| 104 | return ObjectUploadResponse( |
| 105 | url=self.public_url, |
| 106 | size=total_size, |
| 107 | ) |
| 108 | |
| 109 | async def upload_body(self, body: BytesIO) -> None: |
| 110 | """Upload the object to S3.""" |
nothing calls this directly
no test coverage detected