feat(nano): implement fundamental nano structures [part 2]

hathor/nanocontracts/__init__.py

@@ -0,0 +1,24 @@
+# Copyright 2023 Hathor Labs
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from hathor.nanocontracts.context import Context
+from hathor.nanocontracts.exception import NCFail
+from hathor.nanocontracts.types import public, view
+
+__all__ = [
+    'Context',
+    'NCFail',
+    'public',
+    'view',
+]

hathor/nanocontracts/blueprint_syntax_validation.py

@@ -0,0 +1,99 @@
+#  Copyright 2025 Hathor Labs
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from __future__ import annotations
+
+import inspect
+from typing import Callable
+
+from hathor.nanocontracts.exception import BlueprintSyntaxError
+
+
+def validate_has_self_arg(fn: Callable, annotation_name: str) -> None:
+    """Validate the `self` arg of a callable."""
+    arg_spec = inspect.getfullargspec(fn)
+    if len(arg_spec.args) == 0:
+        raise BlueprintSyntaxError(f'@{annotation_name} method must have `self` argument: `{fn.__name__}()`')
+
+    if arg_spec.args[0] != 'self':
+        raise BlueprintSyntaxError(
+            f'@{annotation_name} method first argument must be called `self`: `{fn.__name__}()`'
+        )
+
+    if 'self' in arg_spec.annotations.keys():
+        raise BlueprintSyntaxError(f'@{annotation_name} method `self` argument must not be typed: `{fn.__name__}()`')
+
+
+def validate_method_types(fn: Callable) -> None:
+    """Validate the arg and return types of a callable."""
+    special_args = ['self']
+    arg_spec = inspect.getfullargspec(fn)
+
+    if 'return' not in arg_spec.annotations:
+        raise BlueprintSyntaxError(f'missing return type on method `{fn.__name__}`')
+
+    # TODO: This currently fails for types such as unions, probably because this is the wrong
+    #  parsing function to use. Fix this.
+    # from hathor.nanocontracts.fields import get_field_class_for_attr
+    # return_type = arg_spec.annotations['return']
+    # if return_type is not None:
+    #     try:
+    #         get_field_class_for_attr(return_type)
+    #     except UnknownFieldType:
+    #         raise BlueprintSyntaxError(
+    #             f'unsupported return type `{return_type}` on method `{fn.__name__}`'
+    #         )
+
+    for arg_name in arg_spec.args:
+        if arg_name in special_args:
+            continue
+
+        if arg_name not in arg_spec.annotations:
+            raise BlueprintSyntaxError(f'argument `{arg_name}` on method `{fn.__name__}` must be typed')
+
+        # TODO: This currently fails for @view methods with NamedTuple as args for example,
+        #  because API calls use a different parsing function. Fix this.
+        # arg_type = arg_spec.annotations[arg_name]
+        # try:
+        #     get_field_class_for_attr(arg_type)
+        # except UnknownFieldType:
+        #     raise BlueprintSyntaxError(
+        #         f'unsupported type `{arg_type.__name__}` on argument `{arg_name}` of method `{fn.__name__}`'
+        #     )
+
+
+def validate_has_ctx_arg(fn: Callable, annotation_name: str) -> None:
+    """Validate the context arg of a callable."""
+    arg_spec = inspect.getfullargspec(fn)
+
+    if len(arg_spec.args) < 2:
+        raise BlueprintSyntaxError(
+            f'@{annotation_name} method must have `Context` argument: `{fn.__name__}()`'
+        )
+
+    from hathor.nanocontracts import Context
+    second_arg = arg_spec.args[1]
+    if arg_spec.annotations[second_arg] is not Context:
+        raise BlueprintSyntaxError(
+            f'@{annotation_name} method second arg `{second_arg}` argument must be of type `Context`: '
+            f'`{fn.__name__}()`'
+        )
+
+
+def validate_has_not_ctx_arg(fn: Callable, annotation_name: str) -> None:
+    """Validate that a callable doesn't have a `Context` arg."""
+    from hathor.nanocontracts import Context
+    arg_spec = inspect.getfullargspec(fn)
+    if Context in arg_spec.annotations.values():
+        raise BlueprintSyntaxError(f'@{annotation_name} method cannot have arg with type `Context`: `{fn.__name__}()`')

hathor/nanocontracts/context.py

@@ -0,0 +1,113 @@
+# Copyright 2023 Hathor Labs
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+from itertools import chain
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Any, final
+
+from hathor.crypto.util import get_address_b58_from_bytes
+from hathor.nanocontracts.exception import NCFail
+from hathor.nanocontracts.types import Address, ContractId, NCAction, TokenUid
+from hathor.nanocontracts.vertex_data import VertexData
+
+if TYPE_CHECKING:
+    from hathor.transaction import BaseTransaction
+
+_EMPTY_MAP: MappingProxyType[TokenUid, tuple[NCAction, ...]] = MappingProxyType({})
+
+
+@final
+class Context:
+    """Context passed to a method call. An empty list of actions means the
+    method is being called with no deposits and withdrawals.
+
+    Deposits and withdrawals are grouped by token. Note that it is impossible
+    to have both a deposit and a withdrawal for the same token.
+    """
+    __slots__ = ('__actions', '__address', '__vertex', '__timestamp', '__all_actions__')
+    __actions: MappingProxyType[TokenUid, tuple[NCAction, ...]]
+    __address: Address | ContractId
+    __vertex: VertexData
+    __timestamp: int
+
+    def __init__(
+        self,
+        actions: list[NCAction],
+        vertex: BaseTransaction | VertexData,
+        address: Address | ContractId,
+        timestamp: int,
+    ) -> None:
+        # Dict of action where the key is the token_uid.
+        # If empty, it is a method call without any actions.
+        if not actions:
+            self.__actions = _EMPTY_MAP
+        else:
+            raise NotImplementedError('temporarily removed during nano merge')
+
+        self.__all_actions__: tuple[NCAction, ...] = tuple(chain(*self.__actions.values()))
+
+        # Vertex calling the method.
+        if isinstance(vertex, VertexData):
+            self.__vertex = vertex
+        else:
+            self.__vertex = VertexData.create_from_vertex(vertex)
+
+        # Address calling the method.
+        self.__address = address
+
+        # Timestamp of the first block confirming tx.
+        self.__timestamp = timestamp
+
+    @property
+    def vertex(self) -> VertexData:
+        return self.__vertex
+
+    @property
+    def address(self) -> Address | ContractId:
+        return self.__address
+
+    @property
+    def timestamp(self) -> int:
+        return self.__timestamp
+
+    @property
+    def actions(self) -> MappingProxyType[TokenUid, tuple[NCAction, ...]]:
+        """Get a mapping of actions per token."""
+        return self.__actions
+
+    def get_single_action(self, token_uid: TokenUid) -> NCAction:
+        """Get exactly one action for the provided token, and fail otherwise."""
+        actions = self.actions.get(token_uid)
+        if actions is None or len(actions) != 1:
+            raise NCFail(f'expected exactly 1 action for token {token_uid.hex()}')
+        return actions[0]
+
+    def copy(self) -> Context:
+        """Return a copy of the context."""
+        return Context(
+            actions=list(self.__all_actions__),
+            vertex=self.vertex,
+            address=self.address,
+            timestamp=self.timestamp,
+        )
+
+    def to_json(self) -> dict[str, Any]:
+        """Return a JSON representation of the context."""
+        return {
+            'actions': [action.to_json() for action in self.__all_actions__],
+            'address': get_address_b58_from_bytes(self.address),
+            'timestamp': self.timestamp,
+        }