rustpython_vm/builtins/

bytes.rs

use super::{
    PositionIterInternal, PyDictRef, PyGenericAlias, PyStrRef, PyTuple, PyTupleRef, PyType,
    PyTypeRef, iter::builtins_iter, locked_next,
};
use crate::common::lock::LazyLock;
use crate::{
    AsObject, Context, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult,
    TryFromBorrowedObject, VirtualMachine,
    anystr::{self, AnyStr},
    atomic_func,
    byte::bytes_from_object,
    bytes_inner::{
        ByteInnerFindOptions, ByteInnerHexOptions, ByteInnerNewOptions, ByteInnerPaddingOptions,
        ByteInnerReplaceOptions, ByteInnerSplitOptions, ByteInnerStripOptions, ByteInnerSub,
        ByteInnerTranslateOptions, BytesJoin, DecodeArgs, PyBytesInner, bytes_decode,
    },
    class::{PyClassDef, PyClassImpl},
    common::{hash::PyHash, lock::PyMutex},
    convert::{ToPyObject, ToPyResult},
    function::{ArgBytesLike, ArgIndex, FuncArgs, OptionalArg, PyComparisonValue},
    protocol::{
        BufferDescriptor, BufferFlags, BufferMethods, PyBuffer, PyIterReturn, PyMappingMethods,
        PyNumberMethods, PySequenceMethods,
    },
    sliceable::{SequenceIndex, SliceableSequenceOp},
    types::{
        AsBuffer, AsMapping, AsNumber, AsSequence, Callable, Comparable, Constructor, Hashable,
        IterNext, Iterable, PyComparisonOp, Representable, SelfIter,
    },
};
use bstr::ByteSlice;
use core::{mem::size_of, ops::Deref};
use memchr::memchr;

#[pyclass(module = false, name = "bytes")]
#[derive(Clone, Debug)]
pub struct PyBytes {
    inner: PyBytesInner,
}

pub type PyBytesRef = PyRef<PyBytes>;

impl From<Vec<u8>> for PyBytes {
    fn from(elements: Vec<u8>) -> Self {
        Self {
            inner: PyBytesInner { elements },
        }
    }
}

impl From<PyBytesInner> for PyBytes {
    fn from(inner: PyBytesInner) -> Self {
        Self { inner }
    }
}

impl ToPyObject for Vec<u8> {
    fn to_pyobject(self, vm: &VirtualMachine) -> PyObjectRef {
        vm.ctx.new_bytes(self).into()
    }
}

impl<const N: usize> ToPyObject for &[u8; N] {
    fn to_pyobject(self, vm: &VirtualMachine) -> PyObjectRef {
        vm.ctx.new_bytes(self.as_slice().to_vec()).into()
    }
}

impl Deref for PyBytes {
    type Target = [u8];

    fn deref(&self) -> &[u8] {
        self.as_bytes()
    }
}

impl AsRef<[u8]> for PyBytes {
    fn as_ref(&self) -> &[u8] {
        self.as_bytes()
    }
}
impl AsRef<[u8]> for PyBytesRef {
    fn as_ref(&self) -> &[u8] {
        self.as_bytes()
    }
}

impl PyPayload for PyBytes {
    #[inline]
    fn class(ctx: &Context) -> &'static Py<PyType> {
        ctx.types.bytes_type
    }
}

pub(crate) fn init(context: &'static Context) {
    PyBytes::extend_class(context, context.types.bytes_type);
    PyBytesIterator::extend_class(context, context.types.bytes_iterator_type);
}

impl Constructor for PyBytes {
    type Args = Vec<u8>;

