Mirroristas/planetmint
2
0
Fork 0
mirror of https://github.com/planetmint/planetmint.git synced 2025-11-24 22:45:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

added type hints to transactions module

Browse Source 
Signed-off-by: Lorenz Herzberger <lorenzherzberger@gmail.com>
This commit is contained in:
Lorenz Herzberger 2022-09-27 16:46:44 +02:00
parent cddfe862ef
commit a7d302e94d
No known key found for this signature in database
GPG Key ID: FA5EE906EB55316A
10 changed files with 90 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
planetmint/transactions/common/crypto.py
View File

@ -17,7 +17,7 @@ def hash_data(data):
return sha3_256(data.encode()).hexdigest()
def generate_key_pair():
def generate_key_pair() -> CryptoKeypair:
"""Generates a cryptographic key pair.
Returns:

7
planetmint/transactions/common/input.py
View File

@ -3,6 +3,7 @@
# SPDX-License-Identifier: (Apache-2.0 AND CC-BY-4.0)
# Code is Apache-2.0 and docs are CC-BY-4.0
from typing import Optional
from cryptoconditions import Fulfillment
from cryptoconditions.exceptions import ASN1DecodeError, ASN1EncodeError
@ -27,7 +28,7 @@ class Input(object):
Transaction.
"""
def __init__(self, fulfillment, owners_before, fulfills=None):
def __init__(self, fulfillment: Fulfillment, owners_before: list[str], fulfills: Optional[TransactionLink] = None):
"""Create an instance of an :class:`~.Input`.
Args:
@ -86,7 +87,7 @@ class Input(object):
return input_
@classmethod
def generate(cls, public_keys):
def generate(cls, public_keys: list[str]):
# TODO: write docstring
# The amount here does not really matter. It is only use on the
# output data model but here we only care about the fulfillment
@ -94,7 +95,7 @@ class Input(object):
return cls(output.fulfillment, public_keys)
@classmethod
def from_dict(cls, data):
def from_dict(cls, data: dict):
"""Transforms a Python dictionary to an Input object.
Note:

8
planetmint/transactions/common/memoize.py
View File

@ -1,7 +1,7 @@
import functools
import codecs
from functools import lru_cache
from typing import Callable
class HDict(dict):
def __hash__(self):
@ -9,11 +9,11 @@ class HDict(dict):
@lru_cache(maxsize=16384)
def from_dict(func, *args, **kwargs):
def from_dict(func: Callable, *args, **kwargs):
return func(*args, **kwargs)
def memoize_from_dict(func):
def memoize_from_dict(func: Callable):
@functools.wraps(func)
def memoized_func(*args, **kwargs):
if args[1] is None:
@ -45,7 +45,7 @@ def to_dict(func, tx_wrapped):
return func(tx_wrapped.tx)
def memoize_to_dict(func):
def memoize_to_dict(func: Callable):
@functools.wraps(func)
def memoized_func(*args, **kwargs):

12
planetmint/transactions/common/output.py
View File

@ -4,6 +4,7 @@
# Code is Apache-2.0 and docs are CC-BY-4.0
from functools import reduce
from typing import Union, Optional
import base58
from cryptoconditions import ThresholdSha256, Ed25519Sha256, ZenroomSha256
@ -27,7 +28,7 @@ class Output(object):
MAX_AMOUNT = 9 * 10**18
def __init__(self, fulfillment, public_keys=None, amount=1):
def __init__(self, fulfillment: type[Fulfillment], public_keys: Optional[list[str]] = None, amount: int = 1):
"""Create an instance of a :class:`~.Output`.
Args:
@ -90,7 +91,7 @@ class Output(object):
return output
@classmethod
def generate(cls, public_keys, amount):
def generate(cls, public_keys: list[str], amount: int):
"""Generates a Output from a specifically formed tuple or list.
Note:
@ -136,7 +137,7 @@ class Output(object):
return cls(threshold_cond, public_keys, amount=amount)
@classmethod
def _gen_condition(cls, initial, new_public_keys):
def _gen_condition(cls, initial: type[ThresholdSha256], new_public_keys: Union[list[str],str]) -> type[ThresholdSha256]:
"""Generates ThresholdSha256 conditions from a list of new owners.
Note:
@ -165,7 +166,8 @@ class Output(object):
raise ValueError("Sublist cannot contain single owner")
else:
try:
new_public_keys = new_public_keys.pop()
if isinstance(new_public_keys, list):
new_public_keys = new_public_keys.pop()
except AttributeError:
pass
# NOTE: Instead of submitting base58 encoded addresses, a user
@ -182,7 +184,7 @@ class Output(object):
return initial
@classmethod
def from_dict(cls, data):
def from_dict(cls, data: dict):
"""Transforms a Python dictionary to an Output object.
Note:

