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."""