    fn slot_new(cls: PyTypeRef, args: FuncArgs, vm: &VirtualMachine) -> PyResult {
        let options: ByteInnerNewOptions = args.bind_for(vm, Self::NAME)?;

        // Optimizations for exact bytes type
        if cls.is(vm.ctx.types.bytes_type) {
            // Return empty bytes singleton
            if options.source.is_missing()
                && options.encoding.is_missing()
                && options.errors.is_missing()
            {
                return Ok(vm.ctx.empty_bytes.clone().into());
            }

            // Return exact bytes as-is
            if let OptionalArg::Present(ref obj) = options.source
                && options.encoding.is_missing()
                && options.errors.is_missing()
                && let Ok(b) = obj.clone().downcast_exact::<Self>(vm)
            {
                return Ok(b.into_pyref().into());
            }
        }

        // Handle __bytes__ method - may return PyBytes directly
        if let OptionalArg::Present(ref obj) = options.source
            && options.encoding.is_missing()
            && options.errors.is_missing()
            && let Some(bytes_method) = vm.get_method(obj.clone(), identifier!(vm, __bytes__))
        {
            let bytes = bytes_method?.call((), vm)?;
            // If exact bytes type and __bytes__ returns bytes, use it directly
            if cls.is(vm.ctx.types.bytes_type)
                && let Ok(b) = bytes.clone().downcast::<Self>()
            {
                return Ok(b.into());
            }
            // Otherwise convert to Vec<u8>
            let inner = PyBytesInner::try_from_borrowed_object(vm, &bytes)?;
            let payload = Self::py_new(&cls, inner.elements, vm)?;
            return payload.into_ref_with_type(vm, cls).map(Into::into);
        }

        let elements = options.get_inner(bytes_from_object, vm)?.elements;

        // Return empty bytes singleton for exact bytes types
        if elements.is_empty() && cls.is(vm.ctx.types.bytes_type) {
            return Ok(vm.ctx.empty_bytes.clone().into());
        }

        let payload = Self::py_new(&cls, elements, vm)?;
        payload.into_ref_with_type(vm, cls).map(Into::into)
    }

    fn py_new(_cls: &Py<PyType>, elements: Self::Args, _vm: &VirtualMachine) -> PyResult<Self> {
        Ok(Self::from(elements))
    }
}

impl PyBytes {
    #[deprecated(note = "use PyBytes::from(...).into_ref() instead")]
    pub fn new_ref(data: Vec<u8>, ctx: &Context) -> PyRef<Self> {
        Self::from(data).into_ref(ctx)
    }

    fn _getitem(&self, needle: &PyObject, vm: &VirtualMachine) -> PyResult {
        match SequenceIndex::try_from_borrowed_object(vm, needle, "byte")? {
            SequenceIndex::Int(i) => self
                .getitem_by_index(vm, i)
                .map(|x| vm.ctx.new_int(x).into()),
            SequenceIndex::Slice(slice) => self
                .getitem_by_slice(vm, slice)
                .map(|x| vm.ctx.new_bytes(x).into()),
        }
    }

    /// Check bytes for interior NULs.
    #[inline]
    #[must_use]
    pub fn contains_nuls(&self) -> bool {
        memchr(b'\0', self.as_bytes()).is_some()
    }
}

impl PyRef<PyBytes> {
    fn repeat(self, count: isize, vm: &VirtualMachine) -> PyResult<Self> {
        if count == 1 && self.class().is(vm.ctx.types.bytes_type) {
            // Special case: when some `bytes` is multiplied by `1`,
            // nothing really happens, we need to return an object itself
            // with the same `id()` to be compatible with CPython.
            // This only works for `bytes` itself, not its subclasses.
            return Ok(self);
        }
        self.payload
            .inner
            .mul(count, vm)
            .map(|x| PyBytes::from(x).into_ref(&vm.ctx))
    }
}

impl PyBytes {
    #[inline]
    #[must_use]
    pub const fn __len__(&self) -> usize {
        self.inner.len()
    }

    #[inline]
    #[must_use]
    pub const fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }

    #[inline]
    #[must_use]
    pub const fn as_bytes(&self) -> &[u8] {
        self.inner.as_bytes()
    }

    fn __mul__(zelf: PyRef<Self>, value: ArgIndex, vm: &VirtualMachine) -> PyResult<PyRef<Self>> {
        zelf.repeat(value.into_int_ref().try_to_primitive(vm)?, vm)
    }

    // TODO: Uncomment when Python adds __class_getitem__ to bytes
    // #[pyclassmethod]
    fn __class_getitem__(
        cls: PyTypeRef,
        args: PyObjectRef,
        vm: &VirtualMachine,
    ) -> PyResult<PyGenericAlias> {
        PyGenericAlias::from_args(cls, args, vm)
    }
}

