feat: add connections string builders

Author: u-chaCP

connections/builders/__init__.py

@@ -1 +1 @@
-
+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 and_(self) -> "ConnectionStringBuilder":
+        self._parent.cb.add_option("customlabels", CloudProtocol.compose_custom_labels_string(self._custom_labels))
+        return super().and_()
+
+    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 and_(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

connections/builders/text_protocol_connection_string_builder.py

@@ -0,0 +1,21 @@
+from typing import TYPE_CHECKING, TypeVar
+
+if TYPE_CHECKING:
+    from connection_string_builder import ConnectionStringBuilder
+
+from connections.builders.file_protocol_connection_string_builder import FileProtocolConnectionStringBuilder
+
+Self = TypeVar("Self", bound="TextProtocolConnectionStringBuilder")
+
+
+class TextProtocolConnectionStringBuilder(FileProtocolConnectionStringBuilder):
+    def __init__(self, parent: "ConnectionStringBuilder"):
+        super().__init__(parent)
+
+    def set_pattern(self, pattern: str) -> Self:
+        self._parent.cb.add_option("pattern", pattern)
+        return self
+
+    def set_indent(self, indent: bool) -> Self:
+        self._parent.cb.add_option("indent", indent)
+        return self