Method _deserialize_function

python/pyfory/serializer.py:1506–1591

Deserialize a function from its components.

(self, read_context)

        write_context.write_ref(attrs)

    def _deserialize_function(self, read_context):
        """Deserialize a function from its components."""

        func_type_id = read_context.read_int8()
        if func_type_id == 0:
            policy = read_context.policy
            _authorize_callable_materialization(policy, types.MethodType)
            self_obj = read_context.read_ref()
            method_name = read_context.read_string()
            if policy is DEFAULT_POLICY:
                return getattr(self_obj, method_name)
            return _resolve_validated_bound_method(policy, self_obj, method_name, is_local=_is_local_receiver(self_obj))

        if func_type_id == 1:
            module = read_context.read_string()
            qualname = read_context.read_string()
            mod = _resolve_validated_module_qualname(read_context.policy, module, qualname)
            return _validate_function_value(read_context.policy, mod, is_local=_is_local_callable(mod))

        module = read_context.read_string()
        qualname = read_context.read_string()
        policy = read_context.policy
        mod = _import_validated_module(policy, module, is_local=_is_local_qualname(module, qualname))
        _authorize_callable_materialization(
            policy,
            types.FunctionType,
            module=module,
            qualname=qualname,
            is_local=True,
        )
        name = qualname.rsplit(".")[-1]

        marshalled_code = read_context.read_bytes_and_size()
        code = marshal.loads(marshalled_code)

        has_defaults = read_context.read_bool()
        defaults = None
        if has_defaults:
            num_defaults = read_context.read_var_uint32()
            _check_non_negative_size(num_defaults, "function default")
            default_values = []
            for _ in range(num_defaults):
                default_values.append(read_context.read_ref())
            defaults = tuple(default_values)

        has_closure = read_context.read_bool()
        num_freevars = read_context.read_var_uint32()
        _check_non_negative_size(num_freevars, "function closure")
        closure = None

        closure_values = []
        if has_closure:
            for _ in range(num_freevars):
                closure_values.append(read_context.read_ref())

            closure = tuple(types.CellType(value) for value in closure_values)

        num_freevars = read_context.read_var_uint32()