#[pyclass(
    itemsize = 1,
    flags(BASETYPE, _MATCH_SELF),
    with(
        Py,
        PyRef,
        AsMapping,
        AsSequence,
        Hashable,
        Comparable,
        AsBuffer,
        Iterable,
        Constructor,
        AsNumber,
        Representable,
    )
)]
impl PyBytes {}

#[pyclass]
impl Py<PyBytes> {
    #[inline]
    pub fn as_bytes(&self) -> &[u8] {
        self.payload().as_bytes()
    }

    fn __add__(&self, other: ArgBytesLike) -> Vec<u8> {
        self.payload.inner.add(&other.borrow_buf())
    }

    fn __contains__(&self, needle: PyObjectRef, vm: &VirtualMachine) -> PyResult<bool> {
        let needle = ByteInnerSub::from_contains_arg(needle, vm)?;
        self.payload.inner.contains(needle, vm)
    }

    fn __getitem__(&self, needle: &PyObject, vm: &VirtualMachine) -> PyResult {
        self.payload._getitem(needle, vm)
    }

    fn __mod__(&self, values: PyObjectRef, vm: &VirtualMachine) -> PyResult<PyBytes> {
        let formatted = self.payload.inner.cformat(values, vm)?;
        Ok(formatted.into())
    }

    #[pymethod]
    fn __reduce_ex__(
        &self,
        _proto: usize,
        vm: &VirtualMachine,
    ) -> (PyTypeRef, PyTupleRef, Option<PyDictRef>) {
        self.__reduce__(vm)
    }

    #[pymethod]
    fn __reduce__(&self, vm: &VirtualMachine) -> (PyTypeRef, PyTupleRef, Option<PyDictRef>) {
        let bytes = PyBytes::from(self.as_bytes().to_vec()).to_pyobject(vm);
        (
            self.class().to_owned(),
            PyTuple::new_ref(vec![bytes], &vm.ctx),
            self.as_object().dict(),
        )
    }

    #[pymethod]
    fn __sizeof__(&self) -> usize {
        size_of::<PyBytes>() + self.len() * size_of::<u8>()
    }

    #[pyslot]
    fn slot_str(zelf: &PyObject, vm: &VirtualMachine) -> PyResult<PyStrRef> {
        let zelf = zelf.downcast_ref::<PyBytes>().expect("expected bytes");
        PyBytesInner::warn_on_str("str() on a bytes instance", vm)?;
        Ok(vm.ctx.new_str(zelf.payload.inner.repr_bytes(vm)?))
    }

    #[pystaticmethod]
    fn maketrans(frm: PyBytesInner, to: PyBytesInner, vm: &VirtualMachine) -> PyResult<Vec<u8>> {
        PyBytesInner::maketrans(frm, to, vm)
    }

    #[pymethod]
    fn isalnum(&self) -> bool {
        self.inner.isalnum()
    }

    #[pymethod]
    fn isalpha(&self) -> bool {
        self.inner.isalpha()
    }

    #[pymethod]
    fn isascii(&self) -> bool {
        self.inner.isascii()
    }

    #[pymethod]
    fn isdigit(&self) -> bool {
        self.inner.isdigit()
    }

    #[pymethod]
    fn islower(&self) -> bool {
        self.inner.islower()
    }

    #[pymethod]
    fn isspace(&self) -> bool {
        self.inner.isspace()
    }

    #[pymethod]
    fn isupper(&self) -> bool {
        self.inner.isupper()
    }

    #[pymethod]
    fn istitle(&self) -> bool {
        self.inner.istitle()
    }

    #[pymethod]
    fn lower(&self) -> PyBytes {
        self.inner.lower().into()
    }

    #[pymethod]
    fn upper(&self) -> PyBytes {
        self.inner.upper().into()
    }

    #[pymethod]
    fn capitalize(&self) -> PyBytes {
        self.inner.capitalize().into()
    }

    #[pymethod]
    fn swapcase(&self) -> PyBytes {
        self.inner.swapcase().into()
    }

