Add region, extra_headers and extra_query_params

Now all APIs accepts region, extra_headers and extra_query_params
additional arguments for more control.

Signed-off-by: Bala.FA <[email protected]>

Author: balamurugana

minio/api.py

@@ -264,6 +264,71 @@ def check_status(status: str):
         raise ValueError("status must be 'Enabled' or 'Disabled'")
 
 
+@dataclass(frozen=True)
+class SourceObject:
+    """Source object for copy and compose object."""
+    bucket_name: str
+    object_name: str
+    version_id: Optional[str] = None
+    ssec: Optional[SseCustomerKey] = None
+    offset: int = 0
+    length: int = 0
+    match_etag: Optional[str] = None
+    not_match_etag: Optional[str] = None
+    modified_since: Optional[datetime] = None
+    unmodified_since: Optional[datetime] = None
+    fetch_checksum: bool = False
+    region: Optional[str] = None
+
+    def __post_init__(self):
+        if (
+                self.ssec is not None and
+                not isinstance(self.ssec, SseCustomerKey)
+        ):
+            raise ValueError("ssec must be SseCustomerKey type")
+        if self.offset < 0:
+            raise ValueError("offset should be zero or greater")
+        if self.length <= 0:
+            raise ValueError("length should be greater than zero")
+        if self.match_etag is not None and self.match_etag == "":
+            raise ValueError("match_etag must not be empty")
+        if self.not_match_etag is not None and self.not_match_etag == "":
+            raise ValueError("not_match_etag must not be empty")
+        if (
+                self.modified_since is not None and
+                not isinstance(self.modified_since, datetime)
+        ):
+            raise ValueError("modified_since must be datetime type")
+        if (
+                self.unmodified_since is not None and
+                not isinstance(self.unmodified_since, datetime)
+        ):
+            raise ValueError("unmodified_since must be datetime type")
+
+    def gen_copy_headers(self) -> dict[str, str]:
+        """Generate copy source headers."""
+        copy_source = quote("/" + self.bucket_name + "/" + self.object_name)
+        if self.version_id:
+            copy_source += "?versionId=" + quote(self.version_id)
+
+        headers = {"x-amz-copy-source": copy_source}
+        if self.ssec:
+            headers.update(self.ssec.copy_headers())
+        if self.match_etag:
+            headers["x-amz-copy-source-if-match"] = self.match_etag
+        if self.not_match_etag:
+            headers["x-amz-copy-source-if-none-match"] = self.not_match_etag
+        if self.modified_since:
+            headers["x-amz-copy-source-if-modified-since"] = (
+                to_http_header(self.modified_since)
+            )
+        if self.unmodified_since:
+            headers["x-amz-copy-source-if-unmodified-since"] = (
+                to_http_header(self.unmodified_since)
+            )
+        return headers
+
+
 @dataclass
 class ObjectConditionalReadArgs(ABC):
     """Base argument class holds condition properties for reading object."""

minio/checksum.py

@@ -35,6 +35,7 @@
 from xml.etree import ElementTree as ET
 
 import certifi
+from urllib3._collections import HTTPHeaderDict
 from urllib3.poolmanager import PoolManager
 
 try:
@@ -75,7 +76,7 @@ def _urlopen(
         method: str,
         url: str,
         body: Optional[str | bytes] = None,
-        headers: Optional[dict[str, str | list[str] | tuple[str]]] = None,
+        headers: Optional[HTTPHeaderDict] = None,
 ) -> BaseHTTPResponse:
     """Wrapper of urlopen() handles HTTP status code."""
     res = http_client.urlopen(method, url, body=body, headers=headers)
@@ -167,15 +168,16 @@ def retrieve(self) -> Credentials:
             return self._credentials
 
         utctime = utcnow()
