Skip to content

Commit

Permalink
fixing check issues (in progress)
Browse files Browse the repository at this point in the history
  • Loading branch information
muddymudskipper committed Jan 12, 2024
1 parent 89449ec commit c342074
Show file tree
Hide file tree
Showing 3 changed files with 159 additions and 107 deletions.
35 changes: 22 additions & 13 deletions cmem_plugin_uuid/transform/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,13 +55,14 @@ class UUID1(TransformPlugin):

def __init__(
self,
node=None,
clock_seq=None,
node: str | None,
clock_seq: str | None,
):
self.node = node_to_int(node) if node else None
self.clock_seq = clock_seq_to_int(clock_seq) if clock_seq else None

def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
"""Transform"""
result = []
if len(inputs) != 0:
for collection in inputs:
Expand Down Expand Up @@ -106,13 +107,14 @@ class UUID3(TransformPlugin):

def __init__(
self,
namespace=None,
namespace_as_uuid=False,
namespace: str | None,
namespace_as_uuid: bool | None,
):
self.namespace = namespace
self.namespace_as_uuid = namespace_as_uuid

def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
"""Transform"""
result = []
namespace_uuid = get_namespace_uuid(
namespace_as_uuid=self.namespace_as_uuid,
Expand Down Expand Up @@ -142,6 +144,7 @@ class UUID4(TransformPlugin):
"""UUID4 Transform Plugin"""

def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
"""Transform"""
result = []
if len(inputs) != 0:
for collection in inputs:
Expand Down Expand Up @@ -184,13 +187,14 @@ class UUID5(TransformPlugin):

def __init__(
self,
namespace=None,
namespace_as_uuid=False,
namespace: str | None,
namespace_as_uuid: bool | None,
):
self.namespace = namespace
self.namespace_as_uuid = namespace_as_uuid

def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
"""Transform"""
result = []
namespace_uuid = get_namespace_uuid(
namespace_as_uuid=self.namespace_as_uuid,
Expand Down Expand Up @@ -249,13 +253,14 @@ class UUID6(TransformPlugin):

def __init__(
self,
node=None,
clock_seq=None,
node: str | None,
clock_seq: str | None,
):
self.node = node_to_int(node) if node else None
self.clock_seq = clock_seq_to_int(clock_seq) if clock_seq else None

def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
"""Transform"""
result = []
if len(inputs) != 0:
for collection in inputs:
Expand All @@ -282,6 +287,7 @@ class UUID1ToUUID6(TransformPlugin):
"""UUID1 to UUID6 Transform Plugin"""

def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
"""Transform"""
result = []
if len(inputs) != 0:
for collection in inputs:
Expand Down Expand Up @@ -312,6 +318,7 @@ class UUID7(TransformPlugin):
"""UUID7 Transform Plugin"""

def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
"""Transform"""
result = []
if len(inputs) != 0:
for collection in inputs:
Expand All @@ -335,6 +342,7 @@ class UUID8(TransformPlugin):
"""UUID8 Transform Plugin"""

def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
"""Transform"""
result = []
if len(inputs) != 0:
for collection in inputs:
Expand Down Expand Up @@ -374,13 +382,12 @@ def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
class UUIDConvert(TransformPlugin):
"""Converts UUID representation"""

def __init__(self, from_format=None, to_format=None):
def __init__(self, from_format: str | None, to_format: str | None) -> None:
self.from_ = from_format
self.to_ = to_format

def uuid_validate(self, test_uuid, uuid_string):
"""Warning if UUID string not standard (versions 1 to 8)
"""
def uuid_validate(self, test_uuid: uuid.UUID, uuid_string: str) -> None:
"""Warning if UUID string not standard (versions 1 to 8)"""
pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"

if not re.match(pattern, str(test_uuid)):
Expand All @@ -389,7 +396,7 @@ def uuid_validate(self, test_uuid, uuid_string):
f"the proposed updates"
)

def convert_uuid(self, uuid_string, result=None):
def convert_uuid(self, uuid_string: str) -> str: # noqa: PLR0912 C901
"""Convert UUID string"""
if self.from_ == "uuid_hex":
try:
Expand Down Expand Up @@ -427,6 +434,7 @@ def convert_uuid(self, uuid_string, result=None):
return result

def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
"""Trasnform"""
result = []
if len(inputs) != 0:
for collection in inputs:
Expand All @@ -446,6 +454,7 @@ class UUIDVersion(TransformPlugin):
"""Outputs UUID version number"""

def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
"""Transform"""
result = []
if len(inputs) != 0:
for collection in inputs:
Expand Down
31 changes: 14 additions & 17 deletions cmem_plugin_uuid/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,43 +49,40 @@
uuid_convert_param_out.allow_only_autocompleted_values = True


def node_to_int(node: str):
"""Convert a string representation of a node byte array to an integer.
E.g. 01:23:45:67:89:AB -> 1250999896491
"""
def node_to_int(node: str) -> int:
"""Convert a string representation of a node byte array to an integer"""
try:
byte_string = node.replace(":", "").replace("-", "")
byte_array = unhexlify(byte_string)
return int.from_bytes(byte_array, byteorder="big", signed=False)
except Exception as exc:
except ValueError as exc:
raise ValueError(f"node: {exc} ({node})") from exc


def clock_seq_to_int(clock_seq: str):
def clock_seq_to_int(clock_seq: str) -> int:
"""Convert a string representation of a clock_seq to an integer."""
try:
return int(clock_seq)
except Exception as exc:
except ValueError as exc:
raise ValueError(f"clock_seq: {exc} ({clock_seq})") from exc


def namespace_hex(value, uuid_version):
"""Return hex string from input value
"""
def namespace_hex(value: str, uuid_version: int) -> str | None:
"""Return hex string from input value"""
hex_value = None
if uuid_version == 3:
if uuid_version == 3: # noqa: PLR2004
hex_value = md5(value.encode(), usedforsecurity=False).hexdigest()
elif uuid_version == 5:
elif uuid_version == 5: # noqa: PLR2004
hex_value = sha1(value.encode(), usedforsecurity=False).hexdigest()[:32]
return hex_value


def get_namespace_uuid(
namespace_as_uuid=None,
namespace=None,
uuid_version=None,
):
"""Returns namespace UUID"""
namespace_as_uuid: bool | None,
namespace: str | None,
uuid_version: int | None,
) -> uuid.UUID | None:
"""Return namespace UUID"""
namespace_uuid = None

if namespace == "namespace_url":
Expand Down
Loading

0 comments on commit c342074

Please sign in to comment.