    #[pymethod]
    pub(crate) fn hex(
        &self,
        options: ByteInnerHexOptions,
        vm: &VirtualMachine,
    ) -> PyResult<String> {
        let (sep, bytes_per_sep) = options.resolve(vm)?;
        Ok(self.inner.hex(sep, bytes_per_sep))
    }

    #[pyclassmethod]
    fn fromhex(cls: PyTypeRef, string: PyObjectRef, vm: &VirtualMachine) -> PyResult {
        let bytes = PyBytesInner::fromhex_object(&string, vm)?;
        let bytes = vm.ctx.new_bytes(bytes).into();
        PyType::call(&cls, vec![bytes].into(), vm)
    }

    #[pymethod]
    fn center(&self, options: ByteInnerPaddingOptions, vm: &VirtualMachine) -> PyResult<PyBytes> {
        Ok(self.inner.center(options, vm)?.into())
    }

    #[pymethod]
    fn ljust(&self, options: ByteInnerPaddingOptions, vm: &VirtualMachine) -> PyResult<PyBytes> {
        Ok(self.inner.ljust(options, vm)?.into())
    }

    #[pymethod]
    fn rjust(&self, options: ByteInnerPaddingOptions, vm: &VirtualMachine) -> PyResult<PyBytes> {
        Ok(self.inner.rjust(options, vm)?.into())
    }

    #[pymethod]
    fn count(&self, options: ByteInnerFindOptions, vm: &VirtualMachine) -> PyResult<usize> {
        self.inner.count(options, vm)
    }

    #[pymethod]
    fn join(&self, iterable_of_bytes: PyObjectRef, vm: &VirtualMachine) -> PyResult<PyBytesRef> {
        let items = BytesJoin::new(iterable_of_bytes, vm)?;
        if let Some(only) = items.single_bytes(vm) {
            return Ok(only);
        }
        Ok(vm
            .ctx
            .new_bytes(items.join(self.as_bytes().len(), || self.as_bytes().into(), vm)?))
    }

    #[pymethod]
    fn endswith(&self, options: anystr::StartsEndsWithArgs, vm: &VirtualMachine) -> PyResult<bool> {
        let (affix, substr) =
            match options.prepare(self.as_bytes(), self.len(), |s, r| s.get_bytes(r)) {
                Some(x) => x,
                None => return Ok(false),
            };
        substr.py_starts_ends_with(
            &affix,
            "endswith",
            "bytes",
            |s, x: PyBytesInner| s.ends_with(x.as_bytes()),
            vm,
        )
    }

    #[pymethod]
    fn startswith(
        &self,
        options: anystr::StartsEndsWithArgs,
        vm: &VirtualMachine,
    ) -> PyResult<bool> {
        let (affix, substr) =
            match options.prepare(self.as_bytes(), self.len(), |s, r| s.get_bytes(r)) {
                Some(x) => x,
                None => return Ok(false),
            };
        substr.py_starts_ends_with(
            &affix,
            "startswith",
            "bytes",
            |s, x: PyBytesInner| s.starts_with(x.as_bytes()),
            vm,
        )
    }

    #[pymethod]
    fn find(&self, options: ByteInnerFindOptions, vm: &VirtualMachine) -> PyResult<isize> {
        let index = self.inner.find(options, |h, n| h.find(n), vm)?;
        Ok(index.map_or(-1, |v| v as isize))
    }

    #[pymethod]
    fn index(&self, options: ByteInnerFindOptions, vm: &VirtualMachine) -> PyResult<usize> {
        let index = self.inner.find(options, |h, n| h.find(n), vm)?;
        index.ok_or_else(|| vm.new_value_error("substring not found"))
    }

    #[pymethod]
    fn rfind(&self, options: ByteInnerFindOptions, vm: &VirtualMachine) -> PyResult<isize> {
        let index = self.inner.find(options, |h, n| h.rfind(n), vm)?;
        Ok(index.map_or(-1, |v| v as isize))
    }

    #[pymethod]
    fn rindex(&self, options: ByteInnerFindOptions, vm: &VirtualMachine) -> PyResult<usize> {
        let index = self.inner.find(options, |h, n| h.rfind(n), vm)?;
        index.ok_or_else(|| vm.new_value_error("substring not found"))
    }

