# -*- coding: utf-8 -*-
# Asynchronous MinIO Client SDK for Python
# (C) 2021 MinIO, Inc.
# (C) 2022 Huseyn Mashadiyev <mashadiyev.huseyn@gmail.com>
# (C) 2022 L-ING <hlf01@icloud.com>
#
# 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.
"""
Request/response of PutBucketNotificationConfiguration and
GetBucketNotiicationConfiguration APIs.
"""
from __future__ import absolute_import, annotations
from abc import ABCMeta
from typing import Type, TypeVar, cast
from xml.etree import ElementTree as ET
from .xml import Element, SubElement, find, findall, findtext
A = TypeVar("A", bound="FilterRule")
[docs]
class FilterRule:
"""Filter rule."""
__metaclass__ = ABCMeta
def __init__(self, name: str, value: str):
self._name = name
self._value = value
@property
def name(self) -> str:
"""Get name."""
return self._name
@property
def value(self) -> str:
"""Get value."""
return self._value
[docs]
@classmethod
def fromxml(cls: Type[A], element: ET.Element) -> A:
"""Create new object with values from XML element."""
name = cast(str, findtext(element, "Name", True))
value = cast(str, findtext(element, "Value", True))
return cls(name, value)
[docs]
def toxml(self, element: ET.Element | None) -> ET.Element:
"""Convert to XML."""
if element is None:
raise ValueError("element must be provided")
element = SubElement(element, "FilterRule")
SubElement(element, "Name", self._name)
SubElement(element, "Value", self._value)
return element
[docs]
class PrefixFilterRule(FilterRule):
"""Prefix filter rule."""
def __init__(self, value: str):
super().__init__("prefix", value)
[docs]
class SuffixFilterRule(FilterRule):
"""Suffix filter rule."""
def __init__(self, value: str):
super().__init__("suffix", value)
[docs]
class CommonConfig:
"""Common for cloud-function/queue/topic configuration."""
__metaclass__ = ABCMeta
def __init__(
self,
events: list[str],
config_id: str | None,
prefix_filter_rule: PrefixFilterRule | None,
suffix_filter_rule: SuffixFilterRule | None,
):
if not events:
raise ValueError("events must be provided")
self._events = events
self._config_id = config_id
self._prefix_filter_rule = prefix_filter_rule
self._suffix_filter_rule = suffix_filter_rule
@property
def events(self) -> list[str]:
"""Get events."""
return self._events
@property
def config_id(self) -> str | None:
"""Get configuration ID."""
return self._config_id
@property
def prefix_filter_rule(self) -> PrefixFilterRule | None:
"""Get prefix filter rule."""
return self._prefix_filter_rule
@property
def suffix_filter_rule(self) -> SuffixFilterRule | None:
"""Get suffix filter rule."""
return self._suffix_filter_rule
[docs]
@staticmethod
def parsexml(
element: ET.Element,
) -> tuple[list[str], str | None, PrefixFilterRule | None, SuffixFilterRule | None]:
"""Parse XML."""
elements = findall(element, "Event")
events = []
for tag in elements:
if tag.text is None:
raise ValueError("missing value in XML tag 'Event'")
events.append(tag.text)
config_id = findtext(element, "Id")
elem = find(element, "Filter")
if elem is None:
return events, config_id, None, None
prefix_filter_rule = None
suffix_filter_rule = None
elem = cast(ET.Element, find(elem, "S3Key", True))
elements = findall(elem, "FilterRule")
for tag in elements:
filter_rule = FilterRule.fromxml(tag)
if filter_rule.name == "prefix":
prefix_filter_rule = PrefixFilterRule(filter_rule.value)
else:
suffix_filter_rule = SuffixFilterRule(filter_rule.value)
return events, config_id, prefix_filter_rule, suffix_filter_rule
[docs]
def toxml(self, element: ET.Element | None) -> ET.Element:
"""Convert to XML."""
if element is None:
raise ValueError("element must be provided")
for event in self._events:
SubElement(element, "Event", event)
if self._config_id is not None:
SubElement(element, "Id", self._config_id)
if self._prefix_filter_rule or self._suffix_filter_rule:
rule = SubElement(element, "Filter")
rule = SubElement(rule, "S3Key")
if self._prefix_filter_rule:
self._prefix_filter_rule.toxml(rule)
if self._suffix_filter_rule:
self._suffix_filter_rule.toxml(rule)
return element
B = TypeVar("B", bound="CloudFuncConfig")
[docs]
class CloudFuncConfig(CommonConfig):
"""Cloud function configuration."""
def __init__(
self,
cloud_func: str,
events: list[str],
config_id: str | None = None,
prefix_filter_rule: PrefixFilterRule | None = None,
suffix_filter_rule: SuffixFilterRule | None = None,
):
if not cloud_func:
raise ValueError("cloud function must be provided")
self._cloud_func = cloud_func
super().__init__(
events,
config_id,
prefix_filter_rule,
suffix_filter_rule,
)
@property
def cloud_func(self) -> str:
"""Get cloud function ARN."""
return self._cloud_func
[docs]
@classmethod
def fromxml(cls: Type[B], element: ET.Element) -> B:
"""Create new object with values from XML element."""
cloud_func = cast(str, findtext(element, "CloudFunction", True))
(events, config_id, prefix_filter_rule, suffix_filter_rule) = cls.parsexml(
element
)
return cls(
cloud_func, events, config_id, prefix_filter_rule, suffix_filter_rule
)
[docs]
def toxml(self, element: ET.Element | None) -> ET.Element:
"""Convert to XML."""
if element is None:
raise ValueError("element must be provided")
element = SubElement(element, "CloudFunctionConfiguration")
SubElement(element, "CloudFunction", self._cloud_func)
super().toxml(element)
return element
C = TypeVar("C", bound="QueueConfig")
[docs]
class QueueConfig(CommonConfig):
"""Queue configuration."""
def __init__(
self,
queue: str,
events: list[str],
config_id: str | None = None,
prefix_filter_rule: PrefixFilterRule | None = None,
suffix_filter_rule: SuffixFilterRule | None = None,
):
if not queue:
raise ValueError("queue must be provided")
self._queue = queue
super().__init__(
events,
config_id,
prefix_filter_rule,
suffix_filter_rule,
)
@property
def queue(self) -> str:
"""Get queue ARN."""
return self._queue
[docs]
@classmethod
def fromxml(cls: Type[C], element: ET.Element) -> C:
"""Create new object with values from XML element."""
queue = cast(str, findtext(element, "Queue", True))
(events, config_id, prefix_filter_rule, suffix_filter_rule) = cls.parsexml(
element
)
return cls(queue, events, config_id, prefix_filter_rule, suffix_filter_rule)
[docs]
def toxml(self, element: ET.Element | None) -> ET.Element:
"""Convert to XML."""
if element is None:
raise ValueError("element must be provided")
element = SubElement(element, "QueueConfiguration")
SubElement(element, "Queue", self._queue)
super().toxml(element)
return element
D = TypeVar("D", bound="TopicConfig")
[docs]
class TopicConfig(CommonConfig):
"""Get topic configuration."""
def __init__(
self,
topic: str,
events: list[str],
config_id: str | None = None,
prefix_filter_rule: PrefixFilterRule | None = None,
suffix_filter_rule: SuffixFilterRule | None = None,
):
if not topic:
raise ValueError("topic must be provided")
self._topic = topic
super().__init__(
events,
config_id,
prefix_filter_rule,
suffix_filter_rule,
)
@property
def topic(self) -> str:
"""Get topic ARN."""
return self._topic
[docs]
@classmethod
def fromxml(cls: Type[D], element: ET.Element) -> D:
"""Create new object with values from XML element."""
topic = cast(str, findtext(element, "Topic", True))
(events, config_id, prefix_filter_rule, suffix_filter_rule) = cls.parsexml(
element
)
return cls(topic, events, config_id, prefix_filter_rule, suffix_filter_rule)
[docs]
def toxml(self, element: ET.Element | None) -> ET.Element:
"""Convert to XML."""
if element is None:
raise ValueError("element must be provided")
element = SubElement(element, "TopicConfiguration")
SubElement(element, "Topic", self._topic)
super().toxml(element)
return element
E = TypeVar("E", bound="NotificationConfig")
[docs]
class NotificationConfig:
"""Notification configuration."""
def __init__(
self,
cloud_func_config_list: list[CloudFuncConfig] | None = None,
queue_config_list: list[QueueConfig] | None = None,
topic_config_list: list[TopicConfig] | None = None,
):
self._cloud_func_config_list = cloud_func_config_list or []
self._queue_config_list = queue_config_list or []
self._topic_config_list = topic_config_list or []
@property
def cloud_func_config_list(self) -> list[CloudFuncConfig] | None:
"""Get cloud function configuration list."""
return self._cloud_func_config_list
@property
def queue_config_list(self) -> list[QueueConfig] | None:
"""Get queue configuration list."""
return self._queue_config_list
@property
def topic_config_list(self) -> list[TopicConfig] | None:
"""Get topic configuration list."""
return self._topic_config_list
[docs]
@classmethod
def fromxml(cls: Type[E], element: ET.Element) -> E:
"""Create new object with values from XML element."""
elements = findall(element, "CloudFunctionConfiguration")
cloud_func_config_list = []
for tag in elements:
cloud_func_config_list.append(CloudFuncConfig.fromxml(tag))
elements = findall(element, "QueueConfiguration")
queue_config_list = []
for tag in elements:
queue_config_list.append(QueueConfig.fromxml(tag))
elements = findall(element, "TopicConfiguration")
topic_config_list = []
for tag in elements:
topic_config_list.append(TopicConfig.fromxml(tag))
return cls(
cloud_func_config_list,
queue_config_list,
topic_config_list,
)
[docs]
def toxml(self, element: ET.Element | None) -> ET.Element:
"""Convert to XML."""
element = Element("NotificationConfiguration")
for cloud_func_config in self._cloud_func_config_list:
cloud_func_config.toxml(element)
for queue_config in self._queue_config_list:
queue_config.toxml(element)
for config in self._topic_config_list:
config.toxml(element)
return element