def from_python()

in pyignite/queries/cache_info.py [0:0]

• 17 lines of code
• 9 McCabe index (conditional complexity)

    def from_python(cls, stream, value):
        cache_id = value.cache_id if value else 0
        expiry_policy = value.expiry_policy if value else None
        flags = 0

        stream.write(cache_id.to_bytes(4, byteorder=PROTOCOL_BYTE_ORDER, signed=True))

        if expiry_policy:
            if not value.protocol_context.is_expiry_policy_supported():
                raise NotSupportedByClusterError("'ExpiryPolicy' API is not supported by the cluster")
            flags |= cls.EXPIRY_POLICY_MASK

        tx_id = get_tx_id()
        if value.protocol_context.is_transactions_supported() and tx_id:
            flags |= cls.TRANSACTIONS_MASK

        stream.write(flags.to_bytes(1, byteorder=PROTOCOL_BYTE_ORDER))

        if expiry_policy:
            ExpiryPolicy.write_policy(stream, expiry_policy)

        if flags & cls.TRANSACTIONS_MASK:
            stream.write(tx_id.to_bytes(4, byteorder=PROTOCOL_BYTE_ORDER, signed=True))