    #[pymethod]
    fn translate(
        &self,
        options: ByteInnerTranslateOptions,
        vm: &VirtualMachine,
    ) -> PyResult<PyBytes> {
        Ok(self.inner.translate(options, vm)?.into())
    }

    #[pymethod]
    fn strip(&self, options: ByteInnerStripOptions) -> PyBytes {
        self.inner.strip(options.bytes).into()
    }

    #[pymethod]
    fn removeprefix(&self, prefix: PyBytesInner) -> PyBytes {
        self.inner.removeprefix(prefix).into()
    }

    #[pymethod]
    fn removesuffix(&self, suffix: PyBytesInner) -> PyBytes {
        self.inner.removesuffix(suffix).into()
    }

    #[pymethod]
    fn split(
        &self,
        options: ByteInnerSplitOptions,
        vm: &VirtualMachine,
    ) -> PyResult<Vec<PyObjectRef>> {
        self.inner
            .split(options, |s, vm| vm.ctx.new_bytes(s.to_vec()).into(), vm)
    }

    #[pymethod]
    fn rsplit(
        &self,
        options: ByteInnerSplitOptions,
        vm: &VirtualMachine,
    ) -> PyResult<Vec<PyObjectRef>> {
        self.inner
            .rsplit(options, |s, vm| vm.ctx.new_bytes(s.to_vec()).into(), vm)
    }

    #[pymethod]
    fn partition(&self, sep: PyObjectRef, vm: &VirtualMachine) -> PyResult<PyTupleRef> {
        let sub = PyBytesInner::try_from_borrowed_object(vm, &sep)?;
        let (front, has_mid, back) = self.inner.partition(&sub, vm)?;
        Ok(vm.new_tuple((
            vm.ctx.new_bytes(front),
            if has_mid {
                sep
            } else {
                vm.ctx.new_bytes(Vec::new()).into()
            },
            vm.ctx.new_bytes(back),
        )))
    }

    #[pymethod]
    fn rpartition(&self, sep: PyObjectRef, vm: &VirtualMachine) -> PyResult<PyTupleRef> {
        let sub = PyBytesInner::try_from_borrowed_object(vm, &sep)?;
        let (back, has_mid, front) = self.inner.rpartition(&sub, vm)?;
        Ok(vm.new_tuple((
            vm.ctx.new_bytes(front),
            if has_mid {
                sep
            } else {
                vm.ctx.new_bytes(Vec::new()).into()
            },
            vm.ctx.new_bytes(back),
        )))
    }

    #[pymethod]
    fn expandtabs(&self, options: anystr::ExpandTabsArgs) -> PyBytes {
        self.inner.expandtabs(options).into()
    }

    #[pymethod]
    fn splitlines(&self, options: anystr::SplitLinesArgs, vm: &VirtualMachine) -> Vec<PyObjectRef> {
        self.inner
            .splitlines(options, |x| vm.ctx.new_bytes(x.to_vec()).into())
    }

    #[pymethod]
    fn zfill(&self, width: isize, vm: &VirtualMachine) -> PyResult<PyBytes> {
        Ok(self.inner.zfill(width, vm)?.into())
    }

    #[pymethod]
    fn replace(&self, options: ByteInnerReplaceOptions, vm: &VirtualMachine) -> PyResult<PyBytes> {
        Ok(self.inner.replace(options, vm)?.into())
    }

    #[pymethod]
    fn title(&self) -> PyBytes {
        self.inner.title().into()
    }

    #[pymethod]
    fn __getnewargs__(&self, vm: &VirtualMachine) -> PyTupleRef {
        let param: Vec<PyObjectRef> = self.elements().map(|x| x.to_pyobject(vm)).collect();
        PyTuple::new_ref(param, &vm.ctx)
    }
}

#[pyclass]
impl PyRef<PyBytes> {
    #[pymethod]
    fn __bytes__(self, vm: &VirtualMachine) -> Self {
        if self.is(vm.ctx.types.bytes_type) {
            self
        } else {
            PyBytes::from(self.payload.inner.clone()).into_ref(&vm.ctx)
        }
    }

