Merge pull request #16 from Code-Partners/feat-conn-string-builders

Feat conn string builders

Author: SergeyCP

common/file_rotate.py

@@ -7,3 +7,6 @@ class FileRotate(Enum):
     DAILY = 2
     WEEKLY = 3
     MONTHLY = 4
+
+    def __str__(self):
+        return "%s" % self._name_

common/lookup_table.py

@@ -105,8 +105,9 @@ def clear(self):
     def get_count(self):
         return len(self.__items)
 
-    def __is_valid_integer(self, value: str) -> bool:
-        if self.__value_is_valid(value):
+    @classmethod
+    def __is_valid_integer(cls, value: str) -> bool:
+        if cls.__value_is_valid(value):
             try:
                 int(value)
                 return True
@@ -145,30 +146,31 @@ def get_size_value(self, key: str, default_value: int) -> int:
 
         return result
 
-    def size_to_int(self, value: str, default_value: int) -> int:
+    @classmethod
+    def size_to_int(cls, value: str, default_value: int) -> int:
         if not isinstance(value, str):
             raise TypeError("Value must be a string")
         if not isinstance(default_value, int):
             raise TypeError("Default value must be an int")
 
         result = default_value
-        factor = self.__KB_FACTOR
+        factor = cls.__KB_FACTOR
         value = value.strip()
 
         if len(value) >= 2:
             unit = value[-2:].lower()
 
-            if self.__is_valid_size_unit(unit):
+            if cls.__is_valid_size_unit(unit):
                 value = value[:-2].strip()
 
                 if unit == "kb":
-                    factor = self.__KB_FACTOR
+                    factor = cls.__KB_FACTOR
                 elif unit == "mb":
-                    factor = self.__MB_FACTOR
+                    factor = cls.__MB_FACTOR
                 elif unit == "gb":
-                    factor = self.__GB_FACTOR
+                    factor = cls.__GB_FACTOR
 
-        if self.__is_valid_integer(value):
+        if cls.__is_valid_integer(value):
             try:
                 result = factor * int(value)
             except ValueError:

connections/builders/__init__.py

@@ -1 +1 @@
-from connections.builders.connections_builder import ConnectionsBuilder
+from connections.builders.connection_string_builder import ConnectionStringBuilder

connections/builders/cloud_protocol_connection_string_builder.py

@@ -0,0 +1,70 @@
+import collections
+from typing import TYPE_CHECKING, TypeVar
+
+from common.file_rotate import FileRotate
+from common.lookup_table import LookupTable
+from protocols.cloud.cloud_protocol import CloudProtocol
+
+if TYPE_CHECKING:
+    from connection_string_builder import ConnectionStringBuilder
+
+from connections.builders.tcp_protocol_connection_string_builder import TcpProtocolConnectionStringBuilder
+
+Self = TypeVar("Self", bound="CloudProtocolConnectionStringBuilder")
+
+
+class CloudProtocolConnectionStringBuilder(TcpProtocolConnectionStringBuilder):
+    _custom_labels: collections.OrderedDict = collections.OrderedDict()
+
+    def __init__(self, parent: "ConnectionStringBuilder"):
+        super().__init__(parent)
+
+    def end_protocol(self) -> "ConnectionStringBuilder":
+        self._parent.cb.add_option("customlabels", CloudProtocol.compose_custom_labels_string(self._custom_labels))
+        return super().end_protocol()
+
+    def set_write_key(self, write_key: str) -> Self:
+        self._parent.cb.add_option("writekey", write_key)
+        return self
+
+    def add_custom_label(self, key: str, value: str) -> Self:
+        self._custom_labels[key] = value
+        return self
+
+    def set_region(self, region: str) -> Self:
+        self._parent.cb.add_option("region", region)
+        return self
+
+    def set_chunking_enabled(self, enabled: bool) -> Self:
+        self._parent.cb.add_option("chunking.enabled", enabled)
+        return self
+
+    def set_chunking_max_size(self, max_size: str) -> Self:
+        size_in_bytes = LookupTable.size_to_int(max_size, 0)
+        self._parent.cb.add_option("chunking.maxsize", int(size_in_bytes / 1024))
+        return self
+
+    def set_chunking_max_age_ms(self, max_age: int) -> Self:
+        self._parent.cb.add_option("chunking.maxagems", max_age)
+        return self
+
+    def set_max_size(self, max_size: str) -> Self:
+        size_in_bytes = LookupTable.size_to_int(max_size, 0)
+        self._parent.cb.add_option("maxsize", int(size_in_bytes / 1024))
+        return self
+
+    def set_rotate(self, rotate: FileRotate) -> Self:
+        self._parent.cb.add_option("rotate", rotate)
+        return self
+
+    def set_tls_enabled(self, tls_enabled: bool) -> Self:
+        self._parent.cb.add_option("tls.enabled", tls_enabled)
+        return self
+
+    def set_tls_certificate_location(self, tls_certificate_location: str) -> Self:
+        self._parent.cb.add_option("tls.certificate.location", tls_certificate_location)
+        return self
+
+    def set_tls_certificate_filepath(self, tls_certificate_filepath: str) -> Self:
+        self._parent.cb.add_option("tls.certificate.filepath", tls_certificate_filepath)
+        return self

