Git-Mirrors/veilid
1
0
Fork 0
mirror of https://gitlab.com/veilid/veilid.git synced 2024-10-01 01:26:08 -04:00
Code Issues Packages Projects Releases Wiki Activity

Cleanup types

Browse Source
This commit is contained in:
Teknique 2023-06-16 23:33:10 -07:00
parent 861ca706ab
commit fa92c9902d
No known key found for this signature in database
GPG Key ID: 54D9906A294B7FF3
1 changed files with 169 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

276
veilid-python/veilid/types.py
View File

@ -1,12 +1,11 @@
import time
import json
import base64 import base64
import json
from enum import StrEnum from enum import StrEnum
from typing import Self, Optional, Any, Tuple from typing import Any, Optional, Self, Tuple
#################################################################### ####################################################################
def urlsafe_b64encode_no_pad(b: bytes) -> str: def urlsafe_b64encode_no_pad(b: bytes) -> str:
""" """
Removes any `=` used as padding from the encoded string. Removes any `=` used as padding from the encoded string.
@ -22,6 +21,7 @@ def urlsafe_b64decode_no_pad(s: str) -> bytes:
s = s + ("=" * padding) s = s + ("=" * padding)
return base64.urlsafe_b64decode(s) return base64.urlsafe_b64decode(s)
class VeilidJSONEncoder(json.JSONEncoder): class VeilidJSONEncoder(json.JSONEncoder):
def default(self, o): def default(self, o):
if isinstance(o, bytes): if isinstance(o, bytes):
@ -32,170 +32,203 @@ class VeilidJSONEncoder(json.JSONEncoder):
@staticmethod @staticmethod
def dumps(req: Any, *args, **kwargs) -> str: def dumps(req: Any, *args, **kwargs) -> str:
return json.dumps(req, cls = VeilidJSONEncoder, *args, **kwargs) return json.dumps(req, cls=VeilidJSONEncoder, *args, **kwargs)
#################################################################### ####################################################################
class VeilidLogLevel(StrEnum): class VeilidLogLevel(StrEnum):
ERROR = 'Error' ERROR = "Error"
WARN = 'Warn' WARN = "Warn"
INFO = 'Info' INFO = "Info"
DEBUG = 'Debug' DEBUG = "Debug"
TRACE = 'Trace' TRACE = "Trace"
class CryptoKind(StrEnum): class CryptoKind(StrEnum):
CRYPTO_KIND_NONE = "NONE" CRYPTO_KIND_NONE = "NONE"
CRYPTO_KIND_VLD0 = "VLD0" CRYPTO_KIND_VLD0 = "VLD0"
class Stability(StrEnum): class Stability(StrEnum):
LOW_LATENCY = "LowLatency" LOW_LATENCY = "LowLatency"
RELIABLE = "Reliable" RELIABLE = "Reliable"
class Sequencing(StrEnum): class Sequencing(StrEnum):
NO_PREFERENCE = "NoPreference" NO_PREFERENCE = "NoPreference"
PREFER_ORDERED = "PreferOrdered" PREFER_ORDERED = "PreferOrdered"
ENSURE_ORDERED = "EnsureOrdered" ENSURE_ORDERED = "EnsureOrdered"
class DHTSchemaKind(StrEnum): class DHTSchemaKind(StrEnum):
DFLT = "DFLT" DFLT = "DFLT"
SMPL = "SMPL" SMPL = "SMPL"
#################################################################### ####################################################################
class Timestamp(int): class Timestamp(int):
pass pass
class TimestampDuration(int): class TimestampDuration(int):
pass pass
class ByteCount(int): class ByteCount(int):
pass pass
class OperationId(int): class OperationId(int):
pass pass
class RouteId(str): class RouteId(str):
pass pass
class CryptoKey:
class EncodedString(str):
def to_bytes(self) -> bytes: def to_bytes(self) -> bytes:
return urlsafe_b64decode_no_pad(self) return urlsafe_b64decode_no_pad(self)
class CryptoKeyDistance(CryptoKey, str): @classmethod
@staticmethod def from_bytes(cls, b: bytes) -> Self:
def from_bytes(b: bytes) -> Self: return cls(urlsafe_b64encode_no_pad(b))
return CryptoKeyDistance(urlsafe_b64encode_no_pad(b))
class PublicKey(CryptoKey, str):
@staticmethod
def from_bytes(b: bytes) -> Self:
return PublicKey(urlsafe_b64encode_no_pad(b))
class SecretKey(CryptoKey, str): class CryptoKey(EncodedString):
@staticmethod pass
def from_bytes(b: bytes) -> Self:
return SecretKey(urlsafe_b64encode_no_pad(b))
class SharedSecret(CryptoKey, str):
@staticmethod
def from_bytes(b: bytes) -> Self:
return SharedSecret(urlsafe_b64encode_no_pad(b))
class HashDigest(CryptoKey, str): class CryptoKeyDistance(CryptoKey):
@staticmethod pass
def from_bytes(b: bytes) -> Self:
return HashDigest(urlsafe_b64encode_no_pad(b))
class Signature(str):
@staticmethod
def from_bytes(b: bytes) -> Self:
return Signature(urlsafe_b64encode_no_pad(b))
def to_bytes(self) -> bytes:
return urlsafe_b64decode_no_pad(self)
class Nonce(str): class PublicKey(CryptoKey):
@staticmethod pass
def from_bytes(b: bytes) -> Self:
return Signature(urlsafe_b64encode_no_pad(b))
def to_bytes(self) -> bytes: class SecretKey(CryptoKey):
return urlsafe_b64decode_no_pad(self) pass
class SharedSecret(CryptoKey):
pass
class HashDigest(CryptoKey):
pass
class Signature(EncodedString):
pass
class Nonce(EncodedString):
pass
class KeyPair(str): class KeyPair(str):
@staticmethod @classmethod
def from_parts(key: PublicKey, secret: SecretKey) -> Self: def from_parts(cls, key: PublicKey, secret: SecretKey) -> Self:
return KeyPair(key + ":" + secret) return cls(f"{key}:{secret}")
def key(self) -> PublicKey:
return PublicKey(str.split(":", 1)[0])
def secret(self) -> SecretKey:
return SecretKey(str.split(":", 1)[1])
def to_parts(self) -> Tuple[PublicKey, SecretKey]:
parts = str.split(":", 1)
return (PublicKey(parts[0]), SecretKey(parts[1]))
class CryptoTyped: def key(self) -> PublicKey:
return PublicKey(self.split(":", 1)[0])
def secret(self) -> SecretKey:
return SecretKey(self.split(":", 1)[1])
def to_parts(self) -> Tuple[PublicKey, SecretKey]:
public, secret = self.split(":", 1)
return (PublicKey(public), SecretKey(secret))
class CryptoTyped(str):
def kind(self) -> CryptoKind: def kind(self) -> CryptoKind:
if self[4] != ':': if self[4] != ":":
raise ValueError("Not CryptoTyped") raise ValueError("Not CryptoTyped")
return CryptoKind(self[0:4]) return CryptoKind(self[0:4])
def _value(self) -> str: def _value(self) -> str:
if self[4] != ':': if self[4] != ":":
raise ValueError("Not CryptoTyped") raise ValueError("Not CryptoTyped")
return self[5:] return self[5:]
class TypedKey(CryptoTyped, str):
@staticmethod class TypedKey(CryptoTyped):
def from_value(kind: CryptoKind, value: PublicKey) -> Self: @classmethod
return TypedKey(kind + ":" + value) def from_value(cls, kind: CryptoKind, value: PublicKey) -> Self:
return cls(f"{kind}:{value}")
def value(self) -> PublicKey: def value(self) -> PublicKey:
PublicKey(self._value()) return PublicKey(self._value())
class TypedSecret(CryptoTyped):
@classmethod
def from_value(cls, kind: CryptoKind, value: SecretKey) -> Self:
return cls(f"{kind}:{value}")
class TypedSecret(CryptoTyped, str):
@staticmethod
def from_value(kind: CryptoKind, value: SecretKey) -> Self:
return TypedSecret(kind + ":" + value)
def value(self) -> SecretKey: def value(self) -> SecretKey:
SecretKey(self._value()) return SecretKey(self._value())
class TypedKeyPair(CryptoTyped):
@classmethod
def from_value(cls, kind: CryptoKind, value: KeyPair) -> Self:
return cls(f"{kind}:{value}")
class TypedKeyPair(CryptoTyped, str):
@staticmethod
def from_value(kind: CryptoKind, value: KeyPair) -> Self:
return TypedKeyPair(kind + ":" + value)
def value(self) -> KeyPair: def value(self) -> KeyPair:
KeyPair(self._value()) return KeyPair(self._value())
class TypedSignature(CryptoTyped):
@classmethod
def from_value(cls, kind: CryptoKind, value: Signature) -> Self:
return cls(f"{kind}:{value}")
class TypedSignature(CryptoTyped, str):
@staticmethod
def from_value(kind: CryptoKind, value: Signature) -> Self:
return TypedSignature(kind + ":" + value)
def value(self) -> Signature: def value(self) -> Signature:
Signature(self._value()) return Signature(self._value())
class ValueSubkey(int): class ValueSubkey(int):
pass pass
class ValueSeqNum(int): class ValueSeqNum(int):
pass pass
#################################################################### ####################################################################
class VeilidVersion: class VeilidVersion:
_major: int _major: int
_minor: int _minor: int
_patch: int _patch: int
def __init__(self, major: int, minor: int, patch: int): def __init__(self, major: int, minor: int, patch: int):
self._major = major self._major = major
self._minor = minor self._minor = minor
self._patch = patch self._patch = patch
@property @property
def major(self): def major(self):
return self._major return self._major
@property @property
def minor(self): def minor(self):
return self._minor return self._minor
@property @property
def patch(self): def patch(self):
return self._patch return self._patch
class NewPrivateRouteResult: class NewPrivateRouteResult:
route_id: RouteId route_id: RouteId
blob: bytes blob: bytes
@ -207,26 +240,27 @@ class NewPrivateRouteResult:
def to_tuple(self) -> Tuple[RouteId, bytes]: def to_tuple(self) -> Tuple[RouteId, bytes]:
return (self.route_id, self.blob) return (self.route_id, self.blob)
@staticmethod @classmethod
def from_json(j: dict) -> Self: def from_json(cls, j: dict) -> Self:
return NewPrivateRouteResult( return cls(RouteId(j["route_id"]), urlsafe_b64decode_no_pad(j["blob"]))
RouteId(j['route_id']),
urlsafe_b64decode_no_pad(j['blob']))
class DHTSchemaSMPLMember: class DHTSchemaSMPLMember:
m_key: PublicKey m_key: PublicKey
m_cnt: int m_cnt: int
def __init__(self, m_key: PublicKey, m_cnt: int): def __init__(self, m_key: PublicKey, m_cnt: int):
self.m_key = m_key self.m_key = m_key
self.m_cnt = m_cnt self.m_cnt = m_cnt
@staticmethod
def from_json(j: dict) -> Self: @classmethod
return DHTSchemaSMPLMember( def from_json(cls, j: dict) -> Self:
PublicKey(j['m_key']), return cls(PublicKey(j["m_key"]), j["m_cnt"])
j['m_cnt'])
def to_json(self) -> dict: def to_json(self) -> dict:
return self.__dict__ return self.__dict__
class DHTSchema: class DHTSchema:
kind: DHTSchemaKind kind: DHTSchemaKind
@ -235,50 +269,60 @@ class DHTSchema:
for k, v in kwargs.items(): for k, v in kwargs.items():
setattr(self, k, v) setattr(self, k, v)
@staticmethod @classmethod
def dflt(o_cnt: int) -> Self: def dflt(cls, o_cnt: int) -> Self:
Self(DHTSchemaKind.DFLT, o_cnt = o_cnt) return cls(DHTSchemaKind.DFLT, o_cnt=o_cnt)
@staticmethod @classmethod
def smpl(o_cnt: int, members: list[DHTSchemaSMPLMember]) -> Self: def smpl(cls, o_cnt: int, members: list[DHTSchemaSMPLMember]) -> Self:
Self(DHTSchemaKind.SMPL, o_cnt = o_cnt, members = members) return cls(DHTSchemaKind.SMPL, o_cnt=o_cnt, members=members)
@staticmethod @classmethod
def from_json(j: dict) -> Self: def from_json(cls, j: dict) -> Self:
if DHTSchemaKind(j['kind']) == DHTSchemaKind.DFLT: if DHTSchemaKind(j["kind"]) == DHTSchemaKind.DFLT:
return DHTSchema.dflt(j['o_cnt']) return cls.dflt(j["o_cnt"])
if DHTSchemaKind(j['kind']) == DHTSchemaKind.SMPL: if DHTSchemaKind(j["kind"]) == DHTSchemaKind.SMPL:
return DHTSchema.smpl( return cls.smpl(
j['o_cnt'], j["o_cnt"],
list(map(lambda x: DHTSchemaSMPLMember.from_json(x), j['members']))) [DHTSchemaSMPLMember.from_json(member) for member in j["members"]],
raise Exception("Unknown DHTSchema kind", j['kind']) )
raise Exception("Unknown DHTSchema kind", j["kind"])
def to_json(self) -> dict: def to_json(self) -> dict:
return self.__dict__ return self.__dict__
class DHTRecordDescriptor: class DHTRecordDescriptor:
key: TypedKey key: TypedKey
owner: PublicKey owner: PublicKey
owner_secret: Optional[SecretKey] owner_secret: Optional[SecretKey]
schema: DHTSchema schema: DHTSchema
def __init__(self, key: TypedKey, owner: PublicKey, owner_secret: Optional[SecretKey], schema: DHTSchema): def __init__(
self,
key: TypedKey,
owner: PublicKey,
owner_secret: Optional[SecretKey],
schema: DHTSchema,
):
self.key = key self.key = key
self.owner = owner self.owner = owner
self.owner_secret = owner_secret self.owner_secret = owner_secret
self.schema = schema self.schema = schema
@staticmethod @classmethod
def from_json(j: dict) -> Self: def from_json(cls, j: dict) -> Self:
DHTRecordDescriptor( return cls(
TypedKey(j['key']), TypedKey(j["key"]),
PublicKey(j['owner']), PublicKey(j["owner"]),
None if j['owner_secret'] is None else SecretKey(j['owner_secret']), None if j["owner_secret"] is None else SecretKey(j["owner_secret"]),
DHTSchema.from_json(j['schema'])) DHTSchema.from_json(j["schema"]),
)
def to_json(self) -> dict: def to_json(self) -> dict:
return self.__dict__ return self.__dict__
class ValueData: class ValueData:
seq: ValueSeqNum seq: ValueSeqNum
data: bytes data: bytes
@ -289,13 +333,13 @@ class ValueData:
self.data = data self.data = data
self.writer = writer self.writer = writer
@staticmethod @classmethod
def from_json(j: dict) -> Self: def from_json(cls, j: dict) -> Self:
DHTRecordDescriptor( return cls(
ValueSeqNum(j['seq']), ValueSeqNum(j["seq"]),
urlsafe_b64decode_no_pad(j['data']), urlsafe_b64decode_no_pad(j["data"]),
PublicKey(j['writer'])) PublicKey(j["writer"]),
)
def to_json(self) -> dict: def to_json(self) -> dict:
return self.__dict__ return self.__dict__