    #[pymethod]
    fn lstrip(self, options: ByteInnerStripOptions, vm: &VirtualMachine) -> Self {
        let stripped = self.payload.inner.lstrip(options.bytes);
        if stripped == self.as_bytes() {
            self
        } else {
            vm.ctx.new_bytes(stripped.to_vec())
        }
    }

    #[pymethod]
    fn rstrip(self, options: ByteInnerStripOptions, vm: &VirtualMachine) -> Self {
        let stripped = self.payload.inner.rstrip(options.bytes);
        if stripped == self.as_bytes() {
            self
        } else {
            vm.ctx.new_bytes(stripped.to_vec())
        }
    }

    // currently, only 'utf-8' and 'ascii' implemented
    #[pymethod]
    fn decode(self, args: DecodeArgs, vm: &VirtualMachine) -> PyResult<PyStrRef> {
        bytes_decode(self.into(), args, vm)
    }
}

static BUFFER_METHODS: BufferMethods = BufferMethods {
    obj_bytes: |buffer| buffer.obj_as::<PyBytes>().as_bytes().into(),
    obj_bytes_mut: |_| panic!(),
    release: |_| {},
    retain: |_| {},
};

impl AsBuffer for PyBytes {
    fn slot_as_buffer(
        zelf: &PyObject,
        flags: BufferFlags,
        vm: &VirtualMachine,
    ) -> PyResult<PyBuffer> {
        let zelf = zelf
            .downcast_ref::<Self>()
            .ok_or_else(|| vm.new_type_error("unexpected payload for as_buffer"))?;
        flags.fill_info_check(true, vm)?;
        Self::as_buffer(zelf, vm)
    }

    fn as_buffer(zelf: &Py<Self>, _vm: &VirtualMachine) -> PyResult<PyBuffer> {
        let buf = PyBuffer::new(
            zelf.to_owned().into(),
            BufferDescriptor::simple(zelf.as_bytes().len(), true),
            &BUFFER_METHODS,
        );
        Ok(buf)
    }
}

impl AsMapping for PyBytes {
    fn as_mapping() -> &'static PyMappingMethods {
        static AS_MAPPING: LazyLock<PyMappingMethods> = LazyLock::new(|| PyMappingMethods {
            length: atomic_func!(|mapping, _vm| {
                Ok(PyBytes::mapping_downcast(mapping).as_bytes().len())
            }),
            subscript: atomic_func!(
                |mapping, needle, vm| PyBytes::mapping_downcast(mapping).__getitem__(needle, vm)
            ),
            ..PyMappingMethods::NOT_IMPLEMENTED
        });
        &AS_MAPPING
    }
}

impl AsSequence for PyBytes {
    fn as_sequence() -> &'static PySequenceMethods {
        static AS_SEQUENCE: LazyLock<PySequenceMethods> = LazyLock::new(|| PySequenceMethods {
            length: atomic_func!(|seq, _vm| Ok(PyBytes::sequence_downcast(seq).as_bytes().len())),
            concat: atomic_func!(|seq, other, vm| {
                PyBytes::sequence_downcast(seq)
                    .payload
                    .inner
                    .concat(other, vm)
                    .map(|x| vm.ctx.new_bytes(x).into())
            }),
            repeat: atomic_func!(|seq, n, vm| {
                let zelf = seq.obj.to_owned().downcast::<PyBytes>().map_err(|_| {
                    vm.new_type_error("bad argument type for built-in operation".to_owned())
                })?;
                zelf.repeat(n, vm).to_pyresult(vm)
            }),
            item: atomic_func!(|seq, i, vm| {
                PyBytes::sequence_downcast(seq)
                    .as_bytes()
                    .getitem_by_index(vm, i)
                    .map(|x| vm.ctx.new_bytes(vec![x]).into())
            }),
            contains: atomic_func!(|seq, other, vm| {
                PyBytes::sequence_downcast(seq).__contains__(other.to_owned(), vm)
            }),
            ..PySequenceMethods::NOT_IMPLEMENTED
        });
        &AS_SEQUENCE
    }
}

