add missing fields

Signed-off-by: Bala.FA <bala@minio.io>

Author: balamurugana

minio/minio.py

@@ -66,12 +66,13 @@
                      CompleteMultipartUploadResult, CopyObjectResult,
                      CreateBucketConfiguration, CreateMultipartUploadResponse,
                      DeleteObjectsResponse, DeleteRequest, DeleteResult,
-                     EventIterable, GenericResponse, HeadObjectResponse,
-                     LegalHold, LifecycleConfig, ListAllMyBucketsResult,
-                     ListBucketsResponse, ListMultipartUploadsResponse,
-                     ListPartsResponse, NotificationConfig, Object,
-                     ObjectLockConfig, ObjectWriteResponse, Part, PostPolicy,
-                     ReplicationConfig, Retention, SelectObjectContentRequest,
+                     EventIterable, GenericResponse, HeadBucketResponse,
+                     HeadObjectResponse, LegalHold, LifecycleConfig,
+                     ListAllMyBucketsResult, ListBucketsResponse,
+                     ListMultipartUploadsResponse, ListPartsResponse,
+                     NotificationConfig, Object, ObjectLockConfig,
+                     ObjectWriteResponse, Part, PostPolicy, ReplicationConfig,
+                     Retention, SelectObjectContentRequest,
                      SelectObjectResponse, SSEConfig, StatObjectResponse,
                      Tagging, Tags, UploadPartCopyResponse, UploadPartResponse,
                      VersioningConfig, parse_list_objects)
@@ -1550,6 +1551,43 @@ def list_buckets(
             continuation_token = response.result.continuation_token
             yield from response.result.buckets
 
+    def head_bucket(
+            self,
+            *,
+            bucket_name: str,
+            region: Optional[str] = None,
+            extra_headers: Optional[HTTPHeaderDict] = None,
+            extra_query_params: Optional[HTTPQueryDict] = None,
+    ) -> Optional[HeadBucketResponse]:
+        """Execute HeadBucket API."""
+        check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
+        try:
+            response = self._execute(
+                method="HEAD",
+                bucket_name=bucket_name,
+                region=region,
+                extra_headers=extra_headers,
+                extra_query_params=extra_query_params,
+            )
+            bucket_region = response.headers.get("x-amz-bucket-region", "")
+            if (
+                    bucket_region and
+                    self._region_map.get(bucket_name) != bucket_region
+            ):
+                self._region_map.set(bucket_name, bucket_region)
+            return HeadBucketResponse(
+                headers=response.headers,
+                bucket_name=bucket_name,
+                region=(
+                    region or self._region_map.get(
+                        bucket_name) or bucket_region
+                ),
+            )
+        except S3Error as exc:
+            if exc.code != "NoSuchBucket":
+                raise
+        return None
+
     def bucket_exists(
             self,
             *,

minio/models.py

@@ -195,6 +195,10 @@ class Object:
     is_delete_marker: bool = False
     tags: Optional[Tags] = None
     is_dir: bool = field(default=False, init=False)
+    checksum_algorithms: Optional[List[Algorithm]] = None
+    checksum_type: Optional[ChecksumType] = None
+    is_restore_in_progress: bool = False
+    restore_expiry_date: Optional[datetime] = None
 
     def __post_init__(self):
         object.__setattr__(
@@ -248,6 +252,22 @@ def fromxml(
                 ),
             )
 
+        checksum_algorithms: List[Algorithm] = []
+        for elem in findall(element, "ChecksumAlgorithm"):
+            if elem.text:
+                checksum_algorithms.append(Algorithm[elem.text])
+
+        checksum_type = findtext(element, "ChecksumType")
+
+        is_restore_in_progress = cast(
+            str,
+            findtext(element, "RestoreStatus/IsRestoreInProgress", default=""),
+        )
+
+        restore_expiry_date = findtext(
+            element, "RestoreStatus/RestoreExpiryDate",
+        )
+
         return cls(
             bucket_name=bucket_name,
             object_name=object_name,
@@ -261,7 +281,16 @@ def fromxml(
             owner_name=owner_name,
             metadata=metadata,
             is_delete_marker=is_delete_marker,
-            tags=tags
+            tags=tags,
+            checksum_algorithms=checksum_algorithms or None,
+            checksum_type=(
+                ChecksumType[checksum_type] if checksum_type else None
+            ),
+            is_restore_in_progress=is_restore_in_progress.lower() == "true",
+            restore_expiry_date=(
+                from_iso8601utc(restore_expiry_date) if restore_expiry_date
+                else None
+            ),
         )
 
 
@@ -2519,6 +2548,7 @@ class Bucket:
         name: str
         creation_date: Optional[datetime] = None
         bucket_region: Optional[str] = None
+        bucket_arn: Optional[str] = None
 
         @classmethod
         def fromxml(
@@ -2533,6 +2563,7 @@ def fromxml(
                 creation_date=from_iso8601utc(
                     creation_date) if creation_date else None,
                 bucket_region=findtext(element, "BucketRegion"),
+                bucket_arn=findtext(element, "BucketArn"),
             )
 
 
@@ -2614,6 +2645,8 @@ class Upload:
         owner_name: Optional[str] = None
         storage_class: Optional[str] = None
         initiated_time: Optional[datetime] = None
+        checksum_algorithm: Optional[Algorithm] = None
+        checksum_type: Optional[ChecksumType] = None
 
         def __init__(
                 self, element: ET.Element, encoding_type: Optional[str] = None,
@@ -2661,6 +2694,20 @@ def __init__(
                 "initiated_time",
                 from_iso8601utc(initiated_time) if initiated_time else None,
             )
+            checksum_algorithm = findtext(element, "ChecksumAlgorithm")
+            if checksum_algorithm:
+                object.__setattr__(
+                    self,
+                    "checksum_algorithm",
+                    Algorithm[checksum_algorithm],
+                )
+            checksum_type = findtext(element, "ChecksumType")
+            if checksum_type:
+                object.__setattr__(
+                    self,
+                    "checksum_type",
+                    ChecksumType[checksum_type],
+                )
 
 
 @dataclass(frozen=True)
@@ -2946,6 +2993,48 @@ def __init__(  # pylint: disable=too-many-positional-arguments
         )
 
 
+@dataclass(frozen=True)
+class HeadBucketResponse(GenericResponse):
+    """ Response of HeadBucket API."""
+    bucket_arn: Optional[str] = None
+    location_type: Optional[str] = None
+    location_name: Optional[str] = None
+    access_point_alias: Optional[str] = None
+
+    def __init__(
+            self,
+            *,
+            headers: HTTPHeaderDict,
+            bucket_name: str,
+            region: str,
+    ):
+        super().__init__(
+            headers=headers,
+            bucket_name=bucket_name,
+            region=region,
+        )
+        object.__setattr__(
+            self,
+            "bucket_arn",
+            headers.get("x-amz-bucket-arn"),
+        )
+        object.__setattr__(
+            self,
+            "location_type",
+            headers.get("x-amz-bucket-location-type"),
+        )
+        object.__setattr__(
+            self,
+            "location_name",
+            headers.get("x-amz-bucket-location-name"),
+        )
+        object.__setattr__(
+            self,
+            "access_point_alias",
+            headers.get("x-amz-access-point-alias"),
+        )
+
+
 @dataclass(frozen=True)
 class HeadObjectResponse(GenericResponse):
     """ Response of HeadObject API."""