Basic MinIO user management API

minio/error.py

@@ -53,6 +53,22 @@ def __reduce__(self):
         return type(self), (self._code, self._content_type, self._body)
 
 
+class AdminResponseError(Exception):
+    """Raised to indicate error response from server."""
+
+    def __init__(self, code, content_type, body):
+        self._code = code
+        self._content_type = content_type
+        self._body = body
+        super().__init__(
+            f"Error admin response from server; "
+            f"Response code: {code}, Body: {body}"
+        )
+
+    def __reduce__(self):
+        return type(self), (self._code, self._content_type, self._body)
+
+
 class ServerError(MinioException):
     """Raised to indicate that S3 service returning HTTP server error."""
 

minio/helpers.py

@@ -650,6 +650,71 @@ def build(
         return url_replace(url, netloc=netloc, path=path)
 
 
+class AdminURL:
+    """URL of Minio Admin endpoint"""
+
+    def __init__(self, endpoint):
+        url = urllib.parse.urlsplit(endpoint)
+        host = url.hostname
+
+        if url.scheme.lower() not in ["http", "https"]:
+            raise ValueError("scheme in endpoint must be http or https")
+
+        url = url_replace(url, scheme=url.scheme.lower())
+
+        if url.path and url.path != "/":
+            raise ValueError("path in endpoint is not allowed")
+
+        url = url_replace(url, path="")
+
+        if url.query:
+            raise ValueError("query in endpoint is not allowed")
+
+        if url.fragment:
+            raise ValueError("fragment in endpoint is not allowed")
+
+        try:
+            url.port
+        except ValueError as exc:
+            raise ValueError("invalid port") from exc
+
+        if url.username:
+            raise ValueError("username in endpoint is not allowed")
+
+        if url.password:
+            raise ValueError("password in endpoint is not allowed")
+
+        if (
+                (url.scheme == "http" and url.port == 80) or
+                (url.scheme == "https" and url.port == 443)
+        ):
+            url = url_replace(url, netloc=host)
+
+        self._url = url
+
+    @property
+    def is_https(self):
+        """Check if scheme is HTTPS."""
+        return self._url.scheme == "https"
+
+    def build(
+            self, path, query_params=None
+    ):
+        """Build URL for given information."""
+        url = url_replace(self._url, path=path)
+
+        query = []
+        for key, values in sorted((query_params or {}).items()):
+            values = values if isinstance(values, (list, tuple)) else [values]
+            query += [
+                f"{queryencode(key)}={queryencode(value)}"
+                for value in sorted(values)
+            ]
+        url = url_replace(url, query="&".join(query))
+
+        return url
+
+
 class ObjectWriteResult:
     """Result class of any APIs doing object creation."""