impl AsNumber for PyBytes {
    fn as_number() -> &'static PyNumberMethods {
        static AS_NUMBER: PyNumberMethods = PyNumberMethods {
            remainder: Some(|a, b, vm| {
                if let Some(a) = a.downcast_ref::<PyBytes>() {
                    a.__mod__(b.to_owned(), vm).to_pyresult(vm)
                } else {
                    Ok(vm.ctx.not_implemented())
                }
            }),
            ..PyNumberMethods::NOT_IMPLEMENTED
        };
        &AS_NUMBER
    }
}

impl Hashable for PyBytes {
    #[inline]
    fn hash(zelf: &Py<Self>, vm: &VirtualMachine) -> PyResult<PyHash> {
        Ok(zelf.payload.inner.hash(vm))
    }
}

impl Comparable for PyBytes {
    fn cmp(
        zelf: &Py<Self>,
        other: &PyObject,
        op: PyComparisonOp,
        vm: &VirtualMachine,
    ) -> PyResult<PyComparisonValue> {
        Ok(if let Some(res) = op.identical_optimization(zelf, other) {
            res.into()
        } else if other.fast_isinstance(vm.ctx.types.memoryview_type)
            && op != PyComparisonOp::Eq
            && op != PyComparisonOp::Ne
        {
            return Err(vm.new_type_error(format!(
                "'{}' not supported between instances of '{}' and '{}'",
                op.operator_token(),
                zelf.class().slot_name(),
                other.class().slot_name()
            )));
        } else {
            zelf.payload.inner.cmp(other, op, vm)
        })
    }
}

impl Iterable for PyBytes {
    fn iter(zelf: PyRef<Self>, vm: &VirtualMachine) -> PyResult {
        Ok(PyBytesIterator {
            internal: PyMutex::new(PositionIterInternal::new(zelf, 0)),
        }
        .into_pyobject(vm))
    }
}

impl Representable for PyBytes {
    #[inline]
    fn repr_str(zelf: &Py<Self>, vm: &VirtualMachine) -> PyResult<String> {
        zelf.payload.inner.repr_bytes(vm)
    }
}

#[pyclass(module = false, name = "bytes_iterator")]
#[derive(Debug)]
pub(crate) struct PyBytesIterator {
    internal: PyMutex<PositionIterInternal<PyBytesRef>>,
}

impl PyPayload for PyBytesIterator {
    #[inline]
    fn class(ctx: &Context) -> &'static Py<PyType> {
        ctx.types.bytes_iterator_type
    }
}

#[pyclass(flags(DISALLOW_INSTANTIATION), with(IterNext, Iterable))]
impl Py<PyBytesIterator> {
    #[pymethod]
    fn __length_hint__(&self) -> usize {
        self.internal.lock().length_hint(|obj| obj.as_bytes().len())
    }

    #[pymethod]
    fn __reduce__(&self, vm: &VirtualMachine) -> PyResult<PyTupleRef> {
        let func = builtins_iter(vm)?;
        Ok(self.internal.lock().reduce(
            func,
            |x| x.clone().into(),
            |vm| vm.ctx.empty_tuple.clone().into(),
            vm,
        ))
    }

    #[pymethod]
    fn __setstate__(&self, object: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
        self.internal
            .lock()
            .set_state(&object, |obj, pos| pos.min(obj.as_bytes().len()), vm)
    }
}

impl SelfIter for PyBytesIterator {}
impl IterNext for PyBytesIterator {
    fn next(zelf: &Py<Self>, vm: &VirtualMachine) -> PyResult<PyIterReturn> {
        locked_next(&zelf.internal, |bytes, pos| {
            Ok(PyIterReturn::from_result(
                bytes
                    .as_bytes()
                    .get(pos)
                    .map(|&x| vm.new_pyobj(x))
                    .ok_or(None),
            ))
        })
    }
}

impl<'a> TryFromBorrowedObject<'a> for PyBytes {
    fn try_from_borrowed_object(vm: &VirtualMachine, obj: &'a PyObject) -> PyResult<Self> {
        PyBytesInner::try_from_borrowed_object(vm, obj).map(|x| x.into())
    }
}