connections/builders/connection_string_builder.py

@@ -0,0 +1,50 @@
+from connections.builders.cloud_protocol_connection_string_builder import CloudProtocolConnectionStringBuilder
+from connections.builders.file_protocol_connection_string_builder import FileProtocolConnectionStringBuilder
+from connections.builders.memory_protocol_connection_string_builder import MemoryProtocolConnectionStringBuilder
+from connections.builders.pipe_protocol_connection_string_builder import PipeProtocolConnectionStringBuilder
+from connections.builders.tcp_protocol_connection_string_builder import TcpProtocolConnectionStringBuilder
+from connections.builders.text_protocol_connection_string_builder import TextProtocolConnectionStringBuilder
+from connections.connections_builder import ConnectionsBuilder
+
+
+class ConnectionStringBuilder:
+    _protocols = {
+        "pipe": PipeProtocolConnectionStringBuilder,
+        "file": FileProtocolConnectionStringBuilder,
+        "mem": MemoryProtocolConnectionStringBuilder,
+        "tcp": TcpProtocolConnectionStringBuilder,
+        "text": TextProtocolConnectionStringBuilder,
+        "cloud": CloudProtocolConnectionStringBuilder,
+    }
+
+    def __init__(self):
+        self._cb = ConnectionsBuilder()
+
+    @property
+    def cb(self):
+        return self._cb
+
+    def _add_protocol(self, protocol_name: str):
+        self._cb.begin_protocol(protocol_name)
+        return self._protocols.get(protocol_name)(self)
+
+    def add_pipe_protocol(self) -> PipeProtocolConnectionStringBuilder:
+        return self._add_protocol("pipe")
+
+    def add_file_protocol(self) -> FileProtocolConnectionStringBuilder:
+        return self._add_protocol("file")
+
+    def add_memory_protocol(self) -> MemoryProtocolConnectionStringBuilder:
+        return self._add_protocol("mem")
+
+    def add_tcp_protocol(self) -> TcpProtocolConnectionStringBuilder:
+        return self._add_protocol("tcp")
+
+    def add_text_protocol(self) -> TextProtocolConnectionStringBuilder:
+        return self._add_protocol("text")
+
+    def add_cloud_protocol(self) -> CloudProtocolConnectionStringBuilder:
+        return self._add_protocol("cloud")
+
+    def build(self) -> str:
+        return self._cb.get_connections()

connections/builders/file_protocol_connection_string_builder.py