+        headers = HTTPHeaderDict({
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Host": self._host,
+            "X-Amz-Date": to_amz_date(utctime),
+        })
         headers = sign_v4_sts(
             method="POST",
             url=self._url,
             region=self._region,
-            headers={
-                "Content-Type": "application/x-www-form-urlencoded",
-                "Host": self._host,
-                "X-Amz-Date": to_amz_date(utctime),
-            },
+            headers=headers,
             credentials=Credentials(
                 access_key=self._access_key,
                 secret_key=self._secret_key,
@@ -438,7 +440,7 @@ def __init__(
     def fetch(
         self,
         url: str,
-        headers: Optional[dict[str, str | list[str] | tuple[str]]] = None,
+        headers: Optional[HTTPHeaderDict] = None,
     ) -> Credentials:
         """Fetch credentials from EC2/ECS."""
         res = _urlopen(self._http_client, "GET", url, headers=headers)
@@ -484,11 +486,14 @@ def retrieve(self) -> Credentials:
             self._credentials = provider.retrieve()
             return cast(Credentials, self._credentials)
 
-        headers: Optional[dict[str, str | list[str] | tuple[str]]] = None
+        headers: Optional[HTTPHeaderDict] = None
         if self._relative_uri:
             if not url:
                 url = "http://169.254.170.2" + self._relative_uri
-            headers = {"Authorization": self._token} if self._token else None
+            headers = (
+                HTTPHeaderDict({"Authorization": self._token})
+                if self._token else None
+            )
         elif self._full_uri:
             token = self._token
             if self._token_file:
@@ -499,20 +504,28 @@ def retrieve(self) -> Credentials:
                 if not url:
                     url = self._full_uri
                     _check_loopback_host(url)
-            headers = {"Authorization": token} if token else None
+            headers = (
+                HTTPHeaderDict({"Authorization": token}) if token else None
+            )
         else:
             if not url:
                 url = "http://169.254.169.254"
 
             # Get IMDS Token
+            headers = HTTPHeaderDict(
+                {"X-aws-ec2-metadata-token-ttl-seconds": "21600"},
+            )
             res = _urlopen(
                 self._http_client,
                 "PUT",
                 url+"/latest/api/token",
-                headers={"X-aws-ec2-metadata-token-ttl-seconds": "21600"},
+                headers=headers,
             )
             token = res.data.decode("utf-8")
-            headers = {"X-aws-ec2-metadata-token": token} if token else None
+            headers = (
+                HTTPHeaderDict({"X-aws-ec2-metadata-token": token})
+                if token else None
+            )
 
             # Get role name
             url = urlunsplit(

minio/commonconfig.py

@@ -41,7 +41,7 @@
 
 from .commonconfig import Tags
 from .credentials import Credentials
-from .helpers import check_bucket_name
+from .helpers import HTTPQueryDict, check_bucket_name
 from .signer import get_credential_string, post_presign_v4
 from .time import from_iso8601utc, to_amz_date, to_iso8601utc
 from .xml import find, findall, findtext
@@ -67,10 +67,19 @@ class Bucket:
 class ListAllMyBucketsResult:
     """LissBuckets API result."""
     buckets: list[Bucket]
+    prefix: Optional[str]
+    continuation_token: Optional[str]
+    owner_id: Optional[str] = None
+    owner_name: Optional[str] = None
 
     @classmethod
     def fromxml(cls: Type[A], element: ET.Element) -> A:
         """Create new object with values from XML element."""
+        prefix = findtext(element, "Prefix")
+        continuation_token = findtext(element, "ContinuationToken")
+        owner = find(element, "Owner")
+        owner_id = None if owner is None else findtext(owner, "ID")
+        owner_name = None if owner is None else findtext(owner, "DisplayName")
         element = cast(ET.Element, find(element, "Buckets", True))
         buckets = []
         elements = findall(element, "Bucket")
@@ -81,7 +90,13 @@ def fromxml(cls: Type[A], element: ET.Element) -> A:
                 name,
                 from_iso8601utc(creation_date) if creation_date else None,
             ))
-        return cls(buckets)
+        return cls(
+            buckets=buckets,
+            prefix=prefix,
+            continuation_token=continuation_token,
+            owner_id=owner_id,
+            owner_name=owner_name,
+        )
 
 
 B = TypeVar("B", bound="Object")
@@ -232,15 +247,15 @@ def parse_list_objects(
 class CompleteMultipartUploadResult:
     """CompleteMultipartUpload API result."""
 
-    http_headers: HTTPHeaderDict
+    headers: HTTPHeaderDict
     bucket_name: Optional[str] = None
     object_name: Optional[str] = None
     location: Optional[str] = None
     etag: Optional[str] = None
     version_id: Optional[str] = None
 
     def __init__(self, response: BaseHTTPResponse):
-        object.__setattr__(self, "http_headers", response.headers)
+        object.__setattr__(self, "headers", response.headers)
         element = ET.fromstring(response.data.decode())
         object.__setattr__(self, "bucket_name", findtext(element, "Bucket"))
         object.__setattr__(self, "object_name", findtext(element, "Key"))
@@ -751,16 +766,15 @@ class SiteReplicationStatusOptions:
     entity: Optional[str] = None
     entity_value: Optional[str] = None
 
-    def to_query_params(self) -> dict[str, str]:
+    def to_query_params(self) -> HTTPQueryDict:
         """Convert this options to query parameters."""
-        params = {
-            "buckets": str(self.buckets).lower(),
-            "policies": str(self.policies).lower(),
-            "users": str(self.users).lower(),
-            "groups": str(self.groups).lower(),
-            "metrics": str(self.metrics).lower(),
-            "showDeleted": str(self.show_deleted).lower(),
-        }
+        params = HTTPQueryDict()
+        params["buckets"] = str(self.buckets).lower()
+        params["policies"] = str(self.policies).lower()
+        params["users"] = str(self.users).lower()
+        params["groups"] = str(self.groups).lower()
+        params["metrics"] = str(self.metrics).lower()
+        params["showDeleted"] = str(self.show_deleted).lower()
         if self.entity and self.entity_value:
             params["entity"] = self.entity
             params["entityvalue"] = self.entity_value