Source code for graphenebase.objects

# -*- coding: utf-8 -*-
import json

from collections import OrderedDict
from graphenebase.types import (
    Uint8,
    Int16,
    Uint16,
    Uint32,
    Uint64,
    Varint32,
    Int64,
    String,
    Bytes,
    Void,
    Array,
    PointInTime,
    Signature,
    Bool,
    Set,
    Fixed_array,
    Optional,
    Static_variant,
    Map,
    Id,
    VoteId,
    ObjectId,
    JsonObj,
)
from .chains import known_chains
from .objecttypes import object_type
from .account import PublicKey
from .chains import default_prefix
from .operationids import operations


class Operation(list):
    """ The superclass for an operation. This class i used to instanciate an
        operation, identify the operationid/name and serialize the operation
        into bytes.
    """

    module = "graphenebase.operations"
    fromlist = ["operations"]
    operations = operations

    def __init__(self, op, **kwargs):
        list.__init__(self, [0, GrapheneObject()])

        # Are we dealing with an actual operation as list of opid and payload?
        if isinstance(op, list) and len(op) == 2:
            self._setidanename(op[0])
            self.set(**op[1])

        # Here, we allow to only load the Operation as Template without data
        elif isinstance(op, str) or isinstance(op, int):
            self._setidanename(op)
            if kwargs:
                self.set(**kwargs)

        elif isinstance(op, GrapheneObject):
            self._loadGrapheneObject(op)

        else:
            raise ValueError("Unknown format for Operation({})".format(type(op)))

    @property
    def id(self):
        return self[0]

    @id.setter
    def id(self, value):
        assert isinstance(value, int)
        self[0] = value

    @property
    def operation(self):
        return self[1]

    @operation.setter
    def operation(self, value):
        assert isinstance(value, dict)
        self[1] = value

    @property
    def op(self):
        return self[1]

[docs] def set(self, **data): try: klass = self.klass() except Exception: # pragma: no cover raise NotImplementedError("Unimplemented Operation %s" % self.name) self.operation = klass(**data)
def _setidanename(self, identifier): if isinstance(identifier, int): self.id = int(identifier) self.name = self.getOperationNameForId(self.id) else: assert identifier in self.ops self.id = self.getOperationIdForName(identifier) self.name = identifier @property def opId(self): return self.id @property def klass_name(self): return self.name[0].upper() + self.name[1:] # klassname def _loadGrapheneObject(self, op): assert isinstance(op, GrapheneObject) self.operation = op self.name = op.__class__.__name__.lower() self.id = self.getOperationIdForName(self.name) def __bytes__(self): return bytes(Id(self.id)) + bytes(self.op) def __str__(self): return json.dumps(self.__json__()) def __json__(self): return [self.id, self.op.json()] def _getklass(self, name): module = __import__(self.module, fromlist=self.fromlist) class_ = getattr(module, name) return class_
[docs] def klass(self): return self._getklass(self.klass_name)
@property def ops(self): if callable(self.operations): # Legacy support return self.operations() else: return self.operations
[docs] def getOperationIdForName(self, name): return self.ops[name]
[docs] def getOperationNameForId(self, i): """ Convert an operation id into the corresponding string """ for key in self.ops: if int(self.ops[key]) is int(i): return key raise ValueError("Unknown Operation ID %d" % i)
toJson = __json__ json = __json__ class GrapheneObject(OrderedDict): """ Core abstraction class This class is used for any JSON reflected object in Graphene. * ``instance.__json__()``: encodes data into json format * ``bytes(instance)``: encodes data into wire format * ``str(instances)``: dumps json object as string """ def __init__(self, *args, **kwargs): if len(args) == 1 and isinstance(args[0], (dict, OrderedDict)): if hasattr(self, "detail"): super().__init__(self.detail(**args[0])) else: OrderedDict.__init__(self, args[0]) return elif kwargs and hasattr(self, "detail"): # If I receive kwargs, I need detail() implemented! super().__init__(self.detail(*args, **kwargs)) def __bytes__(self): if len(self) is 0: return bytes() b = b"" for name, value in self.items(): if isinstance(value, str): b += bytes(value, "utf-8") else: b += bytes(value) return b def __json__(self): if len(self) is 0: return {} d = {} # JSON output is *not* ordered for name, value in self.items(): if isinstance(value, Optional) and value.isempty(): continue if isinstance(value, String): d.update({name: str(value)}) else: try: d.update({name: JsonObj(value)}) except Exception: d.update({name: value.__str__()}) return d def __str__(self): return json.dumps(self.__json__()) # Legacy support @property def data(self): # pragma: no cover """ Read data explicitly (backwards compatibility) """ return self @data.setter def data(self, data): # pragma: no cover """ Set data through a setter (backwards compatibility) """ self.update(data) toJson = __json__ json = __json__ # Legacy def isArgsThisClass(self, args): return len(args) == 1 and type(args[0]).__name__ == type(self).__name__ # Common Objects class Asset(GrapheneObject): def __init__(self, *args, **kwargs): if isArgsThisClass(self, args): self.data = args[0].data else: if len(args) == 1 and len(kwargs) == 0: kwargs = args[0] super().__init__( OrderedDict( [ ("amount", Int64(kwargs["amount"])), ("asset_id", ObjectId(kwargs["asset_id"], "asset")), ] ) )