feat(s3-backend): add error handling to object retrieving

This commit is contained in:
2026-08-04 00:56:41 +02:00
parent 099878ba57
commit e758522d80
2 changed files with 64 additions and 0 deletions
+8
View File
@@ -13,6 +13,7 @@
# limitations under the License. # limitations under the License.
import httpx import httpx
from .exceptions import S3InternalServerError, S3ObjectNotFoundError, S3ObjectPermissionError
from typing import AsyncGenerator, Optional from typing import AsyncGenerator, Optional
from urllib.parse import urljoin from urllib.parse import urljoin
@@ -58,5 +59,12 @@ class S3Backend:
urljoin(self.url, object_path), urljoin(self.url, object_path),
headers = headers headers = headers
) as response: ) as response:
if response.status_code == 404:
raise S3ObjectNotFoundError('Object not found', object_path = object_path)
if response.status_code == 403:
raise S3ObjectPermissionError('Object access forbidden', object_path = object_path)
if response.status_code // 100 == 5:
raise S3InternalServerError('Internal server error')
async for chunk in response.aiter_bytes(): async for chunk in response.aiter_bytes():
yield chunk yield chunk
+56
View File
@@ -0,0 +1,56 @@
# Copyright 2026 jCloud
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
__all__ = [
'S3Error',
'S3InternalServerError',
'S3ObjectError',
'S3ObjectNotFoundError',
'S3ObjectPermissionError'
]
class S3Error(Exception):
'''
Base class for all S3 related errors.
'''
class S3InternalServerError(S3Error):
'''
Internal S3 server error.
'''
class S3ObjectError(S3Error):
'''
Base class for S3 object related errors.
'''
def __init__(self, *args: object, object_path: str = '') -> None:
super().__init__(*args)
self.object_path = object_path
def __str__(self):
if self.args:
return f'{self.args[0]}{": " if self.object_path and self.args[0] else ""}{self.object_path if self.args[0] else ""}'
else:
return ''
class S3ObjectNotFoundError(S3ObjectError):
'''
S3 object was not found.
'''
class S3ObjectPermissionError(S3ObjectError):
'''
Bad permissions to retrieve S3 object.
'''