[Python] How can I implement Get Bucket Quota for Minio in Python

I have been pulling my hair out trying to implement an equivalent for this mc cli command in Python.

[me@kompir ~/src/mc]$ mc quota info localhost-reporting-api/bucket-99
Bucket `bucket-99` has hard quota of 954 MiB


It is not implemented in the Python minio API, presumably because of AWS S3 compatibility. I have looked at the go SDK source, and I know next to nothing about the language, but I find this in a file named madmin-go/quota-commands.go

51 │ // GetBucketQuota - get info on a user
52 │ func (adm *AdminClient) GetBucketQuota(ctx context.Context, bucket string) (q BucketQuota, err error) {
53 │ queryValues := url.Values{}
54 │ queryValues.Set("bucket", bucket)
55 │
56 │ reqData := requestData{
57 │ relPath: adminAPIPrefix + "/get-bucket-quota",
58 │ queryValues: queryValues,
59 │ }
60 │


What is adminAPIPrefix, and where do I get this value? There is a hint in utils.go

53 │ // Admin API version prefix
54 │ adminAPIPrefix = "/" + AdminAPIVersion


But I expect this to be available in the Python minio base class somewhere, especially since I don't see an easy way to pass it up to the API's self._execute() method.

From what I can tell queryValues is probably a dictionary with values to set query parameters, and can be passed eg to Python Requests as query parameters.

I am not re-implementing the API in python. I'm just sub-classing Pypi minio, and trying to add two methods. I have added an equivalent to mc du, which was trivial to implement.

Now I need to implement mc admin quota info

Below is my best attempt so far:

class MinioQuota(Minio):
def get_bucket_quota(self,
bucket_name: str,
version_id: str | None = None,
):

check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)

query_params = {"versionId": version_id} if version_id else {}
query_params["bucket"] = bucket_name
# query_params["prefix"] = "get-bucket-quota" # Many combinations tried
query_params["bucket-quota"] = "" # Many combinations tried

try:
response = self._execute(
"GET",
bucket_name,
query_params=cast(DictType, query_params),
)
return response.data
# tagging = unmarshal(Bucket, response.data.decode())
# return tagging.tags
except S3Error as exc:
if exc.code != "NoSuchTagSet":
raise
return None


This doesn't work, it just returns a ListBucketResult.

I did also try appending it to the URL, which resulted in a response assuming there is a bucket named "get-bucket-quota"

Any idea? I'm about ready to just sub-process and run mc admin quota info to get the quota!

Continue reading...