@@ -0,0 +1,51 @@
+from typing import TYPE_CHECKING, TypeVar
+
+from common.file_rotate import FileRotate
+from common.lookup_table import LookupTable
+
+if TYPE_CHECKING:
+    from connection_string_builder import ConnectionStringBuilder
+
+from connections.builders.protocol_connection_string_builder import ProtocolConnectionStringBuilder
+
+Self = TypeVar("Self", bound="FileProtocolConnectionStringBuilder")
+
+
+class FileProtocolConnectionStringBuilder(ProtocolConnectionStringBuilder):
+
+    def __init__(self, parent: "ConnectionStringBuilder"):
+        super().__init__(parent)
+
+    def set_filename(self, filename: str) -> Self:
+        self._parent.cb.add_option("filename", filename)
+        return self
+
+    def set_append(self, append: bool) -> Self:
+        self._parent.cb.add_option("append", append)
+        return self
+
+    def set_buffer(self, buffer: str) -> Self:
+        size_in_bytes = LookupTable.size_to_int(buffer, 0)
+        self._parent.cb.add_option("buffer", int(size_in_bytes / 1024))
+        return self
+
+    def set_rotate(self, rotate: FileRotate) -> Self:
+        self._parent.cb.add_option("rotate", rotate)
+        return self
+
+    def set_max_size(self, max_size: str) -> Self:
+        size_in_bytes = LookupTable.size_to_int(max_size, 0)
+        self._parent.cb.add_option("maxsize", int(size_in_bytes / 1024))
+        return self
+
+    def set_max_parts(self, max_parts: int) -> Self:
+        self._parent.cb.add_option("maxparts", max_parts)
+        return self
+
+    def set_key(self, key: str) -> Self:
+        self._parent.cb.add_option("key", key)
+        return self
+
+    def set_encrypt(self, encrypt: bool) -> Self:
+        self._parent.cb.add_option("encrypt", encrypt)
+        return self

connections/builders/memory_protocol_connection_string_builder.py

@@ -0,0 +1,32 @@
+from typing import TYPE_CHECKING, TypeVar
+
+from common.lookup_table import LookupTable
+
+if TYPE_CHECKING:
+    from connection_string_builder import ConnectionStringBuilder
+
+from connections.builders.protocol_connection_string_builder import ProtocolConnectionStringBuilder
+
+Self = TypeVar("Self", bound="MemoryProtocolConnectionStringBuilder")
+
+
+class MemoryProtocolConnectionStringBuilder(ProtocolConnectionStringBuilder):
+    def __init__(self, parent: "ConnectionStringBuilder"):
+        super().__init__(parent)
+
+    def set_max_size(self, max_size: str) -> Self:
+        size_in_bytes = LookupTable.size_to_int(max_size, 0)
+        self._parent.cb.add_option("maxsize", int(size_in_bytes / 1024))
+        return self
+
+    def set_as_text(self, as_text: bool) -> Self:
+        self._parent.cb.add_option("astext", as_text)
+        return self
+
+    def set_indent(self, indent: bool) -> Self:
+        self._parent.cb.add_option("indent", indent)
+        return self
+
+    def set_pattern(self, pattern: str) -> Self:
+        self._parent.cb.add_option("pattern", pattern)
+        return self

connections/builders/pipe_protocol_connection_string_builder.py

@@ -0,0 +1,17 @@
+from typing import TYPE_CHECKING, TypeVar
+
+if TYPE_CHECKING:
+    from connection_string_builder import ConnectionStringBuilder
+
+from connections.builders.protocol_connection_string_builder import ProtocolConnectionStringBuilder
+
+Self = TypeVar("Self", bound="PipeProtocolConnectionStringBuilder")
+
+
+class PipeProtocolConnectionStringBuilder(ProtocolConnectionStringBuilder):
+    def __init__(self, parent: "ConnectionStringBuilder"):
+        super().__init__(parent)
+
+    def set_pipe_name(self, pipe_name: str) -> Self:
+        self._parent.cb.add_option("pipename", pipe_name)
+        return self

