refactor(files): decompose S3 connector imports
This commit is contained in:
@@ -236,19 +236,16 @@ def _read_smb_file(profile: ConnectorProfile, *, path: str, max_bytes: int) -> C
|
||||
)
|
||||
|
||||
|
||||
def _read_s3_file(profile: ConnectorProfile, *, library_id: str, path: str, max_bytes: int) -> ConnectorDownloadedFile:
|
||||
bucket = _s3_bucket(profile, library_id)
|
||||
if not bucket:
|
||||
raise ConnectorImportError("S3 import requires a bucket in library_id or profile metadata")
|
||||
key = _s3_object_key(profile, path)
|
||||
if not key:
|
||||
raise ConnectorImportError("S3 import requires an object key")
|
||||
def _s3_import_client(profile: ConnectorProfile) -> Any:
|
||||
try:
|
||||
client = _s3_client(profile)
|
||||
return _s3_client(profile)
|
||||
except ConnectorBrowseUnsupported as exc:
|
||||
raise ConnectorImportUnsupported(str(exc)) from exc
|
||||
except ConnectorBrowseError as exc:
|
||||
raise ConnectorImportError(str(exc)) from exc
|
||||
|
||||
|
||||
def _s3_object_detail(client: Any, *, bucket: str, key: str, max_bytes: int) -> dict[str, Any]:
|
||||
try:
|
||||
detail = client.head_object(Bucket=bucket, Key=key)
|
||||
except Exception as exc: # pragma: no cover - concrete exception types are dependency-version specific
|
||||
@@ -258,7 +255,17 @@ def _read_s3_file(profile: ConnectorProfile, *, library_id: str, path: str, max_
|
||||
size = _int(detail.get("ContentLength"))
|
||||
if size is not None and size > max_bytes:
|
||||
raise ConnectorImportError(f"S3 object exceeds limit of {max_bytes} bytes")
|
||||
version_id = _clean(detail.get("VersionId"))
|
||||
return detail
|
||||
|
||||
|
||||
def _download_s3_object(
|
||||
client: Any,
|
||||
*,
|
||||
bucket: str,
|
||||
key: str,
|
||||
version_id: str | None,
|
||||
max_bytes: int,
|
||||
) -> tuple[Any, bytes]:
|
||||
request: dict[str, object] = {"Bucket": bucket, "Key": key}
|
||||
if version_id:
|
||||
request["VersionId"] = version_id
|
||||
@@ -270,6 +277,52 @@ def _read_s3_file(profile: ConnectorProfile, *, library_id: str, path: str, max_
|
||||
raise ConnectorImportError(f"S3 object download failed: {exc}") from exc
|
||||
if len(data) > max_bytes:
|
||||
raise ConnectorImportError(f"S3 object exceeds limit of {max_bytes} bytes")
|
||||
return response, data
|
||||
|
||||
|
||||
def _s3_download_metadata(
|
||||
detail: dict[str, Any],
|
||||
*,
|
||||
bucket: str,
|
||||
key: str,
|
||||
version_id: str | None,
|
||||
etag: str | None,
|
||||
size: int,
|
||||
) -> dict[str, Any]:
|
||||
metadata: dict[str, Any] = {
|
||||
"bucket": bucket,
|
||||
"key": key,
|
||||
"version_id": version_id,
|
||||
"etag": etag,
|
||||
"size": size,
|
||||
}
|
||||
for source_key, target_key in (
|
||||
("ChecksumSHA256", "checksum_sha256"),
|
||||
("ChecksumCRC32", "checksum_crc32"),
|
||||
("StorageClass", "storage_class"),
|
||||
):
|
||||
if source_key in detail:
|
||||
metadata[target_key] = detail[source_key]
|
||||
return metadata
|
||||
|
||||
|
||||
def _read_s3_file(profile: ConnectorProfile, *, library_id: str, path: str, max_bytes: int) -> ConnectorDownloadedFile:
|
||||
bucket = _s3_bucket(profile, library_id)
|
||||
if not bucket:
|
||||
raise ConnectorImportError("S3 import requires a bucket in library_id or profile metadata")
|
||||
key = _s3_object_key(profile, path)
|
||||
if not key:
|
||||
raise ConnectorImportError("S3 import requires an object key")
|
||||
client = _s3_import_client(profile)
|
||||
detail = _s3_object_detail(client, bucket=bucket, key=key, max_bytes=max_bytes)
|
||||
version_id = _clean(detail.get("VersionId"))
|
||||
response, data = _download_s3_object(
|
||||
client,
|
||||
bucket=bucket,
|
||||
key=key,
|
||||
version_id=version_id,
|
||||
max_bytes=max_bytes,
|
||||
)
|
||||
content_type = _clean(response.get("ContentType") if isinstance(response, dict) else None) or _clean(detail.get("ContentType")) or mimetypes.guess_type(key)[0]
|
||||
etag = _clean(response.get("ETag") if isinstance(response, dict) else None) or _clean(detail.get("ETag"))
|
||||
filename = filename_from_path(key)
|
||||
@@ -280,16 +333,14 @@ def _read_s3_file(profile: ConnectorProfile, *, library_id: str, path: str, max_
|
||||
revision=version_id or etag or _clean(detail.get("LastModified")),
|
||||
external_id=f"{bucket}:{key}",
|
||||
external_url=f"s3://{bucket}/{key}",
|
||||
metadata={
|
||||
"bucket": bucket,
|
||||
"key": key,
|
||||
"version_id": version_id,
|
||||
"etag": etag,
|
||||
"size": len(data),
|
||||
**({"checksum_sha256": detail["ChecksumSHA256"]} if "ChecksumSHA256" in detail else {}),
|
||||
**({"checksum_crc32": detail["ChecksumCRC32"]} if "ChecksumCRC32" in detail else {}),
|
||||
**({"storage_class": detail["StorageClass"]} if "StorageClass" in detail else {}),
|
||||
},
|
||||
metadata=_s3_download_metadata(
|
||||
detail,
|
||||
bucket=bucket,
|
||||
key=key,
|
||||
version_id=version_id,
|
||||
etag=etag,
|
||||
size=len(data),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user