Skip to content
Snippets Groups Projects
Commit e7455f29 authored by Moul's avatar Moul
Browse files

[enh] #170: endpoint: Name capturing groups

parent 5722ad54
Branches
Tags
2 merge requests!157v1.0.0rc0: merge dev into master,!145#170: Endpoints' regex improvement
......@@ -143,7 +143,7 @@ BMAEndpointType = TypeVar("BMAEndpointType", bound="BMAEndpoint")
class BMAEndpoint(Endpoint):
API = "BASIC_MERKLED_API"
re_inline = re.compile(
f"^{API}(?: ({const.HOST_REGEX}))?(?: ({const.IPV4_REGEX}))?(?: ({const.IPV6_REGEX}))?(?: ({const.PORT_REGEX}))$"
f"^{API}(?: (?P<host>{const.HOST_REGEX}))?(?: (?P<ipv4>{const.IPV4_REGEX}))?(?: (?P<ipv6>{const.IPV6_REGEX}))?(?: (?P<port>{const.PORT_REGEX}))$"
)
def __init__(self, server: str, ipv4: str, ipv6: str, port: int) -> None:
......@@ -171,10 +171,10 @@ class BMAEndpoint(Endpoint):
m = BMAEndpoint.re_inline.match(inline)
if m is None:
raise MalformedDocumentError(BMAEndpoint.API)
server = m.group(1)
ipv4 = m.group(2)
ipv6 = m.group(3)
port = int(m.group(4))
server = m["host"]
ipv4 = m["ipv4"]
ipv6 = m["ipv6"]
port = int(m["port"])
return cls(server, ipv4, ipv6, port)
def inline(self) -> str:
......@@ -236,7 +236,7 @@ SecuredBMAEndpointType = TypeVar("SecuredBMAEndpointType", bound="SecuredBMAEndp
class SecuredBMAEndpoint(BMAEndpoint):
API = "BMAS"
re_inline = re.compile(
f"^{API}(?: ({const.HOST_REGEX}))?(?: ({const.IPV4_REGEX}))?(?: ({const.IPV6_REGEX}))? ({const.PORT_REGEX})(?: ({const.PATH_REGEX}))?$"
f"^{API}(?: (?P<host>{const.HOST_REGEX}))?(?: (?P<ipv4>{const.IPV4_REGEX}))?(?: (?P<ipv6>{const.IPV6_REGEX}))? (?P<port>{const.PORT_REGEX})(?: (?P<path>{const.PATH_REGEX}))?$"
)
def __init__(self, server: str, ipv4: str, ipv6: str, port: int, path: str) -> None:
......@@ -265,11 +265,11 @@ class SecuredBMAEndpoint(BMAEndpoint):
m = SecuredBMAEndpoint.re_inline.match(inline)
if m is None:
raise MalformedDocumentError(SecuredBMAEndpoint.API)
server = m.group(1)
ipv4 = m.group(2)
ipv6 = m.group(3)
port = int(m.group(4))
path = m.group(5)
server = m["host"]
ipv4 = m["ipv4"]
ipv6 = m["ipv6"]
port = int(m["port"])
path = m["path"]
if not path:
path = ""
return cls(server, ipv4, ipv6, port, path)
......@@ -317,7 +317,7 @@ WS2PEndpointType = TypeVar("WS2PEndpointType", bound="WS2PEndpoint")
class WS2PEndpoint(Endpoint):
API = "WS2P"
re_inline = re.compile(
f"^{API} ({const.WS2PID_REGEX}) ((?:{const.HOST_REGEX})|(?:{const.IPV4_REGEX})|(?:{const.IPV6_REGEX})) ({const.PORT_REGEX})?(?: ({const.PATH_REGEX}))?$"
f"^{API} (?P<ws2pid>{const.WS2PID_REGEX}) (?P<host>(?:{const.HOST_REGEX})|(?:{const.IPV4_REGEX})|(?:{const.IPV6_REGEX})) (?P<port>{const.PORT_REGEX})?(?: (?P<path>{const.PATH_REGEX}))?$"
)
def __init__(self, ws2pid: str, server: str, port: int, path: str) -> None:
......@@ -337,10 +337,10 @@ class WS2PEndpoint(Endpoint):
m = WS2PEndpoint.re_inline.match(inline)
if m is None:
raise MalformedDocumentError(WS2PEndpoint.API)
ws2pid = m.group(1)
server = m.group(2)
port = int(m.group(3))
path = m.group(4)
ws2pid = m["ws2pid"]
server = m["host"]
port = int(m["port"])
path = m["path"]
if not path:
path = ""
return cls(ws2pid, server, port, path)
......@@ -398,7 +398,7 @@ ESCoreEndpointType = TypeVar("ESCoreEndpointType", bound="ESCoreEndpoint")
class ESCoreEndpoint(Endpoint):
API = "ES_CORE_API"
re_inline = re.compile(
f"^{API} ((?:{const.HOST_REGEX})|(?:{const.IPV4_REGEX})) ({const.PORT_REGEX})$"
f"^{API} (?P<host>(?:{const.HOST_REGEX})|(?:{const.IPV4_REGEX})) (?P<port>{const.PORT_REGEX})$"
)
def __init__(self, server: str, port: int) -> None:
......@@ -416,8 +416,8 @@ class ESCoreEndpoint(Endpoint):
m = ESCoreEndpoint.re_inline.match(inline)
if m is None:
raise MalformedDocumentError(ESCoreEndpoint.API)
server = m.group(1)
port = int(m.group(2))
server = m["host"]
port = int(m["port"])
return cls(server, port)
def inline(self) -> str:
......@@ -457,7 +457,7 @@ ESUserEndpointType = TypeVar("ESUserEndpointType", bound="ESUserEndpoint")
class ESUserEndpoint(Endpoint):
API = "ES_USER_API"
re_inline = re.compile(
"^{API} ((?:{const.HOST_REGEX})|(?:{const.IPV4_REGEX})) ({const.PORT_REGEX})$"
"^{API} (?P<host>(?:{const.HOST_REGEX})|(?:{const.IPV4_REGEX})) (?P<port>{const.PORT_REGEX})$"
)
def __init__(self, server: str, port: int) -> None:
......@@ -475,8 +475,8 @@ class ESUserEndpoint(Endpoint):
m = ESUserEndpoint.re_inline.match(inline)
if m is None:
raise MalformedDocumentError(ESUserEndpoint.API)
server = m.group(1)
port = int(m.group(2))
server = m["host"]
port = int(m["port"])
return cls(server, port)
def inline(self) -> str:
......@@ -518,7 +518,7 @@ ESSubscribtionEndpointType = TypeVar(
class ESSubscribtionEndpoint(Endpoint):
API = "ES_SUBSCRIPTION_API"
re_inline = re.compile(
f"^{API} ((?:{const.HOST_REGEX})|(?:{const.IPV4_REGEX})) ({const.PORT_REGEX})$"
f"^{API} (?P<host>(?:{const.HOST_REGEX})|(?:{const.IPV4_REGEX})) (?P<port>{const.PORT_REGEX})$"
)
def __init__(self, server: str, port: int) -> None:
......@@ -538,8 +538,8 @@ class ESSubscribtionEndpoint(Endpoint):
m = ESSubscribtionEndpoint.re_inline.match(inline)
if m is None:
raise MalformedDocumentError(ESSubscribtionEndpoint.API)
server = m.group(1)
port = int(m.group(2))
server = m["host"]
port = int(m["port"])
return cls(server, port)
def inline(self) -> str:
......@@ -578,7 +578,7 @@ GVAEndpointType = TypeVar("GVAEndpointType", bound="GVAEndpoint")
class GVAEndpoint(Endpoint):
API = "GVA"
endpoint_format = f"^GVA(?: ({const.ENDPOINT_FLAGS_REGEX}))?(?: ({const.HOST_REGEX}))?(?: ({const.IPV4_REGEX}))?(?: ({const.IPV6_REGEX}))? ({const.PORT_REGEX})(?: ({const.PATH_REGEX}))?$"
endpoint_format = f"^GVA(?: (?P<flags>{const.ENDPOINT_FLAGS_REGEX}))?(?: (?P<host>{const.HOST_REGEX}))?(?: (?P<ipv4>{const.IPV4_REGEX}))?(?: (?P<ipv6>{const.IPV6_REGEX}))? (?P<port>{const.PORT_REGEX})(?: (?P<path>{const.PATH_REGEX}))?$"
re_inline = re.compile(endpoint_format)
def __init__(
......@@ -618,12 +618,12 @@ class GVAEndpoint(Endpoint):
m = cls.re_inline.match(inline)
if m is None:
raise MalformedDocumentError(cls.API)
flags = m.group(1)
server = m.group(2)
ipv4 = m.group(3)
ipv6 = m.group(4)
port = int(m.group(5))
path = m.group(6)
flags = m["flags"]
server = m["host"]
ipv4 = m["ipv4"]
ipv6 = m["ipv6"]
port = int(m["port"])
path = m["path"]
if not flags:
flags = ""
if not path:
......@@ -707,7 +707,7 @@ GVASUBEndpointType = TypeVar("GVASUBEndpointType", bound="GVASUBEndpoint")
class GVASUBEndpoint(GVAEndpoint):
API = "GVASUB"
endpoint_format = f"^GVASUB(?: ({const.ENDPOINT_FLAGS_REGEX}))?(?: ({const.HOST_REGEX}))?(?: ({const.IPV4_REGEX}))?(?: ({const.IPV6_REGEX}))? ({const.PORT_REGEX})(?: ({const.PATH_REGEX}))?$"
endpoint_format = f"^GVASUB(?: (?P<flags>{const.ENDPOINT_FLAGS_REGEX}))?(?: (?P<host>{const.HOST_REGEX}))?(?: (?P<ipv4>{const.IPV4_REGEX}))?(?: (?P<ipv6>{const.IPV6_REGEX}))? (?P<port>{const.PORT_REGEX})(?: (?P<path>{const.PATH_REGEX}))?$"
re_inline = re.compile(endpoint_format)
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment