8000 minio-py/minio/datatypes.py at master · Novapixel1010/minio-py · GitHub
[go: up one dir, main page]
Skip to content

datatypes.py

Latest commit

 

History

History
1165 lines (980 loc) · 35.3 KB

datatypes.py

File metadata and controls

1165 lines (980 loc) · 35.3 KB
Raw
# -*- coding: utf-8 -*-
# MinIO Python Library for Amazon S3 Compatible Cloud Storage, (C)
# 2020 MinIO, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# pylint: disable=too-many-lines
"""
Response of ListBuckets, ListObjects, ListObjectsV2 and ListObjectVersions API.
"""
from __future__ import absolute_import, annotations
import base64
import json
from collections import OrderedDict
from datetime import datetime
from enum import Enum
from typing import Any, List, Tuple, Type, TypeVar, cast
from urllib.parse import unquote_plus
from xml.etree import ElementTree as ET
from urllib3._collections import HTTPHeaderDict
try:
from urllib3.response import BaseHTTPResponse # type: ignore[attr-defined]
except ImportError:
from urllib3.response import HTTPResponse as BaseHTTPResponse
from .commonconfig import Tags
from .credentials import Credentials
from .helpers import 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
JSONDecodeError: type[ValueError]
try:
from json.decoder import JSONDecodeError
except ImportError:
JSONDecodeError = ValueError
class Bucket:
"""Bucket information."""
def __init__(self, name: str, creation_date: datetime | None):
self._name = name
self._creation_date = creation_date
@property
def name(self) -> str:
"""Get name."""
return self._name
@property
def creation_date(self) -> datetime | None:
"""Get creation date."""
return self._creation_date
def __repr__(self):
return f"{type(self).__name__}('{self.name}')"
def __str__(self):
return self.name
def __eq__(self, other):
if isinstance(other, Bucket):
return self.name == other.name
if isinstance(other, str):
return self.name == other
return NotImplemented
def __hash__(self):
return hash(self.name)
A = TypeVar("A", bound="ListAllMyBucketsResult")
class ListAllMyBucketsResult:
"""LissBuckets API result."""
def __init__(self, buckets: list[Bucket]):
self._buckets = buckets
@property
def buckets(self):
"""Get buckets."""
return self._buckets
@classmethod
def fromxml(cls: Type[A], element: ET.Element) -> A:
"""Create new object with values from XML element."""
element = cast(ET.Element, find(element, "Buckets", True))
buckets = []
elements = findall(element, "Bucket")
for bucket in elements:
name = cast(str, findtext(bucket, "Name", True))
creation_date = findtext(bucket, "CreationDate")
buckets.append(Bucket(
name,
from_iso8601utc(creation_date) if creation_date else None,
))
return cls(buckets)
B = TypeVar("B", bound="Object")
class Object:
"""Object information."""
def __init__( # pylint: disable=too-many-arguments
self,
bucket_name: str,
object_name: str | None,
last_modified: datetime | None = None,
etag: str | None = None,
size: int | None = None,
metadata: dict[str, str] | HTTPHeaderDict | None = None,
version_id: str | None = None,
is_latest: str | None = None,
storage_class: str | None = None,
owner_id: str | None = None,
owner_name: str | None = None,
content_type: str | None = None,
is_delete_marker: bool = False,
tags: Tags | None = None,
):
self._bucket_name = bucket_name
self._object_name = object_name
self._last_modified = last_modified
self._etag = etag
self._size = size
self._metadata = metadata
self._version_id = version_id
self._is_latest = is_latest
self._storage_class = storage_class
self._owner_id = owner_id
self._owner_name = owner_name
self._content_type = content_type
self._is_delete_marker = is_delete_marker
self._tags = tags
@property
def bucket_name(self) -> str:
"""Get bucket name."""
return self._bucket_name
@property
def object_name(self) -> str | None:
"""Get object name."""
return self._object_name
@property
def is_dir(self) -> bool:
"""Get whether this key is a directory."""
return (
self._object_name is not None and self._object_name.endswith("/")
)
@property
def last_modified(self) -> datetime | None:
"""Get last modified time."""
return self._last_modified
@property
def etag(self) -> str | None:
"""Get etag."""
return self._etag
@property
def size(self) -> int | None:
"""Get size."""
return self._size
@property
def metadata(self) -> dict[str, str] | HTTPHeaderDict | None:
"""Get metadata."""
return self._metadata
@property
def version_id(self) -> str | None:
"""Get version ID."""
return self._version_id
@property
def is_latest(self) -> str | None:
"""Get is-latest flag."""
return self._is_latest
@property
def storage_class(self) -> str | None:
"""Get storage class."""
return self._storage_class
@property
def owner_id(self) -> str | None:
"""Get owner ID."""
return self._owner_id
@property
def owner_name(self) -> str | None:
"""Get owner name."""
return self._owner_name
@property
def is_delete_marker(self) -> bool:
"""Get whether this key is a delete marker."""
return self._is_delete_marker
@property
def content_type(self) -> str | None:
"""Get content type."""
return self._content_type
@property
def tags(self) -> Tags | None:
"""Get the tags"""
return self._tags
@classmethod
def fromxml(
cls: Type[B],
element: ET.Element,
bucket_name: str,
is_delete_marker: bool = False,
encoding_type: str | None = None,
) -> B:
"""Create new object with values from XML element."""
tag = findtext(element, "LastModified")
last_modified = None if tag is None else from_iso8601utc(tag)
tag = findtext(element, "ETag")
etag = None if tag is None else tag.replace('"', "")
tag = findtext(element, "Size")
size = None if tag is None else int(tag)
elem = find(element, "Owner")
owner_id, owner_name = (
(None, None) if elem is None
else (findtext(elem, "ID"), findtext(elem, "DisplayName"))
)
elems: ET.Element | list = find(element, "UserMetadata") or []
metadata: dict[str, str] = {}
for child in elems:
key = child.tag.split("}")[1] if "}" in child.tag else child.tag
metadata[key] = child.text or ""
object_name = cast(str, findtext(element, "Key", True))
if encoding_type == "url":
object_name = unquote_plus(object_name)
tags_text = findtext(element, "UserTags")
tags: Tags | None = None
if tags_text:
tags = Tags.new_object_tags()
tags.update(
cast(
List[Tuple[Any, Any]],
[tokens.split("=") for tokens in tags_text.split("&")],
),
)
return cls(
bucket_name,
object_name,
last_modified=last_modified,
etag=etag,
size=size,
version_id=findtext(element, "VersionId"),
is_latest=findtext(element, "IsLatest"),
storage_class=findtext(element, "StorageClass"),
owner_id=owner_id,
owner_name=owner_name,
metadata=metadata,
is_delete_marker=is_delete_marker,
tags=tags
)
def parse_list_objects(
response: BaseHTTPResponse,
bucket_name: str | None = None,
) -> tuple[list[Object], bool, str | None, str | None]:
"""Parse ListObjects/ListObjectsV2/ListObjectVersions response."""
element = ET.fromstring(response.data.decode())
bucket_name = cast(str, findtext(element, "Name", True))
encoding_type = findtext(element, "EncodingType")
elements = findall(element, "Contents")
objects = [
Object.fromxml(tag, bucket_name, encoding_type=encoding_type)
for tag in elements
]
marker = objects[-1].object_name if objects else None
elements = findall(element, "Version")
objects += [
Object.fromxml(tag, bucket_name, encoding_type=encoding_type)
for tag in elements
]
elements = findall(element, "CommonPrefixes")
objects += [
Object(
bucket_name, unquote_plus(findtext(tag, "Prefix", True) or "")
if encoding_type == "url" else findtext(tag, "Prefix", True)
) for tag in elements
]
elements = findall(element, "DeleteMarker")
objects += [
Object.fromxml(tag, bucket_name, is_delete_marker=True,
encoding_type=encoding_type)
for tag in elements
]
is_truncated = (findtext(element, "IsTruncated") or "").lower() == "true"
key_marker = findtext(element, "NextKeyMarker")
if key_marker and encoding_type == "url":
key_marker = unquote_plus(key_marker)
version_id_marker = findtext(element, "NextVersionIdMarker")
continuation_token = findtext(element, "NextContinuationToken")
if key_marker is not None:
continuation_token = key_marker
if continuation_token is None:
continuation_token = findtext(element, "NextMarker")
if continuation_token and encoding_type == "url":
continuation_token = unquote_plus(continuation_token)
if continuation_token is None and is_truncated:
continuation_token = marker
return objects, is_truncated, continuation_token, version_id_marker
class CompleteMultipartUploadResult:
"""CompleteMultipartUpload API result."""
def __init__(self, response: BaseHTTPResponse):
element = ET.fromstring(response.data.decode())
self._bucket_name = findtext(element, "Bucket")
self._object_name = findtext(element, "Key")
self._location = findtext(element, "Location")
self._etag = findtext(element, "ETag")
if self._etag:
self._etag = self._etag.replace('"', "")
self._version_id = response.headers.get("x-amz-version-id")
self._http_headers = response.headers
@property
def bucket_name(self) -> str | None:
"""Get bucket name."""
return self._bucket_name
@property
def object_name(self) -> str | None:
"""Get object name."""
return self._object_name
@property
def location(self) -> str | None:
"""Get location."""
return self._location
@property
def etag(self) -> str | None:
"""Get etag."""
return self._etag
@property
def version_id(self) -> str | None:
"""Get version ID."""
return self._version_id
@property
def http_headers(self) -> HTTPHeaderDict:
"""Get HTTP headers."""
return self._http_headers
C = TypeVar("C", bound="Part")
class Part:
"""Part information of a multipart upload."""
def __init__(
self,
part_number: int,
etag: str,
last_modified: datetime | None = None,
size: int | None = None,
):
self._part_number = part_number
self._etag = etag
self._last_modified = last_modified
self._size = size
@property
def part_number(self) -> int:
"""Get part number. """
return self._part_number
@property
def etag(self) -> str:
"""Get etag."""
return self._etag
@property
def last_modified(self) -> datetime | None:
"""Get last-modified."""
return self._last_modified
@property
def size(self) -> int | None:
"""Get size."""
return self._size
@classmethod
def fromxml(cls: Type[C], element: ET.Element) -> C:
"""Create new object with values from XML element."""
part_number = int(cast(str, findtext(element, "PartNumber", True)))
etag = cast(str, findtext(element, "ETag", True))
etag = etag.replace('"', "")
tag = findtext(element, "LastModified")
last_modified = None if tag is None else from_iso8601utc(tag)
size = findtext(element, "Size")
return cls(
part_number, etag, last_modified, int(size) if size else None,
)
class ListPartsResult:
"""ListParts API result."""
def __init__(self, response: BaseHTTPResponse):
element = ET.fromstring(response.data.decode())
self._bucket_name = findtext(element, "Bucket")
self._object_name = findtext(element, "Key")
tag = find(element, "Initiator")
self._initiator_id = (
None if tag is None else findtext(tag, "ID")
)
self._initiator_name = (
None if tag is None else findtext(tag, "DisplayName")
)
tag = find(element, "Owner")
self._owner_id = (
None if tag is None else findtext(tag, "ID")
)
self._owner_name = (
None if tag is None else findtext(tag, "DisplayName")
)
self._storage_class = findtext(element, "StorageClass")
self._part_number_marker = findtext(element, "PartNumberMarker")
next_part_number_marker = findtext(element, "NextPartNumberMarker")
self._next_part_number_marker = (
int(next_part_number_marker) if next_part_number_marker else None
)
max_parts = findtext(element, "MaxParts")
self._max_parts = int(max_parts) if max_parts else None
is_truncated = findtext(element, "IsTruncated")
self._is_truncated = (
is_truncated is not None and is_truncated.lower() == "true"
)
self._parts = [Part.fromxml(tag) for tag in findall(element, "Part")]
@property
def bucket_name(self) -> str | None:
"""Get bucket name."""
return self._bucket_name
@property
def object_name(self) -> str | None:
"""Get object name."""
return self._object_name
@property
def initiator_id(self) -> str | None:
"""Get initiator ID."""
return self._initiator_id
@property
def initator_name(self) -> str | None:
"""Get initiator name."""
return self._initiator_name
@property
def owner_id(self) -> str | None:
"""Get owner ID."""
return self._owner_id
@property
def owner_name(self) -> str | None:
"""Get owner name."""
return self._owner_name
@property
def storage_class(self) -> str | None:
"""Get storage class."""
return self._storage_class
@property
def part_number_marker(self) -> str | None:
"""Get part number marker."""
return self._part_number_marker
@property
def next_part_number_marker(self) -> int | None:
"""Get next part number marker."""
return self._next_part_number_marker
@property
def max_parts(self) -> int | None:
"""Get max parts."""
return self._max_parts
@property
def is_truncated(self) -> bool:
"""Get is-truncated flag."""
return self._is_truncated
@property
def parts(self) -> list[Part]:
"""Get parts."""
return self._parts
class Upload:
""" Upload information of a multipart upload."""
def __init__(self, element: ET.Element, encoding_type: str | None = None):
self._encoding_type = encoding_type
object_name = cast(str, findtext(element, "Key", True))
self._object_name = (
unquote_plus(object_name) if self._encoding_type == "url"
else object_name
)
self._upload_id = findtext(element, "UploadId")
tag = find(element, "Initiator")
self._initiator_id = (
None if tag is None else findtext(tag, "ID")
)
self._initiator_name = (
None if tag is None else findtext(tag, "DisplayName")
)
tag = find(element, "Owner")
self._owner_id = (
None if tag is None else findtext(tag, "ID")
)
self._owner_name = (
None if tag is None else findtext(tag, "DisplayName")
)
self._storage_class = findtext(element, "StorageClass")
initiated_time = findtext(element, "Initiated")
self._initiated_time = (
from_iso8601utc(initiated_time) if initiated_time else None
)
@property
def object_name(self) -> str:
"""Get object name."""
return self._object_name
@property
def initiator_id(self) -> str | None:
"""Get initiator ID."""
return self._initiator_id
@property
def initator_name(self) -> str | None:
"""Get initiator name."""
return self._initiator_name
@property
def owner_id(self) -> str | None:
"""Get owner ID."""
return self._owner_id
@property
def owner_name(self) -> str | None:
"""Get owner name."""
return self._owner_name
@property
def storage_class(self) -> str | None:
"""Get storage class."""
return self._storage_class
@property
def upload_id(self) -> str | None:
"""Get upload ID."""
return self._upload_id
@property
def initiated_time(self) -> datetime | None:
"""Get initiated time."""
return self._initiated_time
class ListMultipartUploadsResult:
"""ListMultipartUploads API result."""
def __init__(self, response: BaseHTTPResponse):
element = ET.fromstring(response.data.decode())
self._encoding_type = findtext(element, "EncodingType")
self._bucket_name = findtext(element, "Bucket")
self._key_marker = findtext(element, "KeyMarker")
if self._key_marker:
self._key_marker = (
unquote_plus(self._key_marker) if self._encoding_type == "url"
else self._key_marker
)
self._upload_id_marker = findtext(element, "UploadIdMarker")
self._next_key_marker = findtext(element, "NextKeyMarker")
if self._next_key_marker:
self._next_key_marker = (
unquote_plus(self._next_key_marker)
if self._encoding_type == "url" else self._next_key_marker
)
self._next_upload_id_marker = findtext(element, "NextUploadIdMarker")
max_uploads = findtext(element, "MaxUploads")
self._max_uploads = int(max_uploads) if max_uploads else None
is_truncated = findtext(element, "IsTruncated")
self._is_truncated = (
is_truncated is not None and is_truncated.lower() == "true"
)
self._uploads = [
Upload(tag, self._encoding_type)
for tag in findall(element, "Upload")
]
@property
def bucket_name(self) -> str | None:
"""Get bucket name."""
return self._bucket_name
@property
def key_marker(self) -> str | None:
"""Get key marker."""
return self._key_marker
@property
def upload_id_marker(self) -> str | None:
"""Get upload ID marker."""
return self._upload_id_marker
@property
def next_key_marker(self) -> str | None:
"""Get next key marker."""
return self._next_key_marker
@property
def next_upload_id_marker(self) -> str | None:
"""Get next upload ID marker."""
return self._next_upload_id_marker
@property
def max_uploads(self) -> int | None:
"""Get max uploads."""
return self._max_uploads
@property
def is_truncated(self) -> bool:
"""Get is-truncated flag."""
return self._is_truncated
@property
def encoding_type(self) -> str | None:
"""Get encoding type."""
return self._encoding_type
@property
def uploads(self) -> list[Upload]:
"""Get uploads."""
return self._uploads
_RESERVED_ELEMENTS = (
"bucket",
"x-amz-algorithm",
"x-amz-credential",
"x-amz-date",
"policy",
"x-amz-signature",
)
_EQ = "eq"
_STARTS_WITH = "starts-with"
_ALGORITHM = "AWS4-HMAC-SHA256"
def _trim_dollar(value: str) -> str:
"""Trim dollar character if present."""
return value[1:] if value.startswith("$") else value
class PostPolicy:
"""
Post policy information to be used to generate presigned post policy
form-data. Condition elements and respective condition for Post policy
is available at
https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-HTTPPOSTConstructPolicy.html#sigv4-PolicyConditions
"""
def __init__(self, bucket_name: str, expiration: datetime):
check_bucket_name(bucket_name)
if not isinstance(expiration, datetime):
raise ValueError("expiration must be datetime type")
self._bucket_name = bucket_name
self._expiration = expiration
self._conditions: OrderedDict = OrderedDict()
self._conditions[_EQ] = OrderedDict()
self._conditions[_STARTS_WITH] = OrderedDict()
self._lower_limit: int | None = None
self._upper_limit: int | None = None
def add_equals_condition(self, element: str, value: str):
"""Add equals condition of an element and value."""
if not element:
raise ValueError("condition element cannot be empty")
element = _trim_dollar(element)
if (
element in [
"success_action_redirect",
"redirect",
"content-length-range",
]
):
raise ValueError(element + " is unsupported for equals condition")
if element in _RESERVED_ELEMENTS:
raise ValueError(element + " cannot be set")
self._conditions[_EQ][element] = value
def remove_equals_condition(self, element: str):
"""Remove previously set equals condition of an element."""
if not element:
raise ValueError("condition element cannot be empty")
self._conditions[_EQ].pop(element)
def add_starts_with_condition(self, element: str, value: str):
"""
Add starts-with condition of an element and value. Value set to empty
string does matching any content condition.
"""
if not element:
raise ValueError("condition element cannot be empty")
element = _trim_dollar(element)
if (
element in ["success_action_status", "content-length-range"] or
(
element.startswith("x-amz-") and
not element.startswith("x-amz-meta-")
)
):
raise ValueError(
f"{element} is unsupported for starts-with condition",
)
if element in _RESERVED_ELEMENTS:
raise ValueError(element + " cannot be set")
self._conditions[_STARTS_WITH][element] = value
def remove_starts_with_condition(self, element: str):
"""Remove previously set starts-with condition of an element."""
if not element:
raise ValueError("condition element cannot be empty")
self._conditions[_STARTS_WITH].pop(element)
def add_content_length_range_condition( # pylint: disable=invalid-name
self, lower_limit: int, upper_limit: int):
"""Add content-length-range condition with lower and upper limits."""
if lower_limit < 0:
raise ValueError("lower limit cannot be negative number")
if upper_limit < 0:
raise ValueError("upper limit cannot be negative number")
if lower_limit > upper_limit:
raise ValueError("lower limit cannot be greater than upper limit")
self._lower_limit = lower_limit
self._upper_limit = upper_limit
def remove_content_length_range_condition( # pylint: disable=invalid-name
self):
"""Remove previously set content-length-range condition."""
self._lower_limit = None
self._upper_limit = None
def form_data(self, creds: Credentials, region: str):
"""
Return form-data of this post policy. The returned dict contains
x-amz-algorithm, x-amz-credential, x-amz-security-token, x-amz-date,
policy and x-amz-signature.
"""
if not isinstance(creds, Credentials):
raise ValueError("credentials must be Credentials type")
if not region:
raise ValueError("region cannot be empty")
if (
"key" not in self._conditions[_EQ] and
"key" not in self._conditions[_STARTS_WITH]
):
raise ValueError("key condition must be set")
policy: OrderedDict = OrderedDict()
policy["expiration"] = to_iso8601utc(self._expiration)
policy["conditions"] = [[_EQ, "$bucket", self._bucket_name]]
for cond_key, conditions in self._conditions.items():
for key, value in conditions.items():
policy["conditions"].append([cond_key, "$"+key, value])
if self._lower_limit is not None and self._upper_limit is not None:
policy["conditions"].append(
["content-length-range", self._lower_limit, self._upper_limit],
)
utcnow = datetime.utcnow()
credential = get_credential_string(creds.access_key, utcnow, region)
amz_date = to_amz_date(utcnow)
policy["conditions"].append([_EQ, "$x-amz-algorithm", _ALGORITHM])
policy["conditions"].append([_EQ, "$x-amz-credential", credential])
if creds.session_token:
policy["conditions"].append(
[_EQ, "$x-amz-security-token", creds.session_token],
)
policy["conditions"].append([_EQ, "$x-amz-date", amz_date])
policy_encoded = base64.b64encode(
json.dumps(policy).encode(),
).decode("utf-8")
signature = post_presign_v4(
policy_encoded, creds.secret_key, utcnow, region,
)
form_data = {
"x-amz-algorithm": _ALGORITHM,
"x-amz-credential": credential,
"x-amz-date": amz_date,
"policy": policy_encoded,
"x-amz-signature": signature,
}
if creds.session_token:
form_data["x-amz-security-token"] = creds.session_token
return form_data
@property
def bucket_name(self) -> str:
"""Get bucket name."""
return self._bucket_name
def parse_copy_object(
response: BaseHTTPResponse,
) -> tuple[str, datetime | None]:
"""Parse CopyObject/UploadPartCopy response."""
element = ET.fromstring(response.data.decode())
etag = cast(str, findtext(element, "ETag", True)).replace('"', "")
last_modified = findtext(element, "LastModified")
return etag, from_iso8601utc(last_modified) if last_modified else None
class EventIterable:
"""Context manager friendly event iterable."""
def __init__(self, func):
self._func = func
self._response = None
def _close_response(self):
"""Close response."""
if self._response:
self._response.close()
self._response.release_conn()
self._response = None
def __iter__(self):
return self
def _get_records(self):
"""Get event records from response stream."""
try:
line = self._response.readline().strip()
if not line:
return None
if hasattr(line, 'decode'):
line = line.decode()
event = json.loads(line)
if event['Records']:
return event
except (StopIteration, JSONDecodeError):
self._close_response()
return None
def __next__(self):
records = None
while not records:
if not self._response or self._response.closed:
self._response = self._func()
records = self._get_records()
return records
def __enter__(self):
return self
def __exit__(self, exc_type, value, traceback):
self._close_response()
class PeerSite:
"""Represents a cluster/site to be added to the set of replicated sites."""
def __init__(
self,
name: str,
endpoint: str,
access_key: str,
secret_key: str,
):
self._name = name
self._endpoint = endpoint
self._access_key = access_key
self._secret_key = secret_key
def to_dict(self) -> dict[str, str]:
"""Convert to dictionary."""
return {
"name": self._name,
"endpoints": self._endpoint,
"accessKey": self._access_key,
"secretKey": self._secret_key,
}
class SiteReplicationStatusOptions:
"""Represents site replication status options."""
ENTITY_TYPE = Enum(
"ENTITY_TYPE",
{
"BUCKET": "bucket",
"POLICY": "policy",
"USER": "user",
"GROUP": "group",
},
)
def __init__(self):
self._buckets = False
self._policies = False
self._users = False
self._groups = False
self._metrics = False
self._entity = None
self._entity_value = None
self._show_deleted = False
@property
def buckets(self) -> bool:
"""Get buckets."""
return self._buckets
@buckets.setter
def buckets(self, value: bool):
"""Set buckets."""
self._buckets = value
@property
def policies(self) -> bool:
"""Get policies."""
return self._policies
@policies.setter
def policies(self, value: bool):
"""Set policies."""
self._policies = value
@property
def users(self) -> bool:
"""Get users."""
return self._users
@users.setter
def users(self, value: bool):
"""Set users."""
self._users = value
@property
def groups(self) -> bool:
"""Get groups."""
return self._groups
@groups.setter
def groups(self, value: bool):
"""Set groups."""
View remainder of file in raw view
0