81
planetmint/transactions/common/transaction.py
View File

@ -14,6 +14,7 @@ Attributes:
from collections import namedtuple
from copy import deepcopy
from functools import lru_cache
from typing import Optional
import rapidjson
import base58
@ -83,16 +84,16 @@ class Transaction(object):
version (string): Defines the version number of a Transaction.
"""
CREATE = "CREATE"
TRANSFER = "TRANSFER"
VALIDATOR_ELECTION = VALIDATOR_ELECTION
CHAIN_MIGRATION_ELECTION = CHAIN_MIGRATION_ELECTION
VOTE = VOTE
ALLOWED_OPERATIONS = (CREATE, TRANSFER)
ASSET = "asset"
METADATA = "metadata"
DATA = "data"
VERSION = "2.0"
CREATE: str = "CREATE"
TRANSFER: str = "TRANSFER"
VALIDATOR_ELECTION: str = VALIDATOR_ELECTION
CHAIN_MIGRATION_ELECTION: str = CHAIN_MIGRATION_ELECTION
VOTE: str = VOTE
ALLOWED_OPERATIONS: tuple[str, ...] = (CREATE, TRANSFER)
ASSET: str = "asset"
METADATA: str = "metadata"
DATA: str = "data"
VERSION: str = "2.0"
def __init__(
self,
@ -250,7 +251,7 @@ class Transaction(object):
return False
return self.to_dict() == other
def to_inputs(self, indices=None):
def to_inputs(self, indices: Optional[list[int]] = None) -> list[Input]:
"""Converts a Transaction's outputs to spendable inputs.
Note:
@ -272,17 +273,17 @@ class Transaction(object):
"""
# NOTE: If no indices are passed, we just assume to take all outputs
# as inputs.
indices = indices or range(len(self.outputs))
iterable_indices = indices or range(len(self.outputs))
return [
Input(
self.outputs[idx].fulfillment,
self.outputs[idx].public_keys,
TransactionLink(self.id, idx),
)
for idx in indices
for idx in iterable_indices
]
def add_input(self, input_):
def add_input(self, input_: Input) -> None:
"""Adds an input to a Transaction's list of inputs.
Args:
@ -293,7 +294,7 @@ class Transaction(object):
raise TypeError("`input_` must be a Input instance")
self.inputs.append(input_)
def add_output(self, output):
def add_output(self, output: Output) -> None:
"""Adds an output to a Transaction's list of outputs.
Args:
@ -305,7 +306,7 @@ class Transaction(object):
raise TypeError("`output` must be an Output instance or None")
self.outputs.append(output)
def sign(self, private_keys):
def sign(self, private_keys: list[str]):
"""Fulfills a previous Transaction's Output by signing Inputs.
Note:
@ -359,7 +360,7 @@ class Transaction(object):
return self
@classmethod
def _sign_input(cls, input_, message, key_pairs):
def _sign_input(cls, input_: Input, message: str, key_pairs: dict) -> Input:
"""Signs a single Input.
Note:
@ -384,7 +385,7 @@ class Transaction(object):
raise ValueError("Fulfillment couldn't be matched to " "Cryptocondition fulfillment type.")
@classmethod
def _sign_zenroom_fulfillment(cls, input_, message, key_pairs):
def _sign_zenroom_fulfillment(cls, input_: Input, message: str, key_pairs: dict) -> Input:
"""Signs a Zenroomful.
Args:
@ -399,14 +400,14 @@ class Transaction(object):
# this should never happen, but then again, never say never.
input_ = deepcopy(input_)
public_key = input_.owners_before[0]
message = sha3_256(message.encode())
sha3_message = sha3_256(message.encode())
if input_.fulfills:
message.update("{}{}".format(input_.fulfills.txid, input_.fulfills.output).encode())
sha3_message.update("{}{}".format(input_.fulfills.txid, input_.fulfills.output).encode())
try:
# cryptoconditions makes no assumptions of the encoding of the
# message to sign or verify. It only accepts bytestrings
input_.fulfillment.sign(message.digest(), base58.b58decode(key_pairs[public_key].encode()))
input_.fulfillment.sign(sha3_message.digest(), base58.b58decode(key_pairs[public_key].encode()))
except KeyError:
raise KeypairMismatchException(
"Public key {} is not a pair to " "any of the private keys".format(public_key)
@ -414,7 +415,7 @@ class Transaction(object):
return input_
@classmethod
def _sign_simple_signature_fulfillment(cls, input_, message, key_pairs):
def _sign_simple_signature_fulfillment(cls, input_: Input, message: str, key_pairs: dict) -> Input:
"""Signs a Ed25519Fulfillment.
Args:
@ -429,14 +430,14 @@ class Transaction(object):
# this should never happen, but then again, never say never.
input_ = deepcopy(input_)
public_key = input_.owners_before[0]
message = sha3_256(message.encode())
sha3_message = sha3_256(message.encode())
if input_.fulfills:
message.update("{}{}".format(input_.fulfills.txid, input_.fulfills.output).encode())
sha3_message.update("{}{}".format(input_.fulfills.txid, input_.fulfills.output).encode())
try:
# cryptoconditions makes no assumptions of the encoding of the
# message to sign or verify. It only accepts bytestrings
input_.fulfillment.sign(message.digest(), base58.b58decode(key_pairs[public_key].encode()))
input_.fulfillment.sign(sha3_message.digest(), base58.b58decode(key_pairs[public_key].encode()))
except KeyError:
raise KeypairMismatchException(
"Public key {} is not a pair to " "any of the private keys".format(public_key)
@ -444,7 +445,7 @@ class Transaction(object):
return input_
@classmethod
def _sign_threshold_signature_fulfillment(cls, input_, message, key_pairs):
def _sign_threshold_signature_fulfillment(cls, input_: Input, message: str, key_pairs: dict) -> Input:
"""Signs a ThresholdSha256.
Args:
@ -454,9 +455,9 @@ class Transaction(object):
key_pairs (dict): The keys to sign the Transaction with.
"""
input_ = deepcopy(input_)
message = sha3_256(message.encode())
sha3_message = sha3_256(message.encode())
if input_.fulfills:
message.update("{}{}".format(input_.fulfills.txid, input_.fulfills.output).encode())
sha3_message.update("{}{}".format(input_.fulfills.txid, input_.fulfills.output).encode())
for owner_before in set(input_.owners_before):
# TODO: CC should throw a KeypairMismatchException, instead of
@ -484,10 +485,10 @@ class Transaction(object):
# cryptoconditions makes no assumptions of the encoding of the
# message to sign or verify. It only accepts bytestrings
for subffill in subffills:
subffill.sign(message.digest(), base58.b58decode(private_key.encode()))
subffill.sign(sha3_message.digest(), base58.b58decode(private_key.encode()))
return input_
def inputs_valid(self, outputs=None):
def inputs_valid(self, outputs=None) -> bool:
"""Validates the Inputs in the Transaction against given
Outputs.
@ -520,7 +521,7 @@ class Transaction(object):
allowed_ops = ", ".join(self.__class__.ALLOWED_OPERATIONS)
raise TypeError("`operation` must be one of {}".format(allowed_ops))
def _inputs_valid(self, output_condition_uris):
def _inputs_valid(self, output_condition_uris: list[str]) -> bool:
"""Validates an Input against a given set of Outputs.
Note:
@ -550,7 +551,7 @@ class Transaction(object):
return all(validate(i, cond) for i, cond in enumerate(output_condition_uris))
@lru_cache(maxsize=16384)
def _input_valid(self, input_, operation, message, output_condition_uri=None):
def _input_valid(self, input_: Input, operation: str, message: str, output_condition_uri: Optional[str] = None) -> bool:
"""Validates a single Input against a single Output.
Note:
@ -601,16 +602,16 @@ class Transaction(object):
msg = json.loads(message)
ffill_valid = parsed_ffill.validate(message=json.dumps(msg["script"]))
else:
message = sha3_256(message.encode())
sha3_message = sha3_256(message.encode())
if input_.fulfills:
message.update("{}{}".format(input_.fulfills.txid, input_.fulfills.output).encode())
sha3_message.update("{}{}".format(input_.fulfills.txid, input_.fulfills.output).encode())
# NOTE: We pass a timestamp to `.validate`, as in case of a timeout
# condition we'll have to validate against it
# cryptoconditions makes no assumptions of the encoding of the
# message to sign or verify. It only accepts bytestrings
ffill_valid = parsed_ffill.validate(message=message.digest())
ffill_valid = parsed_ffill.validate(message=sha3_message.digest())
return output_valid and ffill_valid
# This function is required by `lru_cache` to create a key for memoization
@ -618,7 +619,7 @@ class Transaction(object):
return hash(self.id)
@memoize_to_dict
def to_dict(self):
def to_dict(self) -> dict:
"""Transforms the object to a Python dictionary.
Returns:
@ -639,7 +640,7 @@ class Transaction(object):
@staticmethod
# TODO: Remove `_dict` prefix of variable.
def _remove_signatures(tx_dict):
def _remove_signatures(tx_dict: dict) -> dict:
"""Takes a Transaction dictionary and removes all signatures.
Args:
@ -716,7 +717,7 @@ class Transaction(object):
return asset_ids.pop()
@staticmethod
def validate_id(tx_body):
def validate_id(tx_body: dict):
"""Validate the transaction ID of a transaction
Args:
@ -741,7 +742,7 @@ class Transaction(object):
@classmethod
@memoize_from_dict
def from_dict(cls, tx, skip_schema_validation=True):
def from_dict(cls, tx: dict, skip_schema_validation=True):
"""Transforms a Python dictionary to a Transaction object.
Args:
@ -848,7 +849,7 @@ class Transaction(object):
tx = list(tx_map.values())[0]
return cls.from_dict(tx)
type_registry = {}
type_registry: dict[type, type] = {}
@staticmethod
def register_type(tx_type, tx_class):

11
planetmint/transactions/common/transaction_link.py
View File

@ -4,6 +4,9 @@
# Code is Apache-2.0 and docs are CC-BY-4.0
from typing import Optional, Union
class TransactionLink(object):
"""An object for unidirectional linking to a Transaction's Output.
@ -13,7 +16,7 @@ class TransactionLink(object):
`txid`.
"""
def __init__(self, txid=None, output=None):
def __init__(self, txid: Optional[str] = None, output: Optional[int] = None):
"""Create an instance of a :class:`~.TransactionLink`.
Note:
@ -42,7 +45,7 @@ class TransactionLink(object):
return hash((self.txid, self.output))
@classmethod
def from_dict(cls, link):
def from_dict(cls, link: dict):
"""Transforms a Python dictionary to a TransactionLink object.
Args:
@ -56,7 +59,7 @@ class TransactionLink(object):
except TypeError:
return cls()
def to_dict(self):
def to_dict(self) -> Union[dict,None]:
"""Transforms the object to a Python dictionary.
Returns:
@ -70,7 +73,7 @@ class TransactionLink(object):
"output_index": self.output,
}
def to_uri(self, path=""):
def to_uri(self, path: str = "") -> Union[str,None]:
if self.txid is None and self.output is None:
return None
return "{}/transactions/{}/outputs/{}".format(path, self.txid, self.output)

25
planetmint/transactions/common/utils.py
View File

@ -7,15 +7,16 @@ import base58
import time
import re
import rapidjson
from typing import Callable
from planetmint.config import Config
from planetmint.transactions.common.exceptions import ValidationError
from cryptoconditions import ThresholdSha256, Ed25519Sha256, ZenroomSha256
from cryptoconditions import ThresholdSha256, Ed25519Sha256, ZenroomSha256, Fulfillment
from planetmint.transactions.common.exceptions import ThresholdTooDeep
from cryptoconditions.exceptions import UnsupportedTypeError
def gen_timestamp():
def gen_timestamp() -> str:
"""The Unix time, rounded to the nearest second.
See https://en.wikipedia.org/wiki/Unix_time
@ -25,7 +26,7 @@ def gen_timestamp():
return str(round(time.time()))
def serialize(data):
def serialize(data: dict) -> str:
"""Serialize a dict into a JSON formatted string.
This function enforces rules like the separator and order of keys.
@ -46,7 +47,7 @@ def serialize(data):
return rapidjson.dumps(data, skipkeys=False, ensure_ascii=False, sort_keys=True)
def deserialize(data):
def deserialize(data: str) -> dict:
"""Deserialize a JSON formatted string into a dict.
Args:
@ -59,7 +60,7 @@ def deserialize(data):
return rapidjson.loads(data)
def validate_txn_obj(obj_name, obj, key, validation_fun):
def validate_txn_obj(obj_name: str, obj: dict, key: str, validation_fun: Callable) -> None:
"""Validate value of `key` in `obj` using `validation_fun`.
Args:
@ -85,7 +86,7 @@ def validate_txn_obj(obj_name, obj, key, validation_fun):
validate_all_items_in_list(obj_name, data, validation_fun)
def validate_all_items_in_list(obj_name, data, validation_fun):
def validate_all_items_in_list(obj_name: str, data: list, validation_fun: Callable) -> None:
for item in data:
if isinstance(item, dict):
validate_all_keys_in_obj(obj_name, item, validation_fun)
@ -93,7 +94,7 @@ def validate_all_items_in_list(obj_name, data, validation_fun):
validate_all_items_in_list(obj_name, item, validation_fun)
def validate_all_keys_in_obj(obj_name, obj, validation_fun):
def validate_all_keys_in_obj(obj_name: str, obj: dict, validation_fun: Callable) -> None:
"""Validate all (nested) keys in `obj` by using `validation_fun`.
Args:
@ -116,7 +117,7 @@ def validate_all_keys_in_obj(obj_name, obj, validation_fun):
validate_all_items_in_list(obj_name, value, validation_fun)
def validate_all_values_for_key_in_obj(obj, key, validation_fun):
def validate_all_values_for_key_in_obj(obj: dict, key: str, validation_fun: Callable) -> None:
"""Validate value for all (nested) occurrence of `key` in `obj`
using `validation_fun`.
@ -138,7 +139,7 @@ def validate_all_values_for_key_in_obj(obj, key, validation_fun):
validate_all_values_for_key_in_list(value, key, validation_fun)
def validate_all_values_for_key_in_list(input_list, key, validation_fun):
def validate_all_values_for_key_in_list(input_list: list, key: str, validation_fun: Callable) -> None:
for item in input_list:
if isinstance(item, dict):
validate_all_values_for_key_in_obj(item, key, validation_fun)
@ -146,7 +147,7 @@ def validate_all_values_for_key_in_list(input_list, key, validation_fun):
validate_all_values_for_key_in_list(item, key, validation_fun)
def validate_key(obj_name, key):
def validate_key(obj_name: str, key: str) -> None:
"""Check if `key` contains ".", "$" or null characters.
https://docs.mongodb.com/manual/reference/limits/#Restrictions-on-Field-Names
@ -170,7 +171,7 @@ def validate_key(obj_name, key):
raise ValidationError(error_str)
def _fulfillment_to_details(fulfillment):
def _fulfillment_to_details(fulfillment: type[Fulfillment]) -> dict:
"""Encode a fulfillment as a details dictionary
Args:
@ -201,7 +202,7 @@ def _fulfillment_to_details(fulfillment):
raise UnsupportedTypeError(fulfillment.type_name)
def _fulfillment_from_details(data, _depth=0):
def _fulfillment_from_details(data: dict, _depth: int = 0):
"""Load a fulfillment for a signing spec dictionary
Args:

7
planetmint/transactions/types/assets/create.py
View File

@ -3,11 +3,10 @@
# SPDX-License-Identifier: (Apache-2.0 AND CC-BY-4.0)
# Code is Apache-2.0 and docs are CC-BY-4.0
from typing import Optional
from cid import is_cid
from planetmint.transactions.common.transaction import Transaction
from planetmint.transactions.common.input import Input
from planetmint.transactions.common.output import Output
class Create(Transaction):
@ -16,7 +15,7 @@ class Create(Transaction):
ALLOWED_OPERATIONS = (OPERATION,)
@classmethod
def validate_create(self, tx_signers, recipients, asset, metadata):
def validate_create(self, tx_signers: list[str], recipients: list[tuple[list[str],int]], asset: Optional[dict], metadata: Optional[dict]):
if not isinstance(tx_signers, list):
raise TypeError("`tx_signers` must be a list instance")
if not isinstance(recipients, list):
@ -36,7 +35,7 @@ class Create(Transaction):
return True
@classmethod
def generate(cls, tx_signers, recipients, metadata=None, asset=None):
def generate(cls, tx_signers: list[str], recipients: list[tuple[list[str],int]], metadata: Optional[dict] = None, asset: Optional[dict] = None):
"""A simple way to generate a `CREATE` transaction.
Note:

7
planetmint/transactions/types/assets/transfer.py
View File

@ -3,7 +3,10 @@
# SPDX-License-Identifier: (Apache-2.0 AND CC-BY-4.0)
# Code is Apache-2.0 and docs are CC-BY-4.0
from typing import Optional
from planetmint.transactions.common.transaction import Transaction
from planetmint.transactions.common.input import Input
from planetmint.transactions.common.output import Output
from copy import deepcopy
@ -14,7 +17,7 @@ class Transfer(Transaction):
ALLOWED_OPERATIONS = (OPERATION,)
@classmethod
def validate_transfer(cls, inputs, recipients, asset_id, metadata):
def validate_transfer(cls, inputs: list[Input], recipients: list[tuple[list[str],int]], asset_id: str, metadata: Optional[dict]):
if not isinstance(inputs, list):
raise TypeError("`inputs` must be a list instance")
if len(inputs) == 0:
@ -39,7 +42,7 @@ class Transfer(Transaction):
return (deepcopy(inputs), outputs)
@classmethod
def generate(cls, inputs, recipients, asset_id, metadata=None):
def generate(cls, inputs: list[Input], recipients: list[tuple[list[str],int]], asset_id: str, metadata: Optional[dict] = None):
"""A simple way to generate a `TRANSFER` transaction.
Note:

9
planetmint/transactions/types/elections/election.py
View File

@ -6,6 +6,7 @@ from collections import OrderedDict
import base58
from uuid import uuid4
from typing import Optional
from planetmint import backend
from planetmint.transactions.types.assets.create import Create
@ -32,13 +33,13 @@ class Election(Transaction):
set to (OPERATION,), CREATE set to OPERATION.
"""
OPERATION = None
OPERATION: Optional[str] = None
# Custom validation schema
TX_SCHEMA_CUSTOM = None
# Election Statuses:
ONGOING = "ongoing"
CONCLUDED = "concluded"
INCONCLUSIVE = "inconclusive"
ONGOING: str = "ongoing"
CONCLUDED: str = "concluded"
INCONCLUSIVE: str = "inconclusive"
# Vote ratio to approve an election
ELECTION_THRESHOLD = 2 / 3