connections/builders/protocol_connection_string_builder.py

@@ -0,0 +1,68 @@
+from typing import TYPE_CHECKING, TypeVar
+
+if TYPE_CHECKING:
+    from connection_string_builder import ConnectionStringBuilder
+
+from common.level import Level
+from common.lookup_table import LookupTable
+
+Self = TypeVar("Self", bound="ProtocolConnectionStringBuilder")
+
+
+class ProtocolConnectionStringBuilder:
+    def __init__(self, parent: "ConnectionStringBuilder"):
+        self._parent = parent
+
+    def end_protocol(self) -> "ConnectionStringBuilder":
+        self._parent.cb.end_protocol()
+        return self._parent
+
+    def set_level(self, level: Level) -> Self:
+        self._parent.cb.add_option("level", level)
+        return self
+
+    def set_caption(self, caption: str) -> Self:
+        self._parent.cb.add_option("caption", caption)
+        return self
+
+    def set_reconnect(self, reconnect: bool) -> Self:
+        self._parent.cb.add_option("reconnect", reconnect)
+        return self
+
+    def set_reconnect_interval(self, reconnect_interval: int) -> Self:
+        self._parent.cb.add_option("reconnect.interval", reconnect_interval)
+        return self
+
+    def set_backlog_enabled(self, backlog_enabled: bool) -> Self:
+        self._parent.cb.add_option("backlog.enabled", backlog_enabled)
+        return self
+
+    def set_backlog_queue(self, backlog_queue: str) -> Self:
+        size_in_bytes = LookupTable.size_to_int(backlog_queue, 0)
+        self._parent.cb.add_option("backlog.queue", int(size_in_bytes / 1024))
+        return self
+
+    def set_backlog_flushon(self, backlog_flushon: Level) -> Self:
+        self._parent.cb.add_option("backlog.flushon", backlog_flushon)
+        return self
+
+    def set_backlog_keepopen(self, backlog_keepopen: bool) -> Self:
+        self._parent.cb.add_option("backlog.keepopen", backlog_keepopen)
+        return self
+
+    def set_async_enabled(self, async_enabled: bool) -> Self:
+        self._parent.cb.add_option("async.enabled", async_enabled)
+        return self
+
+    def set_async_throttle(self, async_throttle: bool) -> Self:
+        self._parent.cb.add_option("async.throttle", async_throttle)
+        return self
+
+    def set_async_queue(self, async_queue: str) -> Self:
+        size_in_bytes = LookupTable.size_to_int(async_queue, 0)
+        self._parent.cb.add_option("async.queue", int(size_in_bytes / 1024))
+        return self
+
+    def set_async_clear_on_disconnect(self, async_clear_on_disconnect: bool) -> Self:
+        self._parent.cb.add_option("async.clearondisconnect", async_clear_on_disconnect)
+        return self

connections/builders/tcp_protocol_connection_string_builder.py

@@ -0,0 +1,25 @@
+from typing import TYPE_CHECKING, TypeVar
+
+if TYPE_CHECKING:
+    from connection_string_builder import ConnectionStringBuilder
+
+from connections.builders.protocol_connection_string_builder import ProtocolConnectionStringBuilder
+
+Self = TypeVar("Self", bound="TcpProtocolConnectionStringBuilder")
+
+
+class TcpProtocolConnectionStringBuilder(ProtocolConnectionStringBuilder):
+    def __init__(self, parent: "ConnectionStringBuilder"):
+        super().__init__(parent)
+
+    def set_host(self, host: str) -> Self:
+        self._parent.cb.add_option("host", host)
+        return self
+
+    def set_port(self, port: int) -> Self:
+        self._parent.cb.add_option("port", port)
+        return self
+
+    def set_timeout(self, timeout: int) -> Self:
+        self._parent.cb.add_option("timeout", timeout)
+        return self