sain.collections.buf

Basic implementation of a cheap container for dealing with byte buffers.

   1# BSD 3-Clause License
   2#
   3# Copyright (c) 2022-Present, nxtlo
   4# All rights reserved.
   5#
   6# Redistribution and use in source and binary forms, with or without
   7# modification, are permitted provided that the following conditions are met:
   8#
   9# * Redistributions of source code must retain the above copyright notice, this
  10#   list of conditions and the following disclaimer.
  11#
  12# * Redistributions in binary form must reproduce the above copyright notice,
  13#   this list of conditions and the following disclaimer in the documentation
  14#   and/or other materials provided with the distribution.
  15#
  16# * Neither the name of the copyright holder nor the names of its
  17#   contributors may be used to endorse or promote products derived from
  18#   this software without specific prior written permission.
  19#
  20# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  21# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  22# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  23# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
  24# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  25# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
  26# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
  27# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  28# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  29# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  30
  31"""Basic implementation of a cheap container for dealing with byte buffers."""
  32
  33from __future__ import annotations
  34
  35__all__ = ("Bytes", "BytesMut", "Rawish", "Buffer")
  36
  37import array
  38import ctypes as _ctypes
  39import io as _io
  40import struct
  41import sys as _sys
  42import typing
  43from collections import abc as collections
  44
  45from sain import convert
  46from sain import iter as _iter
  47from sain import option as _option
  48from sain import result as _result
  49from sain.macros import assert_precondition
  50from sain.macros import deprecated
  51from sain.macros import rustc_diagnostic_item
  52from sain.macros import safe
  53from sain.macros import unsafe
  54
  55from . import slice as _slice
  56from . import vec as _vec
  57
  58if typing.TYPE_CHECKING:
  59    import inspect
  60
  61    from typing_extensions import Self
  62
  63    from sain import Option
  64    from sain import Result
  65
  66    Chars = _iter.Iterator[_ctypes.c_char]
  67    """An iterator that maps each byte in `Bytes` as a character.
  68
  69    It yields `ctypes.c_char` objects.
  70
  71    This is created by calling `Bytes.chars()`
  72    """
  73
  74
  75Rawish: typing.TypeAlias = _io.StringIO | _io.BytesIO | _io.BufferedReader
  76"""A type hint for some raw data type.
  77
  78This can be any of:
  79* `io.StringIO`
  80* `io.BytesIO`
  81* `io.BufferedReader`
  82* `memoryview`
  83"""
  84
  85Buffer: typing.TypeAlias = bytes | bytearray | collections.Iterable[int]
  86"""A type hint for some bytes data type.
  87
  88This can be any of:
  89* `bytes`
  90* `Bytes`
  91* `bytearray`
  92* `Iterable[int]`
  93* `memoryview[int]`
  94"""
  95
  96ENCODING = "utf-8"
  97
  98
  99def unwrap_bytes(data: Rawish) -> bytes:
 100    if isinstance(data, _io.StringIO):
 101        buf = bytes(data.read(), encoding=ENCODING)
 102    else:
 103        # BufferedReader | BytesIO
 104        buf = data.read()
 105    return buf
 106
 107
 108@rustc_diagnostic_item("&[u8]")
 109@typing.final
 110class Bytes(convert.ToString, collections.Sequence[int], _slice.SpecContains[int]):
 111    """Provides immutable abstractions for working with bytes.
 112
 113    It is an efficient container for storing and operating with bytes.
 114    It behaves very much like `array.array[int]` as well has the same layout.
 115
 116    A `Bytes` objects are usually used within networking applications, but can also be used
 117    elsewhere as well.
 118
 119    ## Construction
 120    `Bytes` object accept multiple rawish data types, See `Rawish` for all supported types.
 121
 122    * `Bytes()`: Initialize an empty `Bytes` object
 123    * `from_str`: Create `Bytes` from `str`
 124    * `from_bytes`: Create `Bytes` from a `Buffer` bytes-like type
 125    * `from_raw`: Create `Bytes` from a `Rawish` type
 126    * `from_ptr`: Create `Bytes` that points to an `array.array[int]` without copying it
 127    * `Bytes.zeroed(count)`: Create `Bytes` filled with `zeroes * count`.
 128
 129    Example
 130    -------
 131    ```py
 132    from sain import Bytes
 133
 134    buf = Bytes.from_str("Hello")
 135    print(buf) # [72, 101, 108, 108, 111]
 136    # buf is currently immutable, to make it mutable use `buf.to_mut()`
 137    # the conversion costs nothing, as it just points to the same underlying array.
 138    buf_mut = buf.to_mut()
 139    buf_mut.put(32)
 140    assert buf_mut == b"Hello "
 141    ```
 142    """
 143
 144    __slots__ = ("_buf",)
 145
 146    def __init__(self) -> None:
 147        """Creates a new empty `Bytes`.
 148
 149        This won't allocate the array and the returned `Bytes` will be empty.
 150        """
 151        self._buf: array.array[int] | None = None
 152
 153    # construction
 154
 155    @classmethod
 156    def from_str(cls, s: str) -> Bytes:
 157        """Create a new `Bytes` from a utf-8 string.
 158
 159        Example
 160        -------
 161        ```py
 162        buffer = Bytes.from_str("💀")
 163        ```
 164        """
 165        b = cls()
 166        b._buf = array.array("B", s.encode(ENCODING))
 167        return b
 168
 169    @classmethod
 170    def from_ptr(cls, arr: array.array[int]) -> Self:
 171        """Create a new `Bytes` from an array.
 172
 173        The returned `Bytes` will directly point to `arr` without copying.
 174
 175        Example
 176        -------
 177        ```py
 178        arr = array.array("B", b"Hello")
 179        buffer = Bytes.from_ptr(arr)
 180        ```
 181        """
 182        # this is technically an `assert` line
 183        # but Python isn't smart enough to inline and opt-out
 184        # this out of the generated bytecode.
 185        # so we'll just leave this under `if` statement.
 186        if __debug__:
 187            assert_precondition(
 188                arr.typecode == "B",
 189                f"array type must be `B`, not `{arr.typecode}`",
 190                TypeError,
 191            )
 192
 193        b = cls()
 194        b._buf = arr
 195        return b
 196
 197    @classmethod
 198    @unsafe
 199    def from_ptr_unchecked(cls, arr: array.array[int]) -> Self:
 200        """Create a new `Bytes` from an array, without checking the type code.
 201
 202        The returned `Bytes` will directly point to `arr` without copying.
 203
 204        ## Safety
 205
 206        The caller must ensure that `arr` is of type `array.array[int]` with type code `B`.
 207
 208        Example
 209        -------
 210        ```py
 211        arr = array.array("B", b"Hello")
 212        buffer = Bytes.from_ptr_unchecked(arr)
 213        ```
 214        """
 215        b = cls()
 216        b._buf = arr
 217        return b
 218
 219    @classmethod
 220    def from_bytes(cls, buf: Buffer) -> Self:
 221        """Create a new `Bytes` from an initial bytes.
 222
 223        Example
 224        -------
 225        ```py
 226        buffer = Bytes.from_bytes(b"SIGNATURE")
 227        ```
 228        """
 229        b = cls()
 230        b._buf = array.array("B", buf)
 231        return b
 232
 233    @classmethod
 234    def from_raw(cls, raw: Rawish) -> Self:
 235        """Initialize a new `Bytes` from a `Rawish` data type.
 236
 237        Example
 238        -------
 239        ```py
 240        with open('file.txt', 'rb') as file:
 241            buff = Bytes.from_raw(file)
 242
 243        # in memory bytes io
 244        bytes_io = io.BytesIO(b"data")
 245        buffer1 = Bytes.from_raw(bytes_io)
 246        # in memory string io
 247        string_io = io.StringIO("data")
 248        buffer2 = Bytes.from_raw(string_io)
 249        ```
 250        """
 251        c = cls()
 252        c._buf = array.array("B", unwrap_bytes(raw))
 253        return c
 254
 255    @classmethod
 256    def zeroed(cls, count: int) -> Self:
 257        """Initialize a new `Bytes` filled with `0 * count`.
 258
 259        Example
 260        -------
 261        ```py
 262        ALLOC_SIZE = 1024 * 2
 263        buffer = Bytes.zeros(ALLOC_SIZE)
 264        assert buffer.len() == ALLOC_SIZE
 265        ```
 266        """
 267        c = cls()
 268        c._buf = array.array("B", [0] * count)
 269        return c
 270
 271    # buffer evolution
 272
 273    # These are getting deprecated because they're trivial.
 274    # maybe we impl a `String` type and include them later.
 275    # anyways, they won't be leaving for sometime until 2.0.0.
 276
 277    @deprecated(
 278        since="1.3.0",
 279        removed_in="2.0.0",
 280        use_instead='Bytes.to_bytes().decode("utf8")',
 281        hint="Converting a bytes object to string is fairly trivial.",
 282    )
 283    def to_string(self) -> str:
 284        """Convert the bytes to a string.
 285
 286        Same as `Bytes.to_str`
 287        """
 288        return self.to_str()
 289
 290    @deprecated(
 291        since="1.3.0",
 292        removed_in="2.0.0",
 293        use_instead='Bytes.to_bytes().decode("utf8")',
 294        hint="Converting a bytes object to string is fairly trivial.",
 295    )
 296    def try_to_str(self) -> Result[str, bytes]:
 297        """A safe method to convert `self` into a string.
 298
 299        This may fail if the `self` contains invalid bytes. strings
 300        needs to be valid utf-8.
 301
 302        Example
 303        -------
 304        ```py
 305        buf = Bytes()
 306        sparkles_heart = [240, 159, 146, 150]
 307        buf.put_bytes(sparkles_heart)
 308
 309        assert buf.try_to_str().unwrap() == "💖"
 310        ```
 311
 312        Incorrect bytes
 313        ---------------
 314        ```py
 315        invalid_bytes = Bytes.from_bytes([0, 159, 146, 150])
 316        invalid_bytes.try_to_str().is_err()
 317        ```
 318
 319        Returns
 320        -------
 321        `Result[str, bytes]`
 322            If successful, returns the decoded string, otherwise the original bytes that failed
 323            to get decoded.
 324        """
 325        try:
 326            return _result.Ok(self.to_bytes().decode(ENCODING))
 327        except UnicodeDecodeError as e:
 328            return _result.Err(e.object)
 329
 330    @deprecated(
 331        since="1.3.0",
 332        removed_in="2.0.0",
 333        use_instead='str(Bytes, encoding="utf-8")',
 334        hint="Converting a bytes object to string is fairly trivial.",
 335    )
 336    def to_str(self) -> str:
 337        r"""Convert `self` to a utf-8 string.
 338
 339        During the conversion process, any invalid bytes will get converted to
 340        [REPLACEMENT_CHARACTER](https://en.wikipedia.org/wiki/Specials_(Unicode_block))
 341        which looks like this `�`, so be careful on what you're trying to convert.
 342
 343        Use `.try_to_str` try attempt the conversion in case of failure.
 344
 345        Example
 346        -------
 347        ```py
 348        buf = Bytes()
 349        sparkles_heart = [240, 159, 146, 150]
 350        buf.put_bytes(sparkles_heart)
 351
 352        assert buf.to_str() == "💖"
 353        ```
 354
 355        Incorrect bytes
 356        ---------------
 357        ```py
 358        invalid_bytes = Bytes.from_bytes(b"Hello \xf0\x90\x80World")
 359        assert invalid_bytes.to_str() == "Hello �World"
 360        ```
 361        """
 362        if not self._buf:
 363            return ""
 364
 365        return self._buf.tobytes().decode(ENCODING, errors="replace")
 366
 367    def to_bytes(self) -> bytes:
 368        """Convert `self` into `bytes`, copying the underlying array into a new buffer.
 369
 370        Example
 371        -------
 372        ```py
 373        buf = Bytes.from_str("Hello")
 374        assert buf.to_bytes() == b'Hello'
 375        ```
 376        """
 377        if not self._buf:
 378            return b""
 379
 380        return self._buf.tobytes()
 381
 382    def to_vec(self) -> _vec.Vec[int]:
 383        """Copies `self` into a new `Vec`.
 384
 385        Example
 386        -------
 387        ```py
 388        buffer = Bytes.from_str([1, 2, 3, 4])
 389        # buffer and x can be modified independently.
 390        x = buffer.to_vec()
 391        """
 392        return _vec.Vec(self.copy())
 393
 394    def leak(self) -> array.array[int]:
 395        """Consumes and leaks the `Bytes`, returning the contents as an `array[int]`,
 396
 397        A new empty array is returned if the underlying buffer is not initialized.
 398
 399        `self` will deallocate the underlying array, therefore it becomes unusable.
 400
 401        Safety
 402        ------
 403        It is unsafe to access the leaked array from `self` after calling this function.
 404
 405        Example
 406        -------
 407        ```py
 408        bytes = Bytes.from_str("chunks of data")
 409        consumed = bytes.leak()
 410        # `bytes` doesn't point to anything, this is undefined behavior.
 411        bytes.put(0)
 412        # access the array directly instead.
 413        consumed.tobytes() == b"chunks of data"
 414        ```
 415        """
 416        if self._buf is None:
 417            return array.array("B")
 418
 419        arr = self._buf
 420        # We don't need to reference this anymore since the caller will own the array.
 421        del self._buf
 422        return arr
 423
 424    def as_ptr(self) -> memoryview[int]:
 425        """Returns a read-only pointer to the buffer data.
 426
 427        `pointer` here refers to a `memoryview` object.
 428
 429        A `BufferError` is raised if the underlying sequence is not initialized.
 430
 431        Example
 432        -------
 433        ```py
 434        buffer = Bytes.from_bytes(b"data")
 435        ptr = buffer.as_ptr()
 436        ptr[0] = 1 # TypeError: cannot modify read-only memory
 437        ```
 438        """
 439        return self.__buffer__(256).toreadonly()
 440
 441    def as_ref(self) -> _slice.Slice[int]:
 442        """Get an immutable reference to the underlying sequence, without copying.
 443
 444        An empty slice is returned if the underlying sequence is not initialized.
 445
 446        Example
 447        -------
 448        ```py
 449        async def send_multipart(buf: Sequence[int]) -> None:
 450            ...
 451
 452        buffer = Bytes.from_bytes([0, 0, 0, 0])
 453        await send_multipart(buffer.as_ref()) # no copy.
 454        ```
 455        """
 456        if self._buf is not None:
 457            return _slice.Slice(self)
 458
 459        return _slice.Slice(())
 460
 461    @safe
 462    def to_mut(self) -> BytesMut:
 463        """Convert `self` into `BytesMut`.
 464
 465        This consumes `self` and returns a new `BytesMut` that points to the same underlying array,
 466        The conversion costs nothing.
 467
 468        Notes
 469        -----
 470        * If `self` is not initialized, a new empty `BytesMut` is returned.
 471        * `self` will no longer be usable, as it will not point to the underlying array.
 472
 473        The inverse method for this is `BytesMut.freeze()`
 474
 475        Example
 476        -------
 477        ```py
 478        def modify(buffer: Bytes) -> BytesMut:
 479            buf = buffer.to_mut() # doesn't cost anything.
 480            buf.swap(0, 1)
 481            return buf
 482
 483        buffer = Bytes.from_bytes([1, 2, 3, 4])
 484        new = modify(buffer)
 485        assert new == [2, 1, 3, 4]
 486        ```
 487        """
 488        # SAFETY: `Bytes.leak` returns an empty array
 489        # if `self` is uninitialized.
 490        return BytesMut.from_ptr_unchecked(self.leak())
 491
 492    def raw_parts(
 493        self,
 494    ) -> tuple[int, int]:
 495        """Return `self` as tuple containing the memory address to the buffer and how many bytes it currently contains.
 496
 497        An alias to `array.buffer_into`
 498        """
 499        if not self._buf:
 500            return (0x0, 0)
 501
 502        return self._buf.buffer_info()
 503
 504    def len(self) -> int:
 505        """Return the number of bytes in this buffer.
 506
 507        Example
 508        -------
 509        ```py
 510        buf = Bytes.from_bytes([240, 159, 146, 150])
 511        assert buf.len() == 4
 512        ```
 513        """
 514        return self.__len__()
 515
 516    def size(self) -> int:
 517        """The length in bytes of one array item in the internal representation.
 518
 519
 520        An alias to `array.itemsize`
 521
 522        Example
 523        -------
 524        ```py
 525        buf = Bytes.from_bytes([240, 159, 146, 150])
 526        assert buf.size() == 1
 527        ```
 528        """
 529        if not self._buf:
 530            return 0
 531        return self._buf.itemsize
 532
 533    def iter(self) -> _iter.TrustedIter[int]:
 534        """Returns an iterator over the contained bytes.
 535
 536        This iterator yields all `int`s from start to end.
 537
 538        Example
 539        -------
 540        ```py
 541        buf = Bytes.from_bytes((1, 2, 3))
 542        iterator = buf.iter()
 543
 544        # map each byte to a character
 545        for element in iterator.map(chr):
 546            print(element)
 547        # ☺
 548        # ☻
 549        # ♥
 550        ```
 551        """
 552        return _iter.TrustedIter(self.as_ptr())
 553
 554    def chars(self) -> Chars:
 555        """Returns an iterator over the characters of `Bytes`.
 556
 557        This iterator yields all `int`s from start to end mapped as a `ctypes.c_char`.
 558
 559        Example
 560        -------
 561        ```py
 562        b = Bytes.from_str("Hello")
 563        for char in b.chars():
 564            print(char)
 565
 566        # c_char(b'H')
 567        # c_char(b'e')
 568        # c_char(b'l')
 569        # c_char(b'l')
 570        # c_char(b'o')
 571        ```
 572        """
 573        # The built-in map is actually faster than our own pure python adapter impl.
 574        return _iter.Iter(map(_ctypes.c_char, self))
 575
 576    def is_empty(self) -> bool:
 577        """Check whether `self` contains any bytes or not.
 578
 579        Example
 580        -------
 581        ```py
 582        buffer = Bytes()
 583        assert buffer.is_empty()
 584        ```
 585        """
 586        return not self._buf
 587
 588    def split_off(self, at: int) -> Bytes:
 589        """Split the bytes off at the specified position, returning a new
 590        `Bytes` at the range of `[at : len]`, leaving `self` at `[at : bytes_len]`.
 591
 592        if this bytes is empty, `self` is returned unchanged.
 593
 594        Example
 595        -------
 596        ```py
 597        origin = Bytes.from_bytes((1, 2, 3, 4))
 598        split = origin.split_off(2)
 599
 600        print(origin, split)  # [1, 2], [3, 4]
 601        ```
 602
 603        Raises
 604        ------
 605        `RuntimeError`
 606            This method will raise if `at` > `len(self)`
 607        """
 608        len_ = self.len()
 609        if at > len_:
 610            raise RuntimeError(
 611                f"Index `at` ({at}) should be <= than len of `self` ({len_}) "
 612            ) from None
 613
 614        # Either the list is empty or uninit.
 615        if not self._buf:
 616            return self
 617
 618        split = self[at:len_]  # split the items into a new buffer.
 619        del self._buf[at:len_]  # remove the items from the original list.
 620        return split
 621
 622    def split_first(self) -> Option[tuple[int, Bytes]]:
 623        """Split the first and rest elements of the bytes, If empty, `None` is returned.
 624
 625        Example
 626        -------
 627        ```py
 628        buf = Bytes.from_bytes([1, 2, 3])
 629        split = buf.split_first()
 630        assert split == Some((1, [2, 3]))
 631        ```
 632        """
 633        if not self._buf:
 634            return _option.NOTHING
 635
 636        # optimized to only one element in the buffer.
 637        if self.len() == 1:
 638            return _option.Some((self[0], Bytes()))
 639
 640        first, rest = self[0], self[1:]
 641        return _option.Some((first, rest))
 642
 643    def split_last(self) -> Option[tuple[int, Bytes]]:
 644        """Returns the last and rest of the elements of the bytes, If `self` is empty, `None` is returned.
 645
 646        Example
 647        -------
 648        ```py
 649        buf = Bytes.from_bytes([0, 1, 2])
 650        last, elements = buf.split_last().unwrap()
 651        assert (last, elements) == (3, [1, 2])
 652        ```
 653        """
 654        if not self._buf:
 655            return _option.NOTHING
 656
 657        len_ = self.len()
 658        # optimized to only one element in the buffer.
 659        if len_ == 1:
 660            return _option.Some((self[0], Bytes()))
 661
 662        last, rest = self[-1], self[:-1]
 663        return _option.Some((last, rest))
 664
 665    def split_at(self, mid: int) -> tuple[Bytes, Bytes]:
 666        """Divide `self` into two at an index.
 667
 668        The first will contain all bytes from `[0:mid]` excluding `mid` it self.
 669        and the second will contain the remaining bytes.
 670
 671        if `mid` > `self.len()`, Then all bytes will be moved to the left,
 672        returning an empty bytes in right.
 673
 674        Example
 675        -------
 676        ```py
 677        buffer = Bytes.from_bytes((1, 2, 3, 4))
 678        left, right = buffer.split_at(0)
 679        assert left == [] and right == [1, 2, 3, 4]
 680
 681        left, right = buffer.split_at(2)
 682        assert left == [1, 2] and right == [2, 3]
 683        ```
 684
 685        The is roughly the implementation
 686        ```py
 687        self[0:mid], self[mid:]
 688        ```
 689        """
 690        return self[0:mid], self[mid:]
 691
 692    # layout methods.
 693
 694    @safe
 695    def copy(self) -> Bytes:
 696        """Create a copy of the bytes.
 697
 698        Example
 699        -------
 700        ```py
 701        original = Bytes.from_bytes([255, 255, 255, 0])
 702        copy = original.copy()
 703        ```
 704        """
 705        if not self._buf:
 706            return Bytes()
 707
 708        # SAFETY: `self._buf` is initialized.
 709        return self.from_ptr_unchecked(self._buf[:])
 710
 711    def index(self, v: int, start: int = 0, stop: int = _sys.maxsize) -> int:
 712        """Return the smallest `i` such that `i` is the index of the first occurrence of `v` in the buffer.
 713
 714        The optional arguments start and stop can be specified to search for x within a
 715        subsection of the array. Raise ValueError if x is not found
 716        """
 717        if not self._buf:
 718            raise ValueError from None
 719
 720        return self._buf.index(v, start, stop)
 721
 722    def count(self, x: int) -> int:
 723        """Return the number of occurrences of `x` in the buffer.
 724
 725        Example
 726        --------
 727        ```py
 728        buf = Bytes([32, 32, 31])
 729        assert buf.count(32) == 2
 730        ```
 731        """
 732        if self._buf is None:
 733            return 0
 734
 735        return self._buf.count(x)
 736
 737    # special methods
 738
 739    def __iter__(self) -> collections.Iterator[int]:
 740        if self._buf:
 741            return self._buf.__iter__()
 742
 743        return ().__iter__()
 744
 745    def __len__(self) -> int:
 746        return len(self._buf) if self._buf else 0
 747
 748    def __repr__(self) -> str:
 749        if not self._buf:
 750            return "[]"
 751
 752        return "[" + ", ".join(str(x) for x in self._buf) + "]"
 753
 754    __str__ = __repr__
 755
 756    def __bytes__(self) -> bytes:
 757        return self.to_bytes()
 758
 759    def __buffer__(self, flag: int | inspect.BufferFlags) -> memoryview[int]:
 760        if not self._buf:
 761            raise BufferError("Cannot work with uninitialized bytes.")
 762
 763        if _sys.version_info >= (3, 12):
 764            mem = self._buf.__buffer__(flag)
 765        else:
 766            # arrays in 3.11 and under don't implement the buffer protocol.
 767            mem = memoryview(self._buf)
 768
 769        return mem
 770
 771    def __contains__(self, byte: int) -> bool:
 772        return byte in self._buf if self._buf else False
 773
 774    def __eq__(self, other: object, /) -> bool:
 775        if not self._buf:
 776            return False
 777
 778        if isinstance(other, bytes):
 779            return self._buf.tobytes() == other
 780
 781        # bytes IS a `Sequence[int]`, but not all `Sequence[int]`
 782        # represented as bytes.
 783        elif isinstance(other, collections.Sequence):
 784            return self._buf.tolist() == other
 785
 786        return self._buf.__eq__(other)
 787
 788    def __ne__(self, other: object, /) -> bool:
 789        return not self.__eq__(other)
 790
 791    def __le__(self, other: object) -> bool:
 792        if not self._buf:
 793            return False
 794
 795        if not isinstance(other, Bytes):
 796            return NotImplemented
 797
 798        if not other._buf:
 799            return False
 800
 801        return self._buf <= other._buf
 802
 803    def __ge__(self, other: object) -> bool:
 804        if not self._buf:
 805            return False
 806
 807        if not isinstance(other, Bytes):
 808            return NotImplemented
 809
 810        if not other._buf:
 811            return False
 812
 813        return self._buf >= other._buf
 814
 815    def __lt__(self, other: object) -> bool:
 816        if not self._buf:
 817            return False
 818
 819        if not isinstance(other, Bytes):
 820            return NotImplemented
 821
 822        if not other._buf:
 823            return False
 824
 825        return self._buf < other._buf
 826
 827    def __gt__(self, other: object) -> bool:
 828        if not self._buf:
 829            return False
 830
 831        if not isinstance(other, Bytes):
 832            return NotImplemented
 833
 834        if not other._buf:
 835            return False
 836
 837        return self._buf > other._buf
 838
 839    @typing.overload
 840    def __getitem__(self, index: slice) -> Bytes: ...
 841
 842    @typing.overload
 843    def __getitem__(self, index: int) -> int: ...
 844
 845    @safe
 846    def __getitem__(self, index: int | slice) -> int | Bytes:
 847        if not self._buf:
 848            raise IndexError("Index out of range")
 849
 850        if isinstance(index, slice):
 851            # SAFETY: `self._buf` is initialized.
 852            return self.from_ptr_unchecked(self._buf[index])
 853
 854        return self._buf[index]
 855
 856    def __reversed__(self) -> collections.Iterator[int]:
 857        return reversed(self._buf or ())
 858
 859    # defined like `array`'s
 860    __hash__: typing.ClassVar[None] = None
 861
 862    @safe
 863    def __copy__(self) -> Bytes:
 864        if not self._buf:
 865            return Bytes()
 866
 867        return Bytes.from_ptr_unchecked(self._buf.__copy__())
 868
 869    @safe
 870    def __deepcopy__(self, unused: typing.Any, /) -> Bytes:
 871        if not self._buf:
 872            return Bytes()
 873
 874        return Bytes.from_ptr_unchecked(self._buf.__deepcopy__(unused))
 875
 876
 877@rustc_diagnostic_item("&mut [u8]")
 878@typing.final
 879class BytesMut(
 880    Bytes,  # pyright: ignore - we want to inherit from `Bytes`.
 881    collections.MutableSequence[int],
 882):
 883    """Provides mutable abstractions for working with bytes.
 884
 885    It is an efficient container for storing and operating with bytes,
 886    It is built on-top of `array.array[int]`, which means you get all of `array[int]`'s operations.
 887
 888    A `bytes` object is usually used within networking applications, but can also be used
 889    elsewhere as well.
 890
 891    ## Construction
 892    You can create a `BytesMut` object in multiple ways.
 893
 894    * `BytesMut()`: Initialize an empty `BytesMut` object
 895    * `from_str`: Create `BytesMut` from `str`
 896    * `from_bytes`: Create `BytesMut` from a `Buffer` bytes-like type
 897    * `from_raw`: Create `BytesMut` from a `Rawish` type
 898    * `from_ptr`: Create `BytesMut` that points to an `array.array[int]` without copying it
 899    * `BytesMut.zeroed(count)`: Create `BytesMut` filled with `zeroes * count`.
 900
 901    Example
 902    -------
 903    ```py
 904    from sain import BytesNut
 905
 906    buf = BytesMut()
 907    buffer.put_bytes(b"Hello")
 908    print(buffer) # [72, 101, 108, 108, 111]
 909
 910    buf.put(32) # space
 911    assert buffer.to_bytes() == b"Hello "
 912    ```
 913    """
 914
 915    __slots__ = ("_buf",)
 916
 917    def __init__(self) -> None:
 918        """Creates a new empty `BytesMut`.
 919
 920        This won't allocate the array and the returned `BytesMut` will be empty.
 921        """
 922        super().__init__()
 923
 924    # default methods.
 925
 926    def extend(self, src: Buffer) -> None:
 927        """Extend `self` from a `src`.
 928
 929        Example
 930        -------
 931        ```py
 932        buf = Bytes()
 933        buf.extend([1, 2, 3])
 934        assert buf == [1, 2, 3]
 935        ```
 936
 937        Parameters
 938        ----------
 939        src : `Buffer`
 940            Can be one of `Bytes`, `bytes`, `bytearray` or `Sequence[int]`
 941
 942        Raises
 943        ------
 944        `OverflowError`
 945            If `src` not in range of `0..255`
 946        """
 947        if self._buf is None:
 948            # If it was `None`, we initialize it with a source instead of extending.
 949            self._buf = array.array("B", src)
 950        else:
 951            self._buf.extend(src)
 952
 953    def put(self, src: int) -> None:
 954        """Append a byte at the end of the array.
 955
 956        unlike `.put_bytes`, this method appends instead of extending the array
 957        which is faster if you're putting a single byte in a single call.
 958
 959        Example
 960        -------
 961        ```py
 962        buf = Bytes()
 963        buf.put(32) # append a space to the end of the buffer
 964        assert buf.to_bytes() == b' '
 965        ```
 966
 967        Parameters
 968        ----------
 969        src : `int`
 970            An unsigned integer, also known as `u8` to put.
 971
 972        Raises
 973        ------
 974        `OverflowError`
 975            If `src` not in range of `0..255`
 976        """
 977        if self._buf is None:
 978            # If it was `None`, we initialize it with a source instead of appending.
 979            self._buf = array.array("B", [src])
 980        else:
 981            self._buf.append(src)
 982
 983    def put_float(self, src: float) -> None:
 984        r"""Writes a single-precision (4 bytes) floating point number to `self` in big-endian byte order.
 985
 986        The valid range for the float value is `-3.402823466e38` to `3.402823466e38`.
 987
 988        Example
 989        -------
 990        ```py
 991        buf = BytesMut()
 992        buf.put_float(1.2)
 993        assert buf.to_bytes() == b'\x3f\x99\x99\x9a'
 994        ```
 995
 996        Raises
 997        ------
 998        `OverflowError`
 999            If `src` is out of range.
1000        """
1001        assert_precondition(
1002            -3.402823466e38 <= src <= 3.402823466e38,
1003            f"Float {src} is out of range for a single-precision float",
1004            OverflowError,
1005        )
1006        bits = struct.pack(">f", src)
1007
1008        if self._buf is None:
1009            self._buf = array.array("B", bits)
1010        else:
1011            self._buf.extend(bits)
1012
1013    def put_char(self, char: str) -> None:
1014        """Append a single character to the buffer.
1015
1016        This is the same as `self.put(ord(char))`.
1017
1018        Example
1019        -------
1020        ```py
1021        buf = BytesMut()
1022        buf.put_char('a')
1023        assert buf == b"a"
1024        ```
1025
1026        Parameters
1027        ----------
1028        char : `str`
1029            The character to put.
1030        """
1031        assert (ln := len(char)) == 1, f"Expected a single character, got {ln}"
1032        self.put(ord(char))
1033
1034    def put_raw(self, src: Rawish) -> None:
1035        """Extend `self` from a raw data type source.
1036
1037        Example
1038        -------
1039        ```py
1040        buffer = BytesMut()
1041        # A file descriptor's contents
1042        with open('file.txt', 'rb') as file:
1043            buffer.put_raw(file)
1044
1045        # bytes io
1046        buffer.put(io.BytesIO(b"data"))
1047        # string io
1048        buffer.put(io.StringIO("data"))
1049        ```
1050
1051        Parameters
1052        ----------
1053        src : `Rawish`
1054            A valid raw data type. See `Rawish` for more details.
1055
1056        Raises
1057        ------
1058        `OverflowError`
1059            If `src` not in range of `0..255`
1060        """
1061        if self._buf is None:
1062            # If it was `None`, we initialize it with a source instead of extending.
1063            self._buf = array.array("B", unwrap_bytes(src))
1064        else:
1065            self._buf.extend(unwrap_bytes(src))
1066
1067    def put_bytes(self, src: Buffer) -> None:
1068        """Put `bytes` into `self`.
1069
1070        Example
1071        -------
1072        ```py
1073        buf = BytesMut.from_bytes(b"hello")
1074        buf.put_bytes([32, 119, 111, 114, 108, 100])
1075        assert buf == b"hello world"
1076        ```
1077
1078        Parameters
1079        ----------
1080        src : `Buffer`
1081            Can be one of `Bytes`, `bytes`, `bytearray` or `Sequence[int]`
1082
1083        Raises
1084        ------
1085        `OverflowError`
1086            If `src` not in range of `0..255`
1087        """
1088        if self._buf is None:
1089            # If it was `None`, we initialize it with a source instead of extending.
1090            self._buf = array.array("B", src)
1091        else:
1092            self._buf.extend(src)
1093
1094    def put_str(self, s: str) -> None:
1095        """Put a `utf-8` encoded bytes from a string.
1096
1097        Example
1098        -------
1099        ```py
1100        buffer = BytesMut()
1101        buffer.put_str("hello")
1102
1103        assert buffer == b"hello"
1104        ```
1105
1106        Parameters
1107        ----------
1108        src: `str`
1109            The string
1110        """
1111        self.put_bytes(s.encode(ENCODING))
1112
1113    def replace(self, index: int, byte: int) -> None:
1114        """Replace the byte at `index` with `byte`.
1115
1116        This method is `NOOP` if:
1117        -------------------------
1118        * `self` is empty or unallocated.
1119        * `index` is out of range.
1120
1121        Example
1122        -------
1123        ```py
1124        buf = BytesMut.from_bytes([1, 2, 3])
1125        buf.replace(1, 4)
1126        assert buf == [1, 4, 3]
1127        ```
1128        """
1129        if not self._buf or index < 0 or index >= self.len():
1130            return
1131
1132        self._buf[index] = byte
1133
1134    def replace_with(self, index: int, f: collections.Callable[[int], int]) -> None:
1135        """Replace the byte at `index` with a byte returned from `f`.
1136
1137        The signature of `f` is `Fn(int) -> int`, where the argument is the old byte.
1138
1139        ## This method is `NOOP` if:
1140        * `self` is empty or unallocated.
1141        * `index` is out of range.
1142
1143        Example
1144        -------
1145        ```py
1146        buf = BytesMut.from_bytes([1, 2, 3])
1147        buf.replace_with(1, lambda prev: prev * 2)
1148        assert buf == [1, 4, 3]
1149        ```
1150        """
1151        if not self._buf or index < 0 or index >= self.len():
1152            return
1153
1154        old = self._buf[index]
1155        self._buf[index] = f(old)
1156
1157    def offset(self, f: collections.Callable[[int], int]) -> None:
1158        """Modify each byte in the buffer with a new byte returned from `f`.
1159
1160        The signature of `f` is `Fn(int) -> int`, where the argument is the previous byte.
1161
1162        Example
1163        -------
1164        ```py
1165        buf = BytesMut.from_bytes([1, 2, 3])
1166        buf.offset(lambda prev: prev * 2)
1167        assert buf == [2, 4, 6]
1168        ```
1169
1170        This method is `NOOP` if `self` is empty or unallocated.
1171        """
1172
1173        if not self._buf:
1174            return
1175
1176        for index in range(len(self._buf)):
1177            self._buf[index] = f(self._buf[index])
1178
1179    def fill(self, value: int) -> None:
1180        """Fills `self` with the given byte.
1181
1182        Nothing happens if the buffer is empty or unallocated.
1183
1184        Example
1185        -------
1186        ```py
1187        a = Bytes.from_bytes([0, 1, 2, 3])
1188        a.fill(0)
1189        assert a == [0, 0, 0, 0]
1190        ```
1191        """
1192        if not self._buf:
1193            return
1194
1195        self.as_mut_ptr()[:] = bytearray([value] * self.len())
1196
1197    def fill_with(self, f: collections.Callable[[], int]) -> None:
1198        """Fills `self` with the given byte returned from `f`.
1199
1200        Nothing happens if the buffer is empty or unallocated.
1201
1202        Example
1203        -------
1204        ```py
1205        def default() -> int:
1206            return 0
1207
1208        a = Bytes.from_bytes([0, 1, 2, 3])
1209        a.fill_with(default)
1210        assert a == [0, 0, 0, 0]
1211        ```
1212        """
1213        if not self._buf:
1214            return
1215
1216        self.as_mut_ptr()[:] = bytearray([f()] * self.len())
1217
1218    def swap(self, a: int, b: int):
1219        """Swap two bytes in the buffer.
1220
1221        if `a` equals to `b` then it's guaranteed that elements won't change value.
1222
1223        Example
1224        -------
1225        ```py
1226        buf = Bytes.from_bytes([1, 2, 3, 4])
1227        buf.swap(0, 3)
1228        assert buf == [4, 2, 3, 1]
1229        ```
1230
1231        Raises
1232        ------
1233        IndexError
1234            If the positions of `a` or `b` are out of index.
1235        """
1236        if self[a] == self[b]:
1237            return
1238
1239        self[a], self[b] = self[b], self[a]
1240
1241    def swap_unchecked(self, a: int, b: int):
1242        """Swap two bytes in the buffer. without checking if `a` == `b`.
1243
1244        If you care about `a` and `b` equality, see `Bytes.swap`.
1245
1246        Example
1247        -------
1248        ```py
1249        buf = Bytes.from_bytes([1, 2, 3, 1])
1250        buf.swap_unchecked(0, 3)
1251        assert buf == [1, 2, 3, 1]
1252        ```
1253
1254        Raises
1255        ------
1256        IndexError
1257            If the positions of `a` or `b` are out of index.
1258        """
1259        self[a], self[b] = self[b], self[a]
1260
1261    def as_mut_ptr(self) -> memoryview[int]:
1262        """Returns a mutable pointer to the buffer data.
1263
1264        `pointer` here refers to a `memoryview` object.
1265
1266        A `BufferError` is raised if the underlying sequence is not initialized.
1267
1268        Example
1269        -------
1270        ```py
1271        buffer = BytesMut.from_str("ouv")
1272        ptr = buffer.as_mut_ptr()
1273        ptr[0] = ord(b'L')
1274        assert buffer.to_bytes() == b"Luv"
1275        ```
1276        """
1277        return self.__buffer__(512)
1278
1279    def as_mut(self) -> _slice.SliceMut[int]:
1280        """Get a mutable reference to the underlying sequence, without copying.
1281
1282        An empty slice is returned if the underlying sequence is not initialized.
1283
1284        Example
1285        -------
1286        ```py
1287        buff = BytesMut.from_str("Hello")
1288        ref = buff.as_mut()
1289        ref.append(32)
1290        del ref
1291        assert buff == b"Hello "
1292        ```
1293        """
1294        if self._buf is not None:
1295            return _slice.SliceMut(self)
1296
1297        return _slice.SliceMut([])
1298
1299    @safe
1300    def freeze(self) -> Bytes:
1301        """Convert `self` into an immutable `Bytes`.
1302
1303        This conversion is zero-cost, meaning it doesn't any _hidden-copy_ operations behind the scenes.
1304        This consumes `self` and returns a new `Bytes` that points to the same underlying array,
1305
1306        Notes
1307        -----
1308        * If `self` is not initialized, a new empty `Bytes` is returned.
1309        * `self` will no longer be usable, as it will not point to the underlying array.
1310
1311        The inverse method of this is `Bytes.to_mut()`
1312
1313        Example
1314        -------
1315        ```py
1316        def shrink_to(cap: int, buffer: BytesMut) -> Bytes:
1317            buf.truncate(cap)
1318            return buf.freeze()
1319
1320        buffer = BytesMut.from_bytes([32, 23, 34, 65])
1321        # accessing `buffer` after this is undefined behavior.
1322        modified = shrink_to(2, buffer)
1323        assert modified == [32, 23]
1324        ```
1325        """
1326        # SAFETY: `Bytes.leak` returns an empty array
1327        # if `self` is uninitialized.
1328        return Bytes.from_ptr_unchecked(self.leak())
1329
1330    def swap_remove(self, byte: int) -> int:
1331        """Remove the first appearance of `item` from this buffer and return it.
1332
1333        Raises
1334        ------
1335        * `ValueError`: if `item` is not in this buffer.
1336        * `MemoryError`: if this buffer hasn't allocated, Aka nothing has been pushed to it.
1337
1338        Example
1339        -------
1340        ```py
1341        buf = BytesMut.from_bytes([1, 2, 3, 4])
1342        assert 1 == buf.swap_remove(1)
1343        assert buf == [2, 3, 4]
1344        ```
1345        """
1346        if self._buf is None:
1347            raise MemoryError("`self` is unallocated.") from None
1348
1349        return self._buf.pop(self.index(byte))
1350
1351    def truncate(self, size: int) -> None:
1352        """Shortens the bytes, keeping the first `size` elements and dropping the rest.
1353
1354        Example
1355        -------
1356        ```py
1357        buf = BytesMut.from_bytes([0, 0, 0, 0])
1358        buf.truncate(1)
1359        assert buf.len() == 1
1360        ```
1361        """
1362        if not self._buf:
1363            return
1364
1365        del self._buf[size:]
1366
1367    def split_off_mut(self, at: int) -> BytesMut:
1368        """Split the bytes off at the specified position, returning a new
1369        `BytesMut` at the range of `[at : len]`, leaving `self` at `[at : bytes_len]`.
1370
1371        if this bytes is empty, `self` is returned unchanged.
1372
1373        Example
1374        -------
1375        ```py
1376        origin = BytesMut.from_bytes((1, 2, 3, 4))
1377        split = origin.split_off_mut(2)
1378
1379        print(origin, split)  # [1, 2], [3, 4]
1380        ```
1381
1382        Raises
1383        ------
1384        `IndexError`
1385            This method will raise if `at` > `len(self)`
1386        """
1387        len_ = self.len()
1388        if at > len_:
1389            raise IndexError(
1390                f"the index of `at` ({at}) should be <= than len of `self` ({len_}) "
1391            ) from None
1392
1393        if not self._buf:
1394            return self
1395
1396        split = BytesMut.from_ptr(self._buf[at:len_])
1397        del self._buf[at:len_]
1398        return split
1399
1400    def split_first_mut(self) -> Option[tuple[int, BytesMut]]:
1401        """Split the first and rest elements of the bytes, If empty, `None` is returned.
1402
1403        Returns a tuple of (first, rest) where rest is a mutable sequence.
1404
1405        Example
1406        -------
1407        ```py
1408        buf = BytesMut.from_bytes([1, 2, 3])
1409        split = buf.split_first_mut()
1410        assert split == Some((1, [2, 3]))
1411        ```
1412        """
1413        if not self._buf:
1414            return _option.NOTHING
1415
1416        if self.len() == 1:
1417            return _option.Some((self[0], BytesMut()))
1418
1419        first = self[0]
1420        rest = self[1:]
1421        return _option.Some((first, rest))
1422
1423    def split_last_mut(self) -> Option[tuple[int, Bytes]]:
1424        """Returns the last and rest of the elements of the bytes, If `self` is empty, `None` is returned.
1425
1426        Returns a tuple of (last, rest) where rest is a mutable sequence.
1427
1428        Example
1429        -------
1430        ```py
1431        buf = BytesMut.from_bytes([0, 1, 2])
1432        last, elements = buf.split_last_mut().unwrap()
1433        assert (last, elements) == (2, [0, 1])
1434        ```
1435        """
1436        if not self._buf:
1437            return _option.NOTHING
1438
1439        len_ = self.len()
1440        if len_ == 1:
1441            return _option.Some((self[0], BytesMut()))
1442
1443        last = self[-1]
1444        rest = self[:-1]
1445        return _option.Some((last, rest))
1446
1447    def split_at_mut(self, mid: int) -> tuple[BytesMut, BytesMut]:
1448        """Divide `self` into two at an index.
1449
1450        The first will contain all bytes from `[0:mid]` excluding `mid` itself.
1451        and the second will contain the remaining bytes.
1452
1453        if `mid` > `self.len()`, Then all bytes will be moved to the left,
1454        returning an empty bytes in right.
1455
1456        Example
1457        -------
1458        ```py
1459        buffer = BytesMut.from_bytes((1, 2, 3, 4))
1460        left, right = buffer.split_at_mut(0)
1461        assert left == [] and right == [1, 2, 3, 4]
1462
1463        left, right = buffer.split_at_mut(2)
1464        assert left == [1, 2] and right == [3, 4]
1465        ```
1466
1467        The is roughly the implementation
1468        ```py
1469        self[0:mid], self[mid:]
1470        ```
1471        """
1472        return self[0:mid], self[mid:]
1473
1474    # * Layout * #
1475
1476    def insert(self, index: int, value: int) -> None:
1477        """Insert a new item with `value` in the buffer before position `index`.
1478
1479        Negative values are treated as being relative to the end of the buffer.
1480        """
1481        if self._buf is None:
1482            return
1483
1484        self._buf.insert(index, value)
1485
1486    def pop(self, i: int = -1) -> Option[int]:
1487        """Removes the last element from the buffer and returns it, `Some(None)` if it is empty.
1488
1489        Example
1490        -------
1491        ```py
1492        buf = BytesMut((21, 32, 44))
1493        assert buf.pop() == Some(44)
1494        ```
1495        """
1496        if not self._buf:
1497            return _option.NOTHING
1498
1499        return _option.Some(self._buf.pop(i))
1500
1501    def remove(self, i: int) -> None:
1502        """Remove the first appearance of `i` from `self`.
1503
1504        Example
1505        ------
1506        ```py
1507        buf = BytesMut.from_bytes([1, 1, 2, 3, 4])
1508        buf.remove(1)
1509        print(buf) # [1, 2, 3, 4]
1510        ```
1511        """
1512        if not self._buf:
1513            return
1514
1515        self._buf.remove(i)
1516
1517    def clear(self) -> None:
1518        """Clear the buffer.
1519
1520        Example
1521        -------
1522        ```py
1523        buf = BytesMut.from_bytes([255])
1524        buf.clear()
1525
1526        assert buf.is_empty()
1527        ```
1528        """
1529        if not self._buf:
1530            return
1531
1532        del self._buf[:]
1533
1534    def byteswap(self) -> None:
1535        """Swap the byte order of the bytes in `self`."""
1536        if not self._buf:
1537            return
1538
1539        self._buf.byteswap()
1540
1541    def copy(self) -> BytesMut:
1542        """Create a copy of the bytes.
1543
1544        Example
1545        -------
1546        ```py
1547        original = BytesMut.from_bytes([255, 255, 255, 0])
1548        copy = original.copy()
1549        ```
1550        """
1551        if not self._buf:
1552            return BytesMut()
1553
1554        return self.from_ptr(self._buf[:])
1555
1556    def __setitem__(self, index: int, value: int):
1557        if not self._buf:
1558            raise IndexError("index out of range")
1559
1560        self._buf[index] = value
1561
1562    def __delitem__(self, key: typing.SupportsIndex | slice, /) -> None:
1563        if not self._buf:
1564            raise IndexError("index out of range")
1565
1566        del self._buf[key]
1567
1568    @typing.overload
1569    def __getitem__(self, index: slice) -> BytesMut: ...
1570
1571    @typing.overload
1572    def __getitem__(self, index: int) -> int: ...
1573
1574    @safe
1575    def __getitem__(self, index: int | slice) -> int | BytesMut:
1576        if not self._buf:
1577            raise IndexError("index out of range")
1578
1579        if isinstance(index, slice):
1580            # SAFETY: `self._buf` is initialized.
1581            return self.from_ptr_unchecked(self._buf[index])
1582
1583        return self._buf[index]
1584
1585    @safe
1586    def __copy__(self) -> BytesMut:
1587        if not self._buf:
1588            return BytesMut()
1589
1590        return BytesMut.from_ptr_unchecked(self._buf.__copy__())
1591
1592    @safe
1593    def __deepcopy__(self, unused: typing.Any, /) -> BytesMut:
1594        if not self._buf:
1595            return BytesMut()
1596
1597        return BytesMut.from_ptr_unchecked(self._buf.__deepcopy__(unused))
@rustc_diagnostic_item('&[u8]')
@typing.final
class Bytes(sain.convert.ToString, collections.abc.Sequence[int], sain.collections.slice.SpecContains[int]):
109@rustc_diagnostic_item("&[u8]")
110@typing.final
111class Bytes(convert.ToString, collections.Sequence[int], _slice.SpecContains[int]):
112    """Provides immutable abstractions for working with bytes.
113
114    It is an efficient container for storing and operating with bytes.
115    It behaves very much like `array.array[int]` as well has the same layout.
116
117    A `Bytes` objects are usually used within networking applications, but can also be used
118    elsewhere as well.
119
120    ## Construction
121    `Bytes` object accept multiple rawish data types, See `Rawish` for all supported types.
122
123    * `Bytes()`: Initialize an empty `Bytes` object
124    * `from_str`: Create `Bytes` from `str`
125    * `from_bytes`: Create `Bytes` from a `Buffer` bytes-like type
126    * `from_raw`: Create `Bytes` from a `Rawish` type
127    * `from_ptr`: Create `Bytes` that points to an `array.array[int]` without copying it
128    * `Bytes.zeroed(count)`: Create `Bytes` filled with `zeroes * count`.
129
130    Example
131    -------
132    ```py
133    from sain import Bytes
134
135    buf = Bytes.from_str("Hello")
136    print(buf) # [72, 101, 108, 108, 111]
137    # buf is currently immutable, to make it mutable use `buf.to_mut()`
138    # the conversion costs nothing, as it just points to the same underlying array.
139    buf_mut = buf.to_mut()
140    buf_mut.put(32)
141    assert buf_mut == b"Hello "
142    ```
143    """
144
145    __slots__ = ("_buf",)
146
147    def __init__(self) -> None:
148        """Creates a new empty `Bytes`.
149
150        This won't allocate the array and the returned `Bytes` will be empty.
151        """
152        self._buf: array.array[int] | None = None
153
154    # construction
155
156    @classmethod
157    def from_str(cls, s: str) -> Bytes:
158        """Create a new `Bytes` from a utf-8 string.
159
160        Example
161        -------
162        ```py
163        buffer = Bytes.from_str("💀")
164        ```
165        """
166        b = cls()
167        b._buf = array.array("B", s.encode(ENCODING))
168        return b
169
170    @classmethod
171    def from_ptr(cls, arr: array.array[int]) -> Self:
172        """Create a new `Bytes` from an array.
173
174        The returned `Bytes` will directly point to `arr` without copying.
175
176        Example
177        -------
178        ```py
179        arr = array.array("B", b"Hello")
180        buffer = Bytes.from_ptr(arr)
181        ```
182        """
183        # this is technically an `assert` line
184        # but Python isn't smart enough to inline and opt-out
185        # this out of the generated bytecode.
186        # so we'll just leave this under `if` statement.
187        if __debug__:
188            assert_precondition(
189                arr.typecode == "B",
190                f"array type must be `B`, not `{arr.typecode}`",
191                TypeError,
192            )
193
194        b = cls()
195        b._buf = arr
196        return b
197
198    @classmethod
199    @unsafe
200    def from_ptr_unchecked(cls, arr: array.array[int]) -> Self:
201        """Create a new `Bytes` from an array, without checking the type code.
202
203        The returned `Bytes` will directly point to `arr` without copying.
204
205        ## Safety
206
207        The caller must ensure that `arr` is of type `array.array[int]` with type code `B`.
208
209        Example
210        -------
211        ```py
212        arr = array.array("B", b"Hello")
213        buffer = Bytes.from_ptr_unchecked(arr)
214        ```
215        """
216        b = cls()
217        b._buf = arr
218        return b
219
220    @classmethod
221    def from_bytes(cls, buf: Buffer) -> Self:
222        """Create a new `Bytes` from an initial bytes.
223
224        Example
225        -------
226        ```py
227        buffer = Bytes.from_bytes(b"SIGNATURE")
228        ```
229        """
230        b = cls()
231        b._buf = array.array("B", buf)
232        return b
233
234    @classmethod
235    def from_raw(cls, raw: Rawish) -> Self:
236        """Initialize a new `Bytes` from a `Rawish` data type.
237
238        Example
239        -------
240        ```py
241        with open('file.txt', 'rb') as file:
242            buff = Bytes.from_raw(file)
243
244        # in memory bytes io
245        bytes_io = io.BytesIO(b"data")
246        buffer1 = Bytes.from_raw(bytes_io)
247        # in memory string io
248        string_io = io.StringIO("data")
249        buffer2 = Bytes.from_raw(string_io)
250        ```
251        """
252        c = cls()
253        c._buf = array.array("B", unwrap_bytes(raw))
254        return c
255
256    @classmethod
257    def zeroed(cls, count: int) -> Self:
258        """Initialize a new `Bytes` filled with `0 * count`.
259
260        Example
261        -------
262        ```py
263        ALLOC_SIZE = 1024 * 2
264        buffer = Bytes.zeros(ALLOC_SIZE)
265        assert buffer.len() == ALLOC_SIZE
266        ```
267        """
268        c = cls()
269        c._buf = array.array("B", [0] * count)
270        return c
271
272    # buffer evolution
273
274    # These are getting deprecated because they're trivial.
275    # maybe we impl a `String` type and include them later.
276    # anyways, they won't be leaving for sometime until 2.0.0.
277
278    @deprecated(
279        since="1.3.0",
280        removed_in="2.0.0",
281        use_instead='Bytes.to_bytes().decode("utf8")',
282        hint="Converting a bytes object to string is fairly trivial.",
283    )
284    def to_string(self) -> str:
285        """Convert the bytes to a string.
286
287        Same as `Bytes.to_str`
288        """
289        return self.to_str()
290
291    @deprecated(
292        since="1.3.0",
293        removed_in="2.0.0",
294        use_instead='Bytes.to_bytes().decode("utf8")',
295        hint="Converting a bytes object to string is fairly trivial.",
296    )
297    def try_to_str(self) -> Result[str, bytes]:
298        """A safe method to convert `self` into a string.
299
300        This may fail if the `self` contains invalid bytes. strings
301        needs to be valid utf-8.
302
303        Example
304        -------
305        ```py
306        buf = Bytes()
307        sparkles_heart = [240, 159, 146, 150]
308        buf.put_bytes(sparkles_heart)
309
310        assert buf.try_to_str().unwrap() == "💖"
311        ```
312
313        Incorrect bytes
314        ---------------
315        ```py
316        invalid_bytes = Bytes.from_bytes([0, 159, 146, 150])
317        invalid_bytes.try_to_str().is_err()
318        ```
319
320        Returns
321        -------
322        `Result[str, bytes]`
323            If successful, returns the decoded string, otherwise the original bytes that failed
324            to get decoded.
325        """
326        try:
327            return _result.Ok(self.to_bytes().decode(ENCODING))
328        except UnicodeDecodeError as e:
329            return _result.Err(e.object)
330
331    @deprecated(
332        since="1.3.0",
333        removed_in="2.0.0",
334        use_instead='str(Bytes, encoding="utf-8")',
335        hint="Converting a bytes object to string is fairly trivial.",
336    )
337    def to_str(self) -> str:
338        r"""Convert `self` to a utf-8 string.
339
340        During the conversion process, any invalid bytes will get converted to
341        [REPLACEMENT_CHARACTER](https://en.wikipedia.org/wiki/Specials_(Unicode_block))
342        which looks like this `�`, so be careful on what you're trying to convert.
343
344        Use `.try_to_str` try attempt the conversion in case of failure.
345
346        Example
347        -------
348        ```py
349        buf = Bytes()
350        sparkles_heart = [240, 159, 146, 150]
351        buf.put_bytes(sparkles_heart)
352
353        assert buf.to_str() == "💖"
354        ```
355
356        Incorrect bytes
357        ---------------
358        ```py
359        invalid_bytes = Bytes.from_bytes(b"Hello \xf0\x90\x80World")
360        assert invalid_bytes.to_str() == "Hello �World"
361        ```
362        """
363        if not self._buf:
364            return ""
365
366        return self._buf.tobytes().decode(ENCODING, errors="replace")
367
368    def to_bytes(self) -> bytes:
369        """Convert `self` into `bytes`, copying the underlying array into a new buffer.
370
371        Example
372        -------
373        ```py
374        buf = Bytes.from_str("Hello")
375        assert buf.to_bytes() == b'Hello'
376        ```
377        """
378        if not self._buf:
379            return b""
380
381        return self._buf.tobytes()
382
383    def to_vec(self) -> _vec.Vec[int]:
384        """Copies `self` into a new `Vec`.
385
386        Example
387        -------
388        ```py
389        buffer = Bytes.from_str([1, 2, 3, 4])
390        # buffer and x can be modified independently.
391        x = buffer.to_vec()
392        """
393        return _vec.Vec(self.copy())
394
395    def leak(self) -> array.array[int]:
396        """Consumes and leaks the `Bytes`, returning the contents as an `array[int]`,
397
398        A new empty array is returned if the underlying buffer is not initialized.
399
400        `self` will deallocate the underlying array, therefore it becomes unusable.
401
402        Safety
403        ------
404        It is unsafe to access the leaked array from `self` after calling this function.
405
406        Example
407        -------
408        ```py
409        bytes = Bytes.from_str("chunks of data")
410        consumed = bytes.leak()
411        # `bytes` doesn't point to anything, this is undefined behavior.
412        bytes.put(0)
413        # access the array directly instead.
414        consumed.tobytes() == b"chunks of data"
415        ```
416        """
417        if self._buf is None:
418            return array.array("B")
419
420        arr = self._buf
421        # We don't need to reference this anymore since the caller will own the array.
422        del self._buf
423        return arr
424
425    def as_ptr(self) -> memoryview[int]:
426        """Returns a read-only pointer to the buffer data.
427
428        `pointer` here refers to a `memoryview` object.
429
430        A `BufferError` is raised if the underlying sequence is not initialized.
431
432        Example
433        -------
434        ```py
435        buffer = Bytes.from_bytes(b"data")
436        ptr = buffer.as_ptr()
437        ptr[0] = 1 # TypeError: cannot modify read-only memory
438        ```
439        """
440        return self.__buffer__(256).toreadonly()
441
442    def as_ref(self) -> _slice.Slice[int]:
443        """Get an immutable reference to the underlying sequence, without copying.
444
445        An empty slice is returned if the underlying sequence is not initialized.
446
447        Example
448        -------
449        ```py
450        async def send_multipart(buf: Sequence[int]) -> None:
451            ...
452
453        buffer = Bytes.from_bytes([0, 0, 0, 0])
454        await send_multipart(buffer.as_ref()) # no copy.
455        ```
456        """
457        if self._buf is not None:
458            return _slice.Slice(self)
459
460        return _slice.Slice(())
461
462    @safe
463    def to_mut(self) -> BytesMut:
464        """Convert `self` into `BytesMut`.
465
466        This consumes `self` and returns a new `BytesMut` that points to the same underlying array,
467        The conversion costs nothing.
468
469        Notes
470        -----
471        * If `self` is not initialized, a new empty `BytesMut` is returned.
472        * `self` will no longer be usable, as it will not point to the underlying array.
473
474        The inverse method for this is `BytesMut.freeze()`
475
476        Example
477        -------
478        ```py
479        def modify(buffer: Bytes) -> BytesMut:
480            buf = buffer.to_mut() # doesn't cost anything.
481            buf.swap(0, 1)
482            return buf
483
484        buffer = Bytes.from_bytes([1, 2, 3, 4])
485        new = modify(buffer)
486        assert new == [2, 1, 3, 4]
487        ```
488        """
489        # SAFETY: `Bytes.leak` returns an empty array
490        # if `self` is uninitialized.
491        return BytesMut.from_ptr_unchecked(self.leak())
492
493    def raw_parts(
494        self,
495    ) -> tuple[int, int]:
496        """Return `self` as tuple containing the memory address to the buffer and how many bytes it currently contains.
497
498        An alias to `array.buffer_into`
499        """
500        if not self._buf:
501            return (0x0, 0)
502
503        return self._buf.buffer_info()
504
505    def len(self) -> int:
506        """Return the number of bytes in this buffer.
507
508        Example
509        -------
510        ```py
511        buf = Bytes.from_bytes([240, 159, 146, 150])
512        assert buf.len() == 4
513        ```
514        """
515        return self.__len__()
516
517    def size(self) -> int:
518        """The length in bytes of one array item in the internal representation.
519
520
521        An alias to `array.itemsize`
522
523        Example
524        -------
525        ```py
526        buf = Bytes.from_bytes([240, 159, 146, 150])
527        assert buf.size() == 1
528        ```
529        """
530        if not self._buf:
531            return 0
532        return self._buf.itemsize
533
534    def iter(self) -> _iter.TrustedIter[int]:
535        """Returns an iterator over the contained bytes.
536
537        This iterator yields all `int`s from start to end.
538
539        Example
540        -------
541        ```py
542        buf = Bytes.from_bytes((1, 2, 3))
543        iterator = buf.iter()
544
545        # map each byte to a character
546        for element in iterator.map(chr):
547            print(element)
548        # ☺
549        # ☻
550        # ♥
551        ```
552        """
553        return _iter.TrustedIter(self.as_ptr())
554
555    def chars(self) -> Chars:
556        """Returns an iterator over the characters of `Bytes`.
557
558        This iterator yields all `int`s from start to end mapped as a `ctypes.c_char`.
559
560        Example
561        -------
562        ```py
563        b = Bytes.from_str("Hello")
564        for char in b.chars():
565            print(char)
566
567        # c_char(b'H')
568        # c_char(b'e')
569        # c_char(b'l')
570        # c_char(b'l')
571        # c_char(b'o')
572        ```
573        """
574        # The built-in map is actually faster than our own pure python adapter impl.
575        return _iter.Iter(map(_ctypes.c_char, self))
576
577    def is_empty(self) -> bool:
578        """Check whether `self` contains any bytes or not.
579
580        Example
581        -------
582        ```py
583        buffer = Bytes()
584        assert buffer.is_empty()
585        ```
586        """
587        return not self._buf
588
589    def split_off(self, at: int) -> Bytes:
590        """Split the bytes off at the specified position, returning a new
591        `Bytes` at the range of `[at : len]`, leaving `self` at `[at : bytes_len]`.
592
593        if this bytes is empty, `self` is returned unchanged.
594
595        Example
596        -------
597        ```py
598        origin = Bytes.from_bytes((1, 2, 3, 4))
599        split = origin.split_off(2)
600
601        print(origin, split)  # [1, 2], [3, 4]
602        ```
603
604        Raises
605        ------
606        `RuntimeError`
607            This method will raise if `at` > `len(self)`
608        """
609        len_ = self.len()
610        if at > len_:
611            raise RuntimeError(
612                f"Index `at` ({at}) should be <= than len of `self` ({len_}) "
613            ) from None
614
615        # Either the list is empty or uninit.
616        if not self._buf:
617            return self
618
619        split = self[at:len_]  # split the items into a new buffer.
620        del self._buf[at:len_]  # remove the items from the original list.
621        return split
622
623    def split_first(self) -> Option[tuple[int, Bytes]]:
624        """Split the first and rest elements of the bytes, If empty, `None` is returned.
625
626        Example
627        -------
628        ```py
629        buf = Bytes.from_bytes([1, 2, 3])
630        split = buf.split_first()
631        assert split == Some((1, [2, 3]))
632        ```
633        """
634        if not self._buf:
635            return _option.NOTHING
636
637        # optimized to only one element in the buffer.
638        if self.len() == 1:
639            return _option.Some((self[0], Bytes()))
640
641        first, rest = self[0], self[1:]
642        return _option.Some((first, rest))
643
644    def split_last(self) -> Option[tuple[int, Bytes]]:
645        """Returns the last and rest of the elements of the bytes, If `self` is empty, `None` is returned.
646
647        Example
648        -------
649        ```py
650        buf = Bytes.from_bytes([0, 1, 2])
651        last, elements = buf.split_last().unwrap()
652        assert (last, elements) == (3, [1, 2])
653        ```
654        """
655        if not self._buf:
656            return _option.NOTHING
657
658        len_ = self.len()
659        # optimized to only one element in the buffer.
660        if len_ == 1:
661            return _option.Some((self[0], Bytes()))
662
663        last, rest = self[-1], self[:-1]
664        return _option.Some((last, rest))
665
666    def split_at(self, mid: int) -> tuple[Bytes, Bytes]:
667        """Divide `self` into two at an index.
668
669        The first will contain all bytes from `[0:mid]` excluding `mid` it self.
670        and the second will contain the remaining bytes.
671
672        if `mid` > `self.len()`, Then all bytes will be moved to the left,
673        returning an empty bytes in right.
674
675        Example
676        -------
677        ```py
678        buffer = Bytes.from_bytes((1, 2, 3, 4))
679        left, right = buffer.split_at(0)
680        assert left == [] and right == [1, 2, 3, 4]
681
682        left, right = buffer.split_at(2)
683        assert left == [1, 2] and right == [2, 3]
684        ```
685
686        The is roughly the implementation
687        ```py
688        self[0:mid], self[mid:]
689        ```
690        """
691        return self[0:mid], self[mid:]
692
693    # layout methods.
694
695    @safe
696    def copy(self) -> Bytes:
697        """Create a copy of the bytes.
698
699        Example
700        -------
701        ```py
702        original = Bytes.from_bytes([255, 255, 255, 0])
703        copy = original.copy()
704        ```
705        """
706        if not self._buf:
707            return Bytes()
708
709        # SAFETY: `self._buf` is initialized.
710        return self.from_ptr_unchecked(self._buf[:])
711
712    def index(self, v: int, start: int = 0, stop: int = _sys.maxsize) -> int:
713        """Return the smallest `i` such that `i` is the index of the first occurrence of `v` in the buffer.
714
715        The optional arguments start and stop can be specified to search for x within a
716        subsection of the array. Raise ValueError if x is not found
717        """
718        if not self._buf:
719            raise ValueError from None
720
721        return self._buf.index(v, start, stop)
722
723    def count(self, x: int) -> int:
724        """Return the number of occurrences of `x` in the buffer.
725
726        Example
727        --------
728        ```py
729        buf = Bytes([32, 32, 31])
730        assert buf.count(32) == 2
731        ```
732        """
733        if self._buf is None:
734            return 0
735
736        return self._buf.count(x)
737
738    # special methods
739
740    def __iter__(self) -> collections.Iterator[int]:
741        if self._buf:
742            return self._buf.__iter__()
743
744        return ().__iter__()
745
746    def __len__(self) -> int:
747        return len(self._buf) if self._buf else 0
748
749    def __repr__(self) -> str:
750        if not self._buf:
751            return "[]"
752
753        return "[" + ", ".join(str(x) for x in self._buf) + "]"
754
755    __str__ = __repr__
756
757    def __bytes__(self) -> bytes:
758        return self.to_bytes()
759
760    def __buffer__(self, flag: int | inspect.BufferFlags) -> memoryview[int]:
761        if not self._buf:
762            raise BufferError("Cannot work with uninitialized bytes.")
763
764        if _sys.version_info >= (3, 12):
765            mem = self._buf.__buffer__(flag)
766        else:
767            # arrays in 3.11 and under don't implement the buffer protocol.
768            mem = memoryview(self._buf)
769
770        return mem
771
772    def __contains__(self, byte: int) -> bool:
773        return byte in self._buf if self._buf else False
774
775    def __eq__(self, other: object, /) -> bool:
776        if not self._buf:
777            return False
778
779        if isinstance(other, bytes):
780            return self._buf.tobytes() == other
781
782        # bytes IS a `Sequence[int]`, but not all `Sequence[int]`
783        # represented as bytes.
784        elif isinstance(other, collections.Sequence):
785            return self._buf.tolist() == other
786
787        return self._buf.__eq__(other)
788
789    def __ne__(self, other: object, /) -> bool:
790        return not self.__eq__(other)
791
792    def __le__(self, other: object) -> bool:
793        if not self._buf:
794            return False
795
796        if not isinstance(other, Bytes):
797            return NotImplemented
798
799        if not other._buf:
800            return False
801
802        return self._buf <= other._buf
803
804    def __ge__(self, other: object) -> bool:
805        if not self._buf:
806            return False
807
808        if not isinstance(other, Bytes):
809            return NotImplemented
810
811        if not other._buf:
812            return False
813
814        return self._buf >= other._buf
815
816    def __lt__(self, other: object) -> bool:
817        if not self._buf:
818            return False
819
820        if not isinstance(other, Bytes):
821            return NotImplemented
822
823        if not other._buf:
824            return False
825
826        return self._buf < other._buf
827
828    def __gt__(self, other: object) -> bool:
829        if not self._buf:
830            return False
831
832        if not isinstance(other, Bytes):
833            return NotImplemented
834
835        if not other._buf:
836            return False
837
838        return self._buf > other._buf
839
840    @typing.overload
841    def __getitem__(self, index: slice) -> Bytes: ...
842
843    @typing.overload
844    def __getitem__(self, index: int) -> int: ...
845
846    @safe
847    def __getitem__(self, index: int | slice) -> int | Bytes:
848        if not self._buf:
849            raise IndexError("Index out of range")
850
851        if isinstance(index, slice):
852            # SAFETY: `self._buf` is initialized.
853            return self.from_ptr_unchecked(self._buf[index])
854
855        return self._buf[index]
856
857    def __reversed__(self) -> collections.Iterator[int]:
858        return reversed(self._buf or ())
859
860    # defined like `array`'s
861    __hash__: typing.ClassVar[None] = None
862
863    @safe
864    def __copy__(self) -> Bytes:
865        if not self._buf:
866            return Bytes()
867
868        return Bytes.from_ptr_unchecked(self._buf.__copy__())
869
870    @safe
871    def __deepcopy__(self, unused: typing.Any, /) -> Bytes:
872        if not self._buf:
873            return Bytes()
874
875        return Bytes.from_ptr_unchecked(self._buf.__deepcopy__(unused))

Provides immutable abstractions for working with bytes.

It is an efficient container for storing and operating with bytes. It behaves very much like array.array[int] as well has the same layout.

A Bytes objects are usually used within networking applications, but can also be used elsewhere as well.

Construction

Bytes object accept multiple rawish data types, See Rawish for all supported types.

Example
from sain import Bytes

buf = Bytes.from_str("Hello")
print(buf) # [72, 101, 108, 108, 111]
# buf is currently immutable, to make it mutable use `buf.to_mut()`
# the conversion costs nothing, as it just points to the same underlying array.
buf_mut = buf.to_mut()
buf_mut.put(32)
assert buf_mut == b"Hello "

Implementations

This class implements &[u8] in Rust.

@classmethod
def from_str(cls, s: str) -> Bytes:
156    @classmethod
157    def from_str(cls, s: str) -> Bytes:
158        """Create a new `Bytes` from a utf-8 string.
159
160        Example
161        -------
162        ```py
163        buffer = Bytes.from_str("💀")
164        ```
165        """
166        b = cls()
167        b._buf = array.array("B", s.encode(ENCODING))
168        return b

Create a new Bytes from a utf-8 string.

Example
buffer = Bytes.from_str("💀")
@classmethod
def from_ptr(cls, arr: array.array[int]) -> Self:
170    @classmethod
171    def from_ptr(cls, arr: array.array[int]) -> Self:
172        """Create a new `Bytes` from an array.
173
174        The returned `Bytes` will directly point to `arr` without copying.
175
176        Example
177        -------
178        ```py
179        arr = array.array("B", b"Hello")
180        buffer = Bytes.from_ptr(arr)
181        ```
182        """
183        # this is technically an `assert` line
184        # but Python isn't smart enough to inline and opt-out
185        # this out of the generated bytecode.
186        # so we'll just leave this under `if` statement.
187        if __debug__:
188            assert_precondition(
189                arr.typecode == "B",
190                f"array type must be `B`, not `{arr.typecode}`",
191                TypeError,
192            )
193
194        b = cls()
195        b._buf = arr
196        return b

Create a new Bytes from an array.

The returned Bytes will directly point to arr without copying.

Example
arr = array.array("B", b"Hello")
buffer = Bytes.from_ptr(arr)
@classmethod
@unsafe
def from_ptr_unchecked(cls, arr: array.array[int]) -> Self:
198    @classmethod
199    @unsafe
200    def from_ptr_unchecked(cls, arr: array.array[int]) -> Self:
201        """Create a new `Bytes` from an array, without checking the type code.
202
203        The returned `Bytes` will directly point to `arr` without copying.
204
205        ## Safety
206
207        The caller must ensure that `arr` is of type `array.array[int]` with type code `B`.
208
209        Example
210        -------
211        ```py
212        arr = array.array("B", b"Hello")
213        buffer = Bytes.from_ptr_unchecked(arr)
214        ```
215        """
216        b = cls()
217        b._buf = arr
218        return b

Create a new Bytes from an array, without checking the type code.

The returned Bytes will directly point to arr without copying.

Safety

The caller must ensure that arr is of type array.array[int] with type code B.

Example
arr = array.array("B", b"Hello")
buffer = Bytes.from_ptr_unchecked(arr)

Safety ⚠️

Calling this method without knowing the output is considered undefined behavior.

@classmethod
def from_bytes(cls, buf: bytes | bytearray | Iterable[int]) -> Self:
220    @classmethod
221    def from_bytes(cls, buf: Buffer) -> Self:
222        """Create a new `Bytes` from an initial bytes.
223
224        Example
225        -------
226        ```py
227        buffer = Bytes.from_bytes(b"SIGNATURE")
228        ```
229        """
230        b = cls()
231        b._buf = array.array("B", buf)
232        return b

Create a new Bytes from an initial bytes.

Example
buffer = Bytes.from_bytes(b"SIGNATURE")
@classmethod
def from_raw(cls, raw: _io.StringIO | _io.BytesIO | _io.BufferedReader) -> Self:
234    @classmethod
235    def from_raw(cls, raw: Rawish) -> Self:
236        """Initialize a new `Bytes` from a `Rawish` data type.
237
238        Example
239        -------
240        ```py
241        with open('file.txt', 'rb') as file:
242            buff = Bytes.from_raw(file)
243
244        # in memory bytes io
245        bytes_io = io.BytesIO(b"data")
246        buffer1 = Bytes.from_raw(bytes_io)
247        # in memory string io
248        string_io = io.StringIO("data")
249        buffer2 = Bytes.from_raw(string_io)
250        ```
251        """
252        c = cls()
253        c._buf = array.array("B", unwrap_bytes(raw))
254        return c

Initialize a new Bytes from a Rawish data type.

Example
with open('file.txt', 'rb') as file:
    buff = Bytes.from_raw(file)

# in memory bytes io
bytes_io = io.BytesIO(b"data")
buffer1 = Bytes.from_raw(bytes_io)
# in memory string io
string_io = io.StringIO("data")
buffer2 = Bytes.from_raw(string_io)
@classmethod
def zeroed(cls, count: int) -> Self:
256    @classmethod
257    def zeroed(cls, count: int) -> Self:
258        """Initialize a new `Bytes` filled with `0 * count`.
259
260        Example
261        -------
262        ```py
263        ALLOC_SIZE = 1024 * 2
264        buffer = Bytes.zeros(ALLOC_SIZE)
265        assert buffer.len() == ALLOC_SIZE
266        ```
267        """
268        c = cls()
269        c._buf = array.array("B", [0] * count)
270        return c

Initialize a new Bytes filled with 0 * count.

Example
ALLOC_SIZE = 1024 * 2
buffer = Bytes.zeros(ALLOC_SIZE)
assert buffer.len() == ALLOC_SIZE
@deprecated(since='1.3.0', removed_in='2.0.0', use_instead='Bytes.to_bytes().decode("utf8")', hint='Converting a bytes object to string is fairly trivial.')
def try_to_str(self) -> 'Result[str, bytes]':
291    @deprecated(
292        since="1.3.0",
293        removed_in="2.0.0",
294        use_instead='Bytes.to_bytes().decode("utf8")',
295        hint="Converting a bytes object to string is fairly trivial.",
296    )
297    def try_to_str(self) -> Result[str, bytes]:
298        """A safe method to convert `self` into a string.
299
300        This may fail if the `self` contains invalid bytes. strings
301        needs to be valid utf-8.
302
303        Example
304        -------
305        ```py
306        buf = Bytes()
307        sparkles_heart = [240, 159, 146, 150]
308        buf.put_bytes(sparkles_heart)
309
310        assert buf.try_to_str().unwrap() == "💖"
311        ```
312
313        Incorrect bytes
314        ---------------
315        ```py
316        invalid_bytes = Bytes.from_bytes([0, 159, 146, 150])
317        invalid_bytes.try_to_str().is_err()
318        ```
319
320        Returns
321        -------
322        `Result[str, bytes]`
323            If successful, returns the decoded string, otherwise the original bytes that failed
324            to get decoded.
325        """
326        try:
327            return _result.Ok(self.to_bytes().decode(ENCODING))
328        except UnicodeDecodeError as e:
329            return _result.Err(e.object)

A safe method to convert self into a string.

This may fail if the self contains invalid bytes. strings needs to be valid utf-8.

Example
buf = Bytes()
sparkles_heart = [240, 159, 146, 150]
buf.put_bytes(sparkles_heart)

assert buf.try_to_str().unwrap() == "💖"
Incorrect bytes
invalid_bytes = Bytes.from_bytes([0, 159, 146, 150])
invalid_bytes.try_to_str().is_err()
Returns
  • Result[str, bytes]: If successful, returns the decoded string, otherwise the original bytes that failed to get decoded.
@deprecated(since='1.3.0', removed_in='2.0.0', use_instead='str(Bytes, encoding="utf-8")', hint='Converting a bytes object to string is fairly trivial.')
def to_str(self) -> str:
331    @deprecated(
332        since="1.3.0",
333        removed_in="2.0.0",
334        use_instead='str(Bytes, encoding="utf-8")',
335        hint="Converting a bytes object to string is fairly trivial.",
336    )
337    def to_str(self) -> str:
338        r"""Convert `self` to a utf-8 string.
339
340        During the conversion process, any invalid bytes will get converted to
341        [REPLACEMENT_CHARACTER](https://en.wikipedia.org/wiki/Specials_(Unicode_block))
342        which looks like this `�`, so be careful on what you're trying to convert.
343
344        Use `.try_to_str` try attempt the conversion in case of failure.
345
346        Example
347        -------
348        ```py
349        buf = Bytes()
350        sparkles_heart = [240, 159, 146, 150]
351        buf.put_bytes(sparkles_heart)
352
353        assert buf.to_str() == "💖"
354        ```
355
356        Incorrect bytes
357        ---------------
358        ```py
359        invalid_bytes = Bytes.from_bytes(b"Hello \xf0\x90\x80World")
360        assert invalid_bytes.to_str() == "Hello �World"
361        ```
362        """
363        if not self._buf:
364            return ""
365
366        return self._buf.tobytes().decode(ENCODING, errors="replace")

Convert self to a utf-8 string.

During the conversion process, any invalid bytes will get converted to REPLACEMENT_CHARACTER which looks like this , so be careful on what you're trying to convert.

Use .try_to_str try attempt the conversion in case of failure.

Example
buf = Bytes()
sparkles_heart = [240, 159, 146, 150]
buf.put_bytes(sparkles_heart)

assert buf.to_str() == "💖"
Incorrect bytes
invalid_bytes = Bytes.from_bytes(b"Hello \xf0\x90\x80World")
assert invalid_bytes.to_str() == "Hello �World"
def to_bytes(self) -> bytes:
368    def to_bytes(self) -> bytes:
369        """Convert `self` into `bytes`, copying the underlying array into a new buffer.
370
371        Example
372        -------
373        ```py
374        buf = Bytes.from_str("Hello")
375        assert buf.to_bytes() == b'Hello'
376        ```
377        """
378        if not self._buf:
379            return b""
380
381        return self._buf.tobytes()

Convert self into bytes, copying the underlying array into a new buffer.

Example
buf = Bytes.from_str("Hello")
assert buf.to_bytes() == b'Hello'
def to_vec(self) -> sain.Vec[int]:
383    def to_vec(self) -> _vec.Vec[int]:
384        """Copies `self` into a new `Vec`.
385
386        Example
387        -------
388        ```py
389        buffer = Bytes.from_str([1, 2, 3, 4])
390        # buffer and x can be modified independently.
391        x = buffer.to_vec()
392        """
393        return _vec.Vec(self.copy())

Copies self into a new Vec.

Example

```py buffer = Bytes.from_str([1, 2, 3, 4])

buffer and x can be modified independently.

x = buffer.to_vec()

def leak(self) -> array.array[int]:
395    def leak(self) -> array.array[int]:
396        """Consumes and leaks the `Bytes`, returning the contents as an `array[int]`,
397
398        A new empty array is returned if the underlying buffer is not initialized.
399
400        `self` will deallocate the underlying array, therefore it becomes unusable.
401
402        Safety
403        ------
404        It is unsafe to access the leaked array from `self` after calling this function.
405
406        Example
407        -------
408        ```py
409        bytes = Bytes.from_str("chunks of data")
410        consumed = bytes.leak()
411        # `bytes` doesn't point to anything, this is undefined behavior.
412        bytes.put(0)
413        # access the array directly instead.
414        consumed.tobytes() == b"chunks of data"
415        ```
416        """
417        if self._buf is None:
418            return array.array("B")
419
420        arr = self._buf
421        # We don't need to reference this anymore since the caller will own the array.
422        del self._buf
423        return arr

Consumes and leaks the Bytes, returning the contents as an array[int],

A new empty array is returned if the underlying buffer is not initialized.

self will deallocate the underlying array, therefore it becomes unusable.

Safety

It is unsafe to access the leaked array from self after calling this function.

Example
bytes = Bytes.from_str("chunks of data")
consumed = bytes.leak()
# `bytes` doesn't point to anything, this is undefined behavior.
bytes.put(0)
# access the array directly instead.
consumed.tobytes() == b"chunks of data"
def as_ptr(self) -> 'memoryview[int]':
425    def as_ptr(self) -> memoryview[int]:
426        """Returns a read-only pointer to the buffer data.
427
428        `pointer` here refers to a `memoryview` object.
429
430        A `BufferError` is raised if the underlying sequence is not initialized.
431
432        Example
433        -------
434        ```py
435        buffer = Bytes.from_bytes(b"data")
436        ptr = buffer.as_ptr()
437        ptr[0] = 1 # TypeError: cannot modify read-only memory
438        ```
439        """
440        return self.__buffer__(256).toreadonly()

Returns a read-only pointer to the buffer data.

pointer here refers to a memoryview object.

A BufferError is raised if the underlying sequence is not initialized.

Example
buffer = Bytes.from_bytes(b"data")
ptr = buffer.as_ptr()
ptr[0] = 1 # TypeError: cannot modify read-only memory
def as_ref(self) -> sain.collections.slice.Slice[int]:
442    def as_ref(self) -> _slice.Slice[int]:
443        """Get an immutable reference to the underlying sequence, without copying.
444
445        An empty slice is returned if the underlying sequence is not initialized.
446
447        Example
448        -------
449        ```py
450        async def send_multipart(buf: Sequence[int]) -> None:
451            ...
452
453        buffer = Bytes.from_bytes([0, 0, 0, 0])
454        await send_multipart(buffer.as_ref()) # no copy.
455        ```
456        """
457        if self._buf is not None:
458            return _slice.Slice(self)
459
460        return _slice.Slice(())

Get an immutable reference to the underlying sequence, without copying.

An empty slice is returned if the underlying sequence is not initialized.

Example
async def send_multipart(buf: Sequence[int]) -> None:
    ...

buffer = Bytes.from_bytes([0, 0, 0, 0])
await send_multipart(buffer.as_ref()) # no copy.
@safe
def to_mut(self) -> BytesMut:
462    @safe
463    def to_mut(self) -> BytesMut:
464        """Convert `self` into `BytesMut`.
465
466        This consumes `self` and returns a new `BytesMut` that points to the same underlying array,
467        The conversion costs nothing.
468
469        Notes
470        -----
471        * If `self` is not initialized, a new empty `BytesMut` is returned.
472        * `self` will no longer be usable, as it will not point to the underlying array.
473
474        The inverse method for this is `BytesMut.freeze()`
475
476        Example
477        -------
478        ```py
479        def modify(buffer: Bytes) -> BytesMut:
480            buf = buffer.to_mut() # doesn't cost anything.
481            buf.swap(0, 1)
482            return buf
483
484        buffer = Bytes.from_bytes([1, 2, 3, 4])
485        new = modify(buffer)
486        assert new == [2, 1, 3, 4]
487        ```
488        """
489        # SAFETY: `Bytes.leak` returns an empty array
490        # if `self` is uninitialized.
491        return BytesMut.from_ptr_unchecked(self.leak())

Convert self into BytesMut.

This consumes self and returns a new BytesMut that points to the same underlying array, The conversion costs nothing.

Notes
  • If self is not initialized, a new empty BytesMut is returned.
  • self will no longer be usable, as it will not point to the underlying array.

The inverse method for this is BytesMut.freeze()

Example
def modify(buffer: Bytes) -> BytesMut:
    buf = buffer.to_mut() # doesn't cost anything.
    buf.swap(0, 1)
    return buf

buffer = Bytes.from_bytes([1, 2, 3, 4])
new = modify(buffer)
assert new == [2, 1, 3, 4]
def raw_parts(self) -> tuple[int, int]:
493    def raw_parts(
494        self,
495    ) -> tuple[int, int]:
496        """Return `self` as tuple containing the memory address to the buffer and how many bytes it currently contains.
497
498        An alias to `array.buffer_into`
499        """
500        if not self._buf:
501            return (0x0, 0)
502
503        return self._buf.buffer_info()

Return self as tuple containing the memory address to the buffer and how many bytes it currently contains.

An alias to array.buffer_into

def len(self) -> int:
505    def len(self) -> int:
506        """Return the number of bytes in this buffer.
507
508        Example
509        -------
510        ```py
511        buf = Bytes.from_bytes([240, 159, 146, 150])
512        assert buf.len() == 4
513        ```
514        """
515        return self.__len__()

Return the number of bytes in this buffer.

Example
buf = Bytes.from_bytes([240, 159, 146, 150])
assert buf.len() == 4
def size(self) -> int:
517    def size(self) -> int:
518        """The length in bytes of one array item in the internal representation.
519
520
521        An alias to `array.itemsize`
522
523        Example
524        -------
525        ```py
526        buf = Bytes.from_bytes([240, 159, 146, 150])
527        assert buf.size() == 1
528        ```
529        """
530        if not self._buf:
531            return 0
532        return self._buf.itemsize

The length in bytes of one array item in the internal representation.

An alias to array.itemsize

Example
buf = Bytes.from_bytes([240, 159, 146, 150])
assert buf.size() == 1
def iter(self) -> sain.iter.TrustedIter[int]:
534    def iter(self) -> _iter.TrustedIter[int]:
535        """Returns an iterator over the contained bytes.
536
537        This iterator yields all `int`s from start to end.
538
539        Example
540        -------
541        ```py
542        buf = Bytes.from_bytes((1, 2, 3))
543        iterator = buf.iter()
544
545        # map each byte to a character
546        for element in iterator.map(chr):
547            print(element)
548        # ☺
549        # ☻
550        # ♥
551        ```
552        """
553        return _iter.TrustedIter(self.as_ptr())

Returns an iterator over the contained bytes.

This iterator yields all ints from start to end.

Example
buf = Bytes.from_bytes((1, 2, 3))
iterator = buf.iter()

# map each byte to a character
for element in iterator.map(chr):
    print(element)
# ☺
# ☻
# ♥
def chars(self) -> sain.Iterator[ctypes.c_char]:
555    def chars(self) -> Chars:
556        """Returns an iterator over the characters of `Bytes`.
557
558        This iterator yields all `int`s from start to end mapped as a `ctypes.c_char`.
559
560        Example
561        -------
562        ```py
563        b = Bytes.from_str("Hello")
564        for char in b.chars():
565            print(char)
566
567        # c_char(b'H')
568        # c_char(b'e')
569        # c_char(b'l')
570        # c_char(b'l')
571        # c_char(b'o')
572        ```
573        """
574        # The built-in map is actually faster than our own pure python adapter impl.
575        return _iter.Iter(map(_ctypes.c_char, self))

Returns an iterator over the characters of Bytes.

This iterator yields all ints from start to end mapped as a ctypes.c_char.

Example
b = Bytes.from_str("Hello")
for char in b.chars():
    print(char)

# c_char(b'H')
# c_char(b'e')
# c_char(b'l')
# c_char(b'l')
# c_char(b'o')
def is_empty(self) -> bool:
577    def is_empty(self) -> bool:
578        """Check whether `self` contains any bytes or not.
579
580        Example
581        -------
582        ```py
583        buffer = Bytes()
584        assert buffer.is_empty()
585        ```
586        """
587        return not self._buf

Check whether self contains any bytes or not.

Example
buffer = Bytes()
assert buffer.is_empty()
def split_off(self, at: int) -> Bytes:
589    def split_off(self, at: int) -> Bytes:
590        """Split the bytes off at the specified position, returning a new
591        `Bytes` at the range of `[at : len]`, leaving `self` at `[at : bytes_len]`.
592
593        if this bytes is empty, `self` is returned unchanged.
594
595        Example
596        -------
597        ```py
598        origin = Bytes.from_bytes((1, 2, 3, 4))
599        split = origin.split_off(2)
600
601        print(origin, split)  # [1, 2], [3, 4]
602        ```
603
604        Raises
605        ------
606        `RuntimeError`
607            This method will raise if `at` > `len(self)`
608        """
609        len_ = self.len()
610        if at > len_:
611            raise RuntimeError(
612                f"Index `at` ({at}) should be <= than len of `self` ({len_}) "
613            ) from None
614
615        # Either the list is empty or uninit.
616        if not self._buf:
617            return self
618
619        split = self[at:len_]  # split the items into a new buffer.
620        del self._buf[at:len_]  # remove the items from the original list.
621        return split

Split the bytes off at the specified position, returning a new Bytes at the range of [at : len], leaving self at [at : bytes_len].

if this bytes is empty, self is returned unchanged.

Example
origin = Bytes.from_bytes((1, 2, 3, 4))
split = origin.split_off(2)

print(origin, split)  # [1, 2], [3, 4]
Raises
  • RuntimeError: This method will raise if at > len(self)
def split_first(self) -> 'Option[tuple[int, Bytes]]':
623    def split_first(self) -> Option[tuple[int, Bytes]]:
624        """Split the first and rest elements of the bytes, If empty, `None` is returned.
625
626        Example
627        -------
628        ```py
629        buf = Bytes.from_bytes([1, 2, 3])
630        split = buf.split_first()
631        assert split == Some((1, [2, 3]))
632        ```
633        """
634        if not self._buf:
635            return _option.NOTHING
636
637        # optimized to only one element in the buffer.
638        if self.len() == 1:
639            return _option.Some((self[0], Bytes()))
640
641        first, rest = self[0], self[1:]
642        return _option.Some((first, rest))

Split the first and rest elements of the bytes, If empty, None is returned.

Example
buf = Bytes.from_bytes([1, 2, 3])
split = buf.split_first()
assert split == Some((1, [2, 3]))
def split_last(self) -> 'Option[tuple[int, Bytes]]':
644    def split_last(self) -> Option[tuple[int, Bytes]]:
645        """Returns the last and rest of the elements of the bytes, If `self` is empty, `None` is returned.
646
647        Example
648        -------
649        ```py
650        buf = Bytes.from_bytes([0, 1, 2])
651        last, elements = buf.split_last().unwrap()
652        assert (last, elements) == (3, [1, 2])
653        ```
654        """
655        if not self._buf:
656            return _option.NOTHING
657
658        len_ = self.len()
659        # optimized to only one element in the buffer.
660        if len_ == 1:
661            return _option.Some((self[0], Bytes()))
662
663        last, rest = self[-1], self[:-1]
664        return _option.Some((last, rest))

Returns the last and rest of the elements of the bytes, If self is empty, None is returned.

Example
buf = Bytes.from_bytes([0, 1, 2])
last, elements = buf.split_last().unwrap()
assert (last, elements) == (3, [1, 2])
def split_at( self, mid: int) -> tuple[Bytes, Bytes]:
666    def split_at(self, mid: int) -> tuple[Bytes, Bytes]:
667        """Divide `self` into two at an index.
668
669        The first will contain all bytes from `[0:mid]` excluding `mid` it self.
670        and the second will contain the remaining bytes.
671
672        if `mid` > `self.len()`, Then all bytes will be moved to the left,
673        returning an empty bytes in right.
674
675        Example
676        -------
677        ```py
678        buffer = Bytes.from_bytes((1, 2, 3, 4))
679        left, right = buffer.split_at(0)
680        assert left == [] and right == [1, 2, 3, 4]
681
682        left, right = buffer.split_at(2)
683        assert left == [1, 2] and right == [2, 3]
684        ```
685
686        The is roughly the implementation
687        ```py
688        self[0:mid], self[mid:]
689        ```
690        """
691        return self[0:mid], self[mid:]

Divide self into two at an index.

The first will contain all bytes from [0:mid] excluding mid it self. and the second will contain the remaining bytes.

if mid > self.len(), Then all bytes will be moved to the left, returning an empty bytes in right.

Example
buffer = Bytes.from_bytes((1, 2, 3, 4))
left, right = buffer.split_at(0)
assert left == [] and right == [1, 2, 3, 4]

left, right = buffer.split_at(2)
assert left == [1, 2] and right == [2, 3]

The is roughly the implementation

self[0:mid], self[mid:]
@safe
def copy(self) -> Bytes:
695    @safe
696    def copy(self) -> Bytes:
697        """Create a copy of the bytes.
698
699        Example
700        -------
701        ```py
702        original = Bytes.from_bytes([255, 255, 255, 0])
703        copy = original.copy()
704        ```
705        """
706        if not self._buf:
707            return Bytes()
708
709        # SAFETY: `self._buf` is initialized.
710        return self.from_ptr_unchecked(self._buf[:])

Create a copy of the bytes.

Example
original = Bytes.from_bytes([255, 255, 255, 0])
copy = original.copy()
def index(self, v: int, start: int = 0, stop: int = 9223372036854775807) -> int:
712    def index(self, v: int, start: int = 0, stop: int = _sys.maxsize) -> int:
713        """Return the smallest `i` such that `i` is the index of the first occurrence of `v` in the buffer.
714
715        The optional arguments start and stop can be specified to search for x within a
716        subsection of the array. Raise ValueError if x is not found
717        """
718        if not self._buf:
719            raise ValueError from None
720
721        return self._buf.index(v, start, stop)

Return the smallest i such that i is the index of the first occurrence of v in the buffer.

The optional arguments start and stop can be specified to search for x within a subsection of the array. Raise ValueError if x is not found

def count(self, x: int) -> int:
723    def count(self, x: int) -> int:
724        """Return the number of occurrences of `x` in the buffer.
725
726        Example
727        --------
728        ```py
729        buf = Bytes([32, 32, 31])
730        assert buf.count(32) == 2
731        ```
732        """
733        if self._buf is None:
734            return 0
735
736        return self._buf.count(x)

Return the number of occurrences of x in the buffer.

Example
buf = Bytes([32, 32, 31])
assert buf.count(32) == 2
@rustc_diagnostic_item('&mut [u8]')
@typing.final
class BytesMut(Bytes, collections.abc.MutableSequence[int]):
 878@rustc_diagnostic_item("&mut [u8]")
 879@typing.final
 880class BytesMut(
 881    Bytes,  # pyright: ignore - we want to inherit from `Bytes`.
 882    collections.MutableSequence[int],
 883):
 884    """Provides mutable abstractions for working with bytes.
 885
 886    It is an efficient container for storing and operating with bytes,
 887    It is built on-top of `array.array[int]`, which means you get all of `array[int]`'s operations.
 888
 889    A `bytes` object is usually used within networking applications, but can also be used
 890    elsewhere as well.
 891
 892    ## Construction
 893    You can create a `BytesMut` object in multiple ways.
 894
 895    * `BytesMut()`: Initialize an empty `BytesMut` object
 896    * `from_str`: Create `BytesMut` from `str`
 897    * `from_bytes`: Create `BytesMut` from a `Buffer` bytes-like type
 898    * `from_raw`: Create `BytesMut` from a `Rawish` type
 899    * `from_ptr`: Create `BytesMut` that points to an `array.array[int]` without copying it
 900    * `BytesMut.zeroed(count)`: Create `BytesMut` filled with `zeroes * count`.
 901
 902    Example
 903    -------
 904    ```py
 905    from sain import BytesNut
 906
 907    buf = BytesMut()
 908    buffer.put_bytes(b"Hello")
 909    print(buffer) # [72, 101, 108, 108, 111]
 910
 911    buf.put(32) # space
 912    assert buffer.to_bytes() == b"Hello "
 913    ```
 914    """
 915
 916    __slots__ = ("_buf",)
 917
 918    def __init__(self) -> None:
 919        """Creates a new empty `BytesMut`.
 920
 921        This won't allocate the array and the returned `BytesMut` will be empty.
 922        """
 923        super().__init__()
 924
 925    # default methods.
 926
 927    def extend(self, src: Buffer) -> None:
 928        """Extend `self` from a `src`.
 929
 930        Example
 931        -------
 932        ```py
 933        buf = Bytes()
 934        buf.extend([1, 2, 3])
 935        assert buf == [1, 2, 3]
 936        ```
 937
 938        Parameters
 939        ----------
 940        src : `Buffer`
 941            Can be one of `Bytes`, `bytes`, `bytearray` or `Sequence[int]`
 942
 943        Raises
 944        ------
 945        `OverflowError`
 946            If `src` not in range of `0..255`
 947        """
 948        if self._buf is None:
 949            # If it was `None`, we initialize it with a source instead of extending.
 950            self._buf = array.array("B", src)
 951        else:
 952            self._buf.extend(src)
 953
 954    def put(self, src: int) -> None:
 955        """Append a byte at the end of the array.
 956
 957        unlike `.put_bytes`, this method appends instead of extending the array
 958        which is faster if you're putting a single byte in a single call.
 959
 960        Example
 961        -------
 962        ```py
 963        buf = Bytes()
 964        buf.put(32) # append a space to the end of the buffer
 965        assert buf.to_bytes() == b' '
 966        ```
 967
 968        Parameters
 969        ----------
 970        src : `int`
 971            An unsigned integer, also known as `u8` to put.
 972
 973        Raises
 974        ------
 975        `OverflowError`
 976            If `src` not in range of `0..255`
 977        """
 978        if self._buf is None:
 979            # If it was `None`, we initialize it with a source instead of appending.
 980            self._buf = array.array("B", [src])
 981        else:
 982            self._buf.append(src)
 983
 984    def put_float(self, src: float) -> None:
 985        r"""Writes a single-precision (4 bytes) floating point number to `self` in big-endian byte order.
 986
 987        The valid range for the float value is `-3.402823466e38` to `3.402823466e38`.
 988
 989        Example
 990        -------
 991        ```py
 992        buf = BytesMut()
 993        buf.put_float(1.2)
 994        assert buf.to_bytes() == b'\x3f\x99\x99\x9a'
 995        ```
 996
 997        Raises
 998        ------
 999        `OverflowError`
1000            If `src` is out of range.
1001        """
1002        assert_precondition(
1003            -3.402823466e38 <= src <= 3.402823466e38,
1004            f"Float {src} is out of range for a single-precision float",
1005            OverflowError,
1006        )
1007        bits = struct.pack(">f", src)
1008
1009        if self._buf is None:
1010            self._buf = array.array("B", bits)
1011        else:
1012            self._buf.extend(bits)
1013
1014    def put_char(self, char: str) -> None:
1015        """Append a single character to the buffer.
1016
1017        This is the same as `self.put(ord(char))`.
1018
1019        Example
1020        -------
1021        ```py
1022        buf = BytesMut()
1023        buf.put_char('a')
1024        assert buf == b"a"
1025        ```
1026
1027        Parameters
1028        ----------
1029        char : `str`
1030            The character to put.
1031        """
1032        assert (ln := len(char)) == 1, f"Expected a single character, got {ln}"
1033        self.put(ord(char))
1034
1035    def put_raw(self, src: Rawish) -> None:
1036        """Extend `self` from a raw data type source.
1037
1038        Example
1039        -------
1040        ```py
1041        buffer = BytesMut()
1042        # A file descriptor's contents
1043        with open('file.txt', 'rb') as file:
1044            buffer.put_raw(file)
1045
1046        # bytes io
1047        buffer.put(io.BytesIO(b"data"))
1048        # string io
1049        buffer.put(io.StringIO("data"))
1050        ```
1051
1052        Parameters
1053        ----------
1054        src : `Rawish`
1055            A valid raw data type. See `Rawish` for more details.
1056
1057        Raises
1058        ------
1059        `OverflowError`
1060            If `src` not in range of `0..255`
1061        """
1062        if self._buf is None:
1063            # If it was `None`, we initialize it with a source instead of extending.
1064            self._buf = array.array("B", unwrap_bytes(src))
1065        else:
1066            self._buf.extend(unwrap_bytes(src))
1067
1068    def put_bytes(self, src: Buffer) -> None:
1069        """Put `bytes` into `self`.
1070
1071        Example
1072        -------
1073        ```py
1074        buf = BytesMut.from_bytes(b"hello")
1075        buf.put_bytes([32, 119, 111, 114, 108, 100])
1076        assert buf == b"hello world"
1077        ```
1078
1079        Parameters
1080        ----------
1081        src : `Buffer`
1082            Can be one of `Bytes`, `bytes`, `bytearray` or `Sequence[int]`
1083
1084        Raises
1085        ------
1086        `OverflowError`
1087            If `src` not in range of `0..255`
1088        """
1089        if self._buf is None:
1090            # If it was `None`, we initialize it with a source instead of extending.
1091            self._buf = array.array("B", src)
1092        else:
1093            self._buf.extend(src)
1094
1095    def put_str(self, s: str) -> None:
1096        """Put a `utf-8` encoded bytes from a string.
1097
1098        Example
1099        -------
1100        ```py
1101        buffer = BytesMut()
1102        buffer.put_str("hello")
1103
1104        assert buffer == b"hello"
1105        ```
1106
1107        Parameters
1108        ----------
1109        src: `str`
1110            The string
1111        """
1112        self.put_bytes(s.encode(ENCODING))
1113
1114    def replace(self, index: int, byte: int) -> None:
1115        """Replace the byte at `index` with `byte`.
1116
1117        This method is `NOOP` if:
1118        -------------------------
1119        * `self` is empty or unallocated.
1120        * `index` is out of range.
1121
1122        Example
1123        -------
1124        ```py
1125        buf = BytesMut.from_bytes([1, 2, 3])
1126        buf.replace(1, 4)
1127        assert buf == [1, 4, 3]
1128        ```
1129        """
1130        if not self._buf or index < 0 or index >= self.len():
1131            return
1132
1133        self._buf[index] = byte
1134
1135    def replace_with(self, index: int, f: collections.Callable[[int], int]) -> None:
1136        """Replace the byte at `index` with a byte returned from `f`.
1137
1138        The signature of `f` is `Fn(int) -> int`, where the argument is the old byte.
1139
1140        ## This method is `NOOP` if:
1141        * `self` is empty or unallocated.
1142        * `index` is out of range.
1143
1144        Example
1145        -------
1146        ```py
1147        buf = BytesMut.from_bytes([1, 2, 3])
1148        buf.replace_with(1, lambda prev: prev * 2)
1149        assert buf == [1, 4, 3]
1150        ```
1151        """
1152        if not self._buf or index < 0 or index >= self.len():
1153            return
1154
1155        old = self._buf[index]
1156        self._buf[index] = f(old)
1157
1158    def offset(self, f: collections.Callable[[int], int]) -> None:
1159        """Modify each byte in the buffer with a new byte returned from `f`.
1160
1161        The signature of `f` is `Fn(int) -> int`, where the argument is the previous byte.
1162
1163        Example
1164        -------
1165        ```py
1166        buf = BytesMut.from_bytes([1, 2, 3])
1167        buf.offset(lambda prev: prev * 2)
1168        assert buf == [2, 4, 6]
1169        ```
1170
1171        This method is `NOOP` if `self` is empty or unallocated.
1172        """
1173
1174        if not self._buf:
1175            return
1176
1177        for index in range(len(self._buf)):
1178            self._buf[index] = f(self._buf[index])
1179
1180    def fill(self, value: int) -> None:
1181        """Fills `self` with the given byte.
1182
1183        Nothing happens if the buffer is empty or unallocated.
1184
1185        Example
1186        -------
1187        ```py
1188        a = Bytes.from_bytes([0, 1, 2, 3])
1189        a.fill(0)
1190        assert a == [0, 0, 0, 0]
1191        ```
1192        """
1193        if not self._buf:
1194            return
1195
1196        self.as_mut_ptr()[:] = bytearray([value] * self.len())
1197
1198    def fill_with(self, f: collections.Callable[[], int]) -> None:
1199        """Fills `self` with the given byte returned from `f`.
1200
1201        Nothing happens if the buffer is empty or unallocated.
1202
1203        Example
1204        -------
1205        ```py
1206        def default() -> int:
1207            return 0
1208
1209        a = Bytes.from_bytes([0, 1, 2, 3])
1210        a.fill_with(default)
1211        assert a == [0, 0, 0, 0]
1212        ```
1213        """
1214        if not self._buf:
1215            return
1216
1217        self.as_mut_ptr()[:] = bytearray([f()] * self.len())
1218
1219    def swap(self, a: int, b: int):
1220        """Swap two bytes in the buffer.
1221
1222        if `a` equals to `b` then it's guaranteed that elements won't change value.
1223
1224        Example
1225        -------
1226        ```py
1227        buf = Bytes.from_bytes([1, 2, 3, 4])
1228        buf.swap(0, 3)
1229        assert buf == [4, 2, 3, 1]
1230        ```
1231
1232        Raises
1233        ------
1234        IndexError
1235            If the positions of `a` or `b` are out of index.
1236        """
1237        if self[a] == self[b]:
1238            return
1239
1240        self[a], self[b] = self[b], self[a]
1241
1242    def swap_unchecked(self, a: int, b: int):
1243        """Swap two bytes in the buffer. without checking if `a` == `b`.
1244
1245        If you care about `a` and `b` equality, see `Bytes.swap`.
1246
1247        Example
1248        -------
1249        ```py
1250        buf = Bytes.from_bytes([1, 2, 3, 1])
1251        buf.swap_unchecked(0, 3)
1252        assert buf == [1, 2, 3, 1]
1253        ```
1254
1255        Raises
1256        ------
1257        IndexError
1258            If the positions of `a` or `b` are out of index.
1259        """
1260        self[a], self[b] = self[b], self[a]
1261
1262    def as_mut_ptr(self) -> memoryview[int]:
1263        """Returns a mutable pointer to the buffer data.
1264
1265        `pointer` here refers to a `memoryview` object.
1266
1267        A `BufferError` is raised if the underlying sequence is not initialized.
1268
1269        Example
1270        -------
1271        ```py
1272        buffer = BytesMut.from_str("ouv")
1273        ptr = buffer.as_mut_ptr()
1274        ptr[0] = ord(b'L')
1275        assert buffer.to_bytes() == b"Luv"
1276        ```
1277        """
1278        return self.__buffer__(512)
1279
1280    def as_mut(self) -> _slice.SliceMut[int]:
1281        """Get a mutable reference to the underlying sequence, without copying.
1282
1283        An empty slice is returned if the underlying sequence is not initialized.
1284
1285        Example
1286        -------
1287        ```py
1288        buff = BytesMut.from_str("Hello")
1289        ref = buff.as_mut()
1290        ref.append(32)
1291        del ref
1292        assert buff == b"Hello "
1293        ```
1294        """
1295        if self._buf is not None:
1296            return _slice.SliceMut(self)
1297
1298        return _slice.SliceMut([])
1299
1300    @safe
1301    def freeze(self) -> Bytes:
1302        """Convert `self` into an immutable `Bytes`.
1303
1304        This conversion is zero-cost, meaning it doesn't any _hidden-copy_ operations behind the scenes.
1305        This consumes `self` and returns a new `Bytes` that points to the same underlying array,
1306
1307        Notes
1308        -----
1309        * If `self` is not initialized, a new empty `Bytes` is returned.
1310        * `self` will no longer be usable, as it will not point to the underlying array.
1311
1312        The inverse method of this is `Bytes.to_mut()`
1313
1314        Example
1315        -------
1316        ```py
1317        def shrink_to(cap: int, buffer: BytesMut) -> Bytes:
1318            buf.truncate(cap)
1319            return buf.freeze()
1320
1321        buffer = BytesMut.from_bytes([32, 23, 34, 65])
1322        # accessing `buffer` after this is undefined behavior.
1323        modified = shrink_to(2, buffer)
1324        assert modified == [32, 23]
1325        ```
1326        """
1327        # SAFETY: `Bytes.leak` returns an empty array
1328        # if `self` is uninitialized.
1329        return Bytes.from_ptr_unchecked(self.leak())
1330
1331    def swap_remove(self, byte: int) -> int:
1332        """Remove the first appearance of `item` from this buffer and return it.
1333
1334        Raises
1335        ------
1336        * `ValueError`: if `item` is not in this buffer.
1337        * `MemoryError`: if this buffer hasn't allocated, Aka nothing has been pushed to it.
1338
1339        Example
1340        -------
1341        ```py
1342        buf = BytesMut.from_bytes([1, 2, 3, 4])
1343        assert 1 == buf.swap_remove(1)
1344        assert buf == [2, 3, 4]
1345        ```
1346        """
1347        if self._buf is None:
1348            raise MemoryError("`self` is unallocated.") from None
1349
1350        return self._buf.pop(self.index(byte))
1351
1352    def truncate(self, size: int) -> None:
1353        """Shortens the bytes, keeping the first `size` elements and dropping the rest.
1354
1355        Example
1356        -------
1357        ```py
1358        buf = BytesMut.from_bytes([0, 0, 0, 0])
1359        buf.truncate(1)
1360        assert buf.len() == 1
1361        ```
1362        """
1363        if not self._buf:
1364            return
1365
1366        del self._buf[size:]
1367
1368    def split_off_mut(self, at: int) -> BytesMut:
1369        """Split the bytes off at the specified position, returning a new
1370        `BytesMut` at the range of `[at : len]`, leaving `self` at `[at : bytes_len]`.
1371
1372        if this bytes is empty, `self` is returned unchanged.
1373
1374        Example
1375        -------
1376        ```py
1377        origin = BytesMut.from_bytes((1, 2, 3, 4))
1378        split = origin.split_off_mut(2)
1379
1380        print(origin, split)  # [1, 2], [3, 4]
1381        ```
1382
1383        Raises
1384        ------
1385        `IndexError`
1386            This method will raise if `at` > `len(self)`
1387        """
1388        len_ = self.len()
1389        if at > len_:
1390            raise IndexError(
1391                f"the index of `at` ({at}) should be <= than len of `self` ({len_}) "
1392            ) from None
1393
1394        if not self._buf:
1395            return self
1396
1397        split = BytesMut.from_ptr(self._buf[at:len_])
1398        del self._buf[at:len_]
1399        return split
1400
1401    def split_first_mut(self) -> Option[tuple[int, BytesMut]]:
1402        """Split the first and rest elements of the bytes, If empty, `None` is returned.
1403
1404        Returns a tuple of (first, rest) where rest is a mutable sequence.
1405
1406        Example
1407        -------
1408        ```py
1409        buf = BytesMut.from_bytes([1, 2, 3])
1410        split = buf.split_first_mut()
1411        assert split == Some((1, [2, 3]))
1412        ```
1413        """
1414        if not self._buf:
1415            return _option.NOTHING
1416
1417        if self.len() == 1:
1418            return _option.Some((self[0], BytesMut()))
1419
1420        first = self[0]
1421        rest = self[1:]
1422        return _option.Some((first, rest))
1423
1424    def split_last_mut(self) -> Option[tuple[int, Bytes]]:
1425        """Returns the last and rest of the elements of the bytes, If `self` is empty, `None` is returned.
1426
1427        Returns a tuple of (last, rest) where rest is a mutable sequence.
1428
1429        Example
1430        -------
1431        ```py
1432        buf = BytesMut.from_bytes([0, 1, 2])
1433        last, elements = buf.split_last_mut().unwrap()
1434        assert (last, elements) == (2, [0, 1])
1435        ```
1436        """
1437        if not self._buf:
1438            return _option.NOTHING
1439
1440        len_ = self.len()
1441        if len_ == 1:
1442            return _option.Some((self[0], BytesMut()))
1443
1444        last = self[-1]
1445        rest = self[:-1]
1446        return _option.Some((last, rest))
1447
1448    def split_at_mut(self, mid: int) -> tuple[BytesMut, BytesMut]:
1449        """Divide `self` into two at an index.
1450
1451        The first will contain all bytes from `[0:mid]` excluding `mid` itself.
1452        and the second will contain the remaining bytes.
1453
1454        if `mid` > `self.len()`, Then all bytes will be moved to the left,
1455        returning an empty bytes in right.
1456
1457        Example
1458        -------
1459        ```py
1460        buffer = BytesMut.from_bytes((1, 2, 3, 4))
1461        left, right = buffer.split_at_mut(0)
1462        assert left == [] and right == [1, 2, 3, 4]
1463
1464        left, right = buffer.split_at_mut(2)
1465        assert left == [1, 2] and right == [3, 4]
1466        ```
1467
1468        The is roughly the implementation
1469        ```py
1470        self[0:mid], self[mid:]
1471        ```
1472        """
1473        return self[0:mid], self[mid:]
1474
1475    # * Layout * #
1476
1477    def insert(self, index: int, value: int) -> None:
1478        """Insert a new item with `value` in the buffer before position `index`.
1479
1480        Negative values are treated as being relative to the end of the buffer.
1481        """
1482        if self._buf is None:
1483            return
1484
1485        self._buf.insert(index, value)
1486
1487    def pop(self, i: int = -1) -> Option[int]:
1488        """Removes the last element from the buffer and returns it, `Some(None)` if it is empty.
1489
1490        Example
1491        -------
1492        ```py
1493        buf = BytesMut((21, 32, 44))
1494        assert buf.pop() == Some(44)
1495        ```
1496        """
1497        if not self._buf:
1498            return _option.NOTHING
1499
1500        return _option.Some(self._buf.pop(i))
1501
1502    def remove(self, i: int) -> None:
1503        """Remove the first appearance of `i` from `self`.
1504
1505        Example
1506        ------
1507        ```py
1508        buf = BytesMut.from_bytes([1, 1, 2, 3, 4])
1509        buf.remove(1)
1510        print(buf) # [1, 2, 3, 4]
1511        ```
1512        """
1513        if not self._buf:
1514            return
1515
1516        self._buf.remove(i)
1517
1518    def clear(self) -> None:
1519        """Clear the buffer.
1520
1521        Example
1522        -------
1523        ```py
1524        buf = BytesMut.from_bytes([255])
1525        buf.clear()
1526
1527        assert buf.is_empty()
1528        ```
1529        """
1530        if not self._buf:
1531            return
1532
1533        del self._buf[:]
1534
1535    def byteswap(self) -> None:
1536        """Swap the byte order of the bytes in `self`."""
1537        if not self._buf:
1538            return
1539
1540        self._buf.byteswap()
1541
1542    def copy(self) -> BytesMut:
1543        """Create a copy of the bytes.
1544
1545        Example
1546        -------
1547        ```py
1548        original = BytesMut.from_bytes([255, 255, 255, 0])
1549        copy = original.copy()
1550        ```
1551        """
1552        if not self._buf:
1553            return BytesMut()
1554
1555        return self.from_ptr(self._buf[:])
1556
1557    def __setitem__(self, index: int, value: int):
1558        if not self._buf:
1559            raise IndexError("index out of range")
1560
1561        self._buf[index] = value
1562
1563    def __delitem__(self, key: typing.SupportsIndex | slice, /) -> None:
1564        if not self._buf:
1565            raise IndexError("index out of range")
1566
1567        del self._buf[key]
1568
1569    @typing.overload
1570    def __getitem__(self, index: slice) -> BytesMut: ...
1571
1572    @typing.overload
1573    def __getitem__(self, index: int) -> int: ...
1574
1575    @safe
1576    def __getitem__(self, index: int | slice) -> int | BytesMut:
1577        if not self._buf:
1578            raise IndexError("index out of range")
1579
1580        if isinstance(index, slice):
1581            # SAFETY: `self._buf` is initialized.
1582            return self.from_ptr_unchecked(self._buf[index])
1583
1584        return self._buf[index]
1585
1586    @safe
1587    def __copy__(self) -> BytesMut:
1588        if not self._buf:
1589            return BytesMut()
1590
1591        return BytesMut.from_ptr_unchecked(self._buf.__copy__())
1592
1593    @safe
1594    def __deepcopy__(self, unused: typing.Any, /) -> BytesMut:
1595        if not self._buf:
1596            return BytesMut()
1597
1598        return BytesMut.from_ptr_unchecked(self._buf.__deepcopy__(unused))

Provides mutable abstractions for working with bytes.

It is an efficient container for storing and operating with bytes, It is built on-top of array.array[int], which means you get all of array[int]'s operations.

A bytes object is usually used within networking applications, but can also be used elsewhere as well.

Construction

You can create a BytesMut object in multiple ways.

Example
from sain import BytesNut

buf = BytesMut()
buffer.put_bytes(b"Hello")
print(buffer) # [72, 101, 108, 108, 111]

buf.put(32) # space
assert buffer.to_bytes() == b"Hello "

Implementations

This class implements &mut [u8] in Rust.

def extend(self, src: bytes | bytearray | Iterable[int]) -> None:
927    def extend(self, src: Buffer) -> None:
928        """Extend `self` from a `src`.
929
930        Example
931        -------
932        ```py
933        buf = Bytes()
934        buf.extend([1, 2, 3])
935        assert buf == [1, 2, 3]
936        ```
937
938        Parameters
939        ----------
940        src : `Buffer`
941            Can be one of `Bytes`, `bytes`, `bytearray` or `Sequence[int]`
942
943        Raises
944        ------
945        `OverflowError`
946            If `src` not in range of `0..255`
947        """
948        if self._buf is None:
949            # If it was `None`, we initialize it with a source instead of extending.
950            self._buf = array.array("B", src)
951        else:
952            self._buf.extend(src)

Extend self from a src.

Example
buf = Bytes()
buf.extend([1, 2, 3])
assert buf == [1, 2, 3]
Parameters
  • src (Buffer): Can be one of Bytes, bytes, bytearray or Sequence[int]
Raises
  • OverflowError: If src not in range of 0..255
def put(self, src: int) -> None:
954    def put(self, src: int) -> None:
955        """Append a byte at the end of the array.
956
957        unlike `.put_bytes`, this method appends instead of extending the array
958        which is faster if you're putting a single byte in a single call.
959
960        Example
961        -------
962        ```py
963        buf = Bytes()
964        buf.put(32) # append a space to the end of the buffer
965        assert buf.to_bytes() == b' '
966        ```
967
968        Parameters
969        ----------
970        src : `int`
971            An unsigned integer, also known as `u8` to put.
972
973        Raises
974        ------
975        `OverflowError`
976            If `src` not in range of `0..255`
977        """
978        if self._buf is None:
979            # If it was `None`, we initialize it with a source instead of appending.
980            self._buf = array.array("B", [src])
981        else:
982            self._buf.append(src)

Append a byte at the end of the array.

unlike .put_bytes, this method appends instead of extending the array which is faster if you're putting a single byte in a single call.

Example
buf = Bytes()
buf.put(32) # append a space to the end of the buffer
assert buf.to_bytes() == b' '
Parameters
  • src (int): An unsigned integer, also known as u8 to put.
Raises
  • OverflowError: If src not in range of 0..255
def put_float(self, src: float) -> None:
 984    def put_float(self, src: float) -> None:
 985        r"""Writes a single-precision (4 bytes) floating point number to `self` in big-endian byte order.
 986
 987        The valid range for the float value is `-3.402823466e38` to `3.402823466e38`.
 988
 989        Example
 990        -------
 991        ```py
 992        buf = BytesMut()
 993        buf.put_float(1.2)
 994        assert buf.to_bytes() == b'\x3f\x99\x99\x9a'
 995        ```
 996
 997        Raises
 998        ------
 999        `OverflowError`
1000            If `src` is out of range.
1001        """
1002        assert_precondition(
1003            -3.402823466e38 <= src <= 3.402823466e38,
1004            f"Float {src} is out of range for a single-precision float",
1005            OverflowError,
1006        )
1007        bits = struct.pack(">f", src)
1008
1009        if self._buf is None:
1010            self._buf = array.array("B", bits)
1011        else:
1012            self._buf.extend(bits)

Writes a single-precision (4 bytes) floating point number to self in big-endian byte order.

The valid range for the float value is -3.402823466e38 to 3.402823466e38.

Example
buf = BytesMut()
buf.put_float(1.2)
assert buf.to_bytes() == b'\x3f\x99\x99\x9a'
Raises
  • OverflowError: If src is out of range.
def put_char(self, char: str) -> None:
1014    def put_char(self, char: str) -> None:
1015        """Append a single character to the buffer.
1016
1017        This is the same as `self.put(ord(char))`.
1018
1019        Example
1020        -------
1021        ```py
1022        buf = BytesMut()
1023        buf.put_char('a')
1024        assert buf == b"a"
1025        ```
1026
1027        Parameters
1028        ----------
1029        char : `str`
1030            The character to put.
1031        """
1032        assert (ln := len(char)) == 1, f"Expected a single character, got {ln}"
1033        self.put(ord(char))

Append a single character to the buffer.

This is the same as self.put(ord(char)).

Example
buf = BytesMut()
buf.put_char('a')
assert buf == b"a"
Parameters
  • char (str): The character to put.
def put_raw(self, src: _io.StringIO | _io.BytesIO | _io.BufferedReader) -> None:
1035    def put_raw(self, src: Rawish) -> None:
1036        """Extend `self` from a raw data type source.
1037
1038        Example
1039        -------
1040        ```py
1041        buffer = BytesMut()
1042        # A file descriptor's contents
1043        with open('file.txt', 'rb') as file:
1044            buffer.put_raw(file)
1045
1046        # bytes io
1047        buffer.put(io.BytesIO(b"data"))
1048        # string io
1049        buffer.put(io.StringIO("data"))
1050        ```
1051
1052        Parameters
1053        ----------
1054        src : `Rawish`
1055            A valid raw data type. See `Rawish` for more details.
1056
1057        Raises
1058        ------
1059        `OverflowError`
1060            If `src` not in range of `0..255`
1061        """
1062        if self._buf is None:
1063            # If it was `None`, we initialize it with a source instead of extending.
1064            self._buf = array.array("B", unwrap_bytes(src))
1065        else:
1066            self._buf.extend(unwrap_bytes(src))

Extend self from a raw data type source.

Example
buffer = BytesMut()
# A file descriptor's contents
with open('file.txt', 'rb') as file:
    buffer.put_raw(file)

# bytes io
buffer.put(io.BytesIO(b"data"))
# string io
buffer.put(io.StringIO("data"))
Parameters
  • src (Rawish): A valid raw data type. See Rawish for more details.
Raises
  • OverflowError: If src not in range of 0..255
def put_bytes(self, src: bytes | bytearray | Iterable[int]) -> None:
1068    def put_bytes(self, src: Buffer) -> None:
1069        """Put `bytes` into `self`.
1070
1071        Example
1072        -------
1073        ```py
1074        buf = BytesMut.from_bytes(b"hello")
1075        buf.put_bytes([32, 119, 111, 114, 108, 100])
1076        assert buf == b"hello world"
1077        ```
1078
1079        Parameters
1080        ----------
1081        src : `Buffer`
1082            Can be one of `Bytes`, `bytes`, `bytearray` or `Sequence[int]`
1083
1084        Raises
1085        ------
1086        `OverflowError`
1087            If `src` not in range of `0..255`
1088        """
1089        if self._buf is None:
1090            # If it was `None`, we initialize it with a source instead of extending.
1091            self._buf = array.array("B", src)
1092        else:
1093            self._buf.extend(src)

Put bytes into self.

Example
buf = BytesMut.from_bytes(b"hello")
buf.put_bytes([32, 119, 111, 114, 108, 100])
assert buf == b"hello world"
Parameters
  • src (Buffer): Can be one of Bytes, bytes, bytearray or Sequence[int]
Raises
  • OverflowError: If src not in range of 0..255
def put_str(self, s: str) -> None:
1095    def put_str(self, s: str) -> None:
1096        """Put a `utf-8` encoded bytes from a string.
1097
1098        Example
1099        -------
1100        ```py
1101        buffer = BytesMut()
1102        buffer.put_str("hello")
1103
1104        assert buffer == b"hello"
1105        ```
1106
1107        Parameters
1108        ----------
1109        src: `str`
1110            The string
1111        """
1112        self.put_bytes(s.encode(ENCODING))

Put a utf-8 encoded bytes from a string.

Example
buffer = BytesMut()
buffer.put_str("hello")

assert buffer == b"hello"
Parameters
  • src (str): The string
def replace(self, index: int, byte: int) -> None:
1114    def replace(self, index: int, byte: int) -> None:
1115        """Replace the byte at `index` with `byte`.
1116
1117        This method is `NOOP` if:
1118        -------------------------
1119        * `self` is empty or unallocated.
1120        * `index` is out of range.
1121
1122        Example
1123        -------
1124        ```py
1125        buf = BytesMut.from_bytes([1, 2, 3])
1126        buf.replace(1, 4)
1127        assert buf == [1, 4, 3]
1128        ```
1129        """
1130        if not self._buf or index < 0 or index >= self.len():
1131            return
1132
1133        self._buf[index] = byte

Replace the byte at index with byte.

This method is NOOP if:

  • self is empty or unallocated.
  • index is out of range.
Example
buf = BytesMut.from_bytes([1, 2, 3])
buf.replace(1, 4)
assert buf == [1, 4, 3]
def replace_with(self, index: int, f: Callable[[int], int]) -> None:
1135    def replace_with(self, index: int, f: collections.Callable[[int], int]) -> None:
1136        """Replace the byte at `index` with a byte returned from `f`.
1137
1138        The signature of `f` is `Fn(int) -> int`, where the argument is the old byte.
1139
1140        ## This method is `NOOP` if:
1141        * `self` is empty or unallocated.
1142        * `index` is out of range.
1143
1144        Example
1145        -------
1146        ```py
1147        buf = BytesMut.from_bytes([1, 2, 3])
1148        buf.replace_with(1, lambda prev: prev * 2)
1149        assert buf == [1, 4, 3]
1150        ```
1151        """
1152        if not self._buf or index < 0 or index >= self.len():
1153            return
1154
1155        old = self._buf[index]
1156        self._buf[index] = f(old)

Replace the byte at index with a byte returned from f.

The signature of f is Fn(int) -> int, where the argument is the old byte.

This method is NOOP if:

  • self is empty or unallocated.
  • index is out of range.
Example
buf = BytesMut.from_bytes([1, 2, 3])
buf.replace_with(1, lambda prev: prev * 2)
assert buf == [1, 4, 3]
def offset(self, f: Callable[[int], int]) -> None:
1158    def offset(self, f: collections.Callable[[int], int]) -> None:
1159        """Modify each byte in the buffer with a new byte returned from `f`.
1160
1161        The signature of `f` is `Fn(int) -> int`, where the argument is the previous byte.
1162
1163        Example
1164        -------
1165        ```py
1166        buf = BytesMut.from_bytes([1, 2, 3])
1167        buf.offset(lambda prev: prev * 2)
1168        assert buf == [2, 4, 6]
1169        ```
1170
1171        This method is `NOOP` if `self` is empty or unallocated.
1172        """
1173
1174        if not self._buf:
1175            return
1176
1177        for index in range(len(self._buf)):
1178            self._buf[index] = f(self._buf[index])

Modify each byte in the buffer with a new byte returned from f.

The signature of f is Fn(int) -> int, where the argument is the previous byte.

Example
buf = BytesMut.from_bytes([1, 2, 3])
buf.offset(lambda prev: prev * 2)
assert buf == [2, 4, 6]

This method is NOOP if self is empty or unallocated.

def fill(self, value: int) -> None:
1180    def fill(self, value: int) -> None:
1181        """Fills `self` with the given byte.
1182
1183        Nothing happens if the buffer is empty or unallocated.
1184
1185        Example
1186        -------
1187        ```py
1188        a = Bytes.from_bytes([0, 1, 2, 3])
1189        a.fill(0)
1190        assert a == [0, 0, 0, 0]
1191        ```
1192        """
1193        if not self._buf:
1194            return
1195
1196        self.as_mut_ptr()[:] = bytearray([value] * self.len())

Fills self with the given byte.

Nothing happens if the buffer is empty or unallocated.

Example
a = Bytes.from_bytes([0, 1, 2, 3])
a.fill(0)
assert a == [0, 0, 0, 0]
def fill_with(self, f: Callable[[], int]) -> None:
1198    def fill_with(self, f: collections.Callable[[], int]) -> None:
1199        """Fills `self` with the given byte returned from `f`.
1200
1201        Nothing happens if the buffer is empty or unallocated.
1202
1203        Example
1204        -------
1205        ```py
1206        def default() -> int:
1207            return 0
1208
1209        a = Bytes.from_bytes([0, 1, 2, 3])
1210        a.fill_with(default)
1211        assert a == [0, 0, 0, 0]
1212        ```
1213        """
1214        if not self._buf:
1215            return
1216
1217        self.as_mut_ptr()[:] = bytearray([f()] * self.len())

Fills self with the given byte returned from f.

Nothing happens if the buffer is empty or unallocated.

Example
def default() -> int:
    return 0

a = Bytes.from_bytes([0, 1, 2, 3])
a.fill_with(default)
assert a == [0, 0, 0, 0]
def swap(self, a: int, b: int):
1219    def swap(self, a: int, b: int):
1220        """Swap two bytes in the buffer.
1221
1222        if `a` equals to `b` then it's guaranteed that elements won't change value.
1223
1224        Example
1225        -------
1226        ```py
1227        buf = Bytes.from_bytes([1, 2, 3, 4])
1228        buf.swap(0, 3)
1229        assert buf == [4, 2, 3, 1]
1230        ```
1231
1232        Raises
1233        ------
1234        IndexError
1235            If the positions of `a` or `b` are out of index.
1236        """
1237        if self[a] == self[b]:
1238            return
1239
1240        self[a], self[b] = self[b], self[a]

Swap two bytes in the buffer.

if a equals to b then it's guaranteed that elements won't change value.

Example
buf = Bytes.from_bytes([1, 2, 3, 4])
buf.swap(0, 3)
assert buf == [4, 2, 3, 1]
Raises
  • IndexError: If the positions of a or b are out of index.
def swap_unchecked(self, a: int, b: int):
1242    def swap_unchecked(self, a: int, b: int):
1243        """Swap two bytes in the buffer. without checking if `a` == `b`.
1244
1245        If you care about `a` and `b` equality, see `Bytes.swap`.
1246
1247        Example
1248        -------
1249        ```py
1250        buf = Bytes.from_bytes([1, 2, 3, 1])
1251        buf.swap_unchecked(0, 3)
1252        assert buf == [1, 2, 3, 1]
1253        ```
1254
1255        Raises
1256        ------
1257        IndexError
1258            If the positions of `a` or `b` are out of index.
1259        """
1260        self[a], self[b] = self[b], self[a]

Swap two bytes in the buffer. without checking if a == b.

If you care about a and b equality, see Bytes.swap.

Example
buf = Bytes.from_bytes([1, 2, 3, 1])
buf.swap_unchecked(0, 3)
assert buf == [1, 2, 3, 1]
Raises
  • IndexError: If the positions of a or b are out of index.
def as_mut_ptr(self) -> 'memoryview[int]':
1262    def as_mut_ptr(self) -> memoryview[int]:
1263        """Returns a mutable pointer to the buffer data.
1264
1265        `pointer` here refers to a `memoryview` object.
1266
1267        A `BufferError` is raised if the underlying sequence is not initialized.
1268
1269        Example
1270        -------
1271        ```py
1272        buffer = BytesMut.from_str("ouv")
1273        ptr = buffer.as_mut_ptr()
1274        ptr[0] = ord(b'L')
1275        assert buffer.to_bytes() == b"Luv"
1276        ```
1277        """
1278        return self.__buffer__(512)

Returns a mutable pointer to the buffer data.

pointer here refers to a memoryview object.

A BufferError is raised if the underlying sequence is not initialized.

Example
buffer = BytesMut.from_str("ouv")
ptr = buffer.as_mut_ptr()
ptr[0] = ord(b'L')
assert buffer.to_bytes() == b"Luv"
def as_mut(self) -> sain.collections.slice.SliceMut[int]:
1280    def as_mut(self) -> _slice.SliceMut[int]:
1281        """Get a mutable reference to the underlying sequence, without copying.
1282
1283        An empty slice is returned if the underlying sequence is not initialized.
1284
1285        Example
1286        -------
1287        ```py
1288        buff = BytesMut.from_str("Hello")
1289        ref = buff.as_mut()
1290        ref.append(32)
1291        del ref
1292        assert buff == b"Hello "
1293        ```
1294        """
1295        if self._buf is not None:
1296            return _slice.SliceMut(self)
1297
1298        return _slice.SliceMut([])

Get a mutable reference to the underlying sequence, without copying.

An empty slice is returned if the underlying sequence is not initialized.

Example
buff = BytesMut.from_str("Hello")
ref = buff.as_mut()
ref.append(32)
del ref
assert buff == b"Hello "
@safe
def freeze(self) -> Bytes:
1300    @safe
1301    def freeze(self) -> Bytes:
1302        """Convert `self` into an immutable `Bytes`.
1303
1304        This conversion is zero-cost, meaning it doesn't any _hidden-copy_ operations behind the scenes.
1305        This consumes `self` and returns a new `Bytes` that points to the same underlying array,
1306
1307        Notes
1308        -----
1309        * If `self` is not initialized, a new empty `Bytes` is returned.
1310        * `self` will no longer be usable, as it will not point to the underlying array.
1311
1312        The inverse method of this is `Bytes.to_mut()`
1313
1314        Example
1315        -------
1316        ```py
1317        def shrink_to(cap: int, buffer: BytesMut) -> Bytes:
1318            buf.truncate(cap)
1319            return buf.freeze()
1320
1321        buffer = BytesMut.from_bytes([32, 23, 34, 65])
1322        # accessing `buffer` after this is undefined behavior.
1323        modified = shrink_to(2, buffer)
1324        assert modified == [32, 23]
1325        ```
1326        """
1327        # SAFETY: `Bytes.leak` returns an empty array
1328        # if `self` is uninitialized.
1329        return Bytes.from_ptr_unchecked(self.leak())

Convert self into an immutable Bytes.

This conversion is zero-cost, meaning it doesn't any _hidden-copy_ operations behind the scenes. This consumes self and returns a new Bytes that points to the same underlying array,

Notes
  • If self is not initialized, a new empty Bytes is returned.
  • self will no longer be usable, as it will not point to the underlying array.

The inverse method of this is Bytes.to_mut()

Example
def shrink_to(cap: int, buffer: BytesMut) -> Bytes:
    buf.truncate(cap)
    return buf.freeze()

buffer = BytesMut.from_bytes([32, 23, 34, 65])
# accessing `buffer` after this is undefined behavior.
modified = shrink_to(2, buffer)
assert modified == [32, 23]
def swap_remove(self, byte: int) -> int:
1331    def swap_remove(self, byte: int) -> int:
1332        """Remove the first appearance of `item` from this buffer and return it.
1333
1334        Raises
1335        ------
1336        * `ValueError`: if `item` is not in this buffer.
1337        * `MemoryError`: if this buffer hasn't allocated, Aka nothing has been pushed to it.
1338
1339        Example
1340        -------
1341        ```py
1342        buf = BytesMut.from_bytes([1, 2, 3, 4])
1343        assert 1 == buf.swap_remove(1)
1344        assert buf == [2, 3, 4]
1345        ```
1346        """
1347        if self._buf is None:
1348            raise MemoryError("`self` is unallocated.") from None
1349
1350        return self._buf.pop(self.index(byte))

Remove the first appearance of item from this buffer and return it.

Raises
  • * ValueError (if item is not in this buffer.):

  • * MemoryError (if this buffer hasn't allocated, Aka nothing has been pushed to it.):

Example
buf = BytesMut.from_bytes([1, 2, 3, 4])
assert 1 == buf.swap_remove(1)
assert buf == [2, 3, 4]
def truncate(self, size: int) -> None:
1352    def truncate(self, size: int) -> None:
1353        """Shortens the bytes, keeping the first `size` elements and dropping the rest.
1354
1355        Example
1356        -------
1357        ```py
1358        buf = BytesMut.from_bytes([0, 0, 0, 0])
1359        buf.truncate(1)
1360        assert buf.len() == 1
1361        ```
1362        """
1363        if not self._buf:
1364            return
1365
1366        del self._buf[size:]

Shortens the bytes, keeping the first size elements and dropping the rest.

Example
buf = BytesMut.from_bytes([0, 0, 0, 0])
buf.truncate(1)
assert buf.len() == 1
def split_off_mut(self, at: int) -> BytesMut:
1368    def split_off_mut(self, at: int) -> BytesMut:
1369        """Split the bytes off at the specified position, returning a new
1370        `BytesMut` at the range of `[at : len]`, leaving `self` at `[at : bytes_len]`.
1371
1372        if this bytes is empty, `self` is returned unchanged.
1373
1374        Example
1375        -------
1376        ```py
1377        origin = BytesMut.from_bytes((1, 2, 3, 4))
1378        split = origin.split_off_mut(2)
1379
1380        print(origin, split)  # [1, 2], [3, 4]
1381        ```
1382
1383        Raises
1384        ------
1385        `IndexError`
1386            This method will raise if `at` > `len(self)`
1387        """
1388        len_ = self.len()
1389        if at > len_:
1390            raise IndexError(
1391                f"the index of `at` ({at}) should be <= than len of `self` ({len_}) "
1392            ) from None
1393
1394        if not self._buf:
1395            return self
1396
1397        split = BytesMut.from_ptr(self._buf[at:len_])
1398        del self._buf[at:len_]
1399        return split

Split the bytes off at the specified position, returning a new BytesMut at the range of [at : len], leaving self at [at : bytes_len].

if this bytes is empty, self is returned unchanged.

Example
origin = BytesMut.from_bytes((1, 2, 3, 4))
split = origin.split_off_mut(2)

print(origin, split)  # [1, 2], [3, 4]
Raises
  • IndexError: This method will raise if at > len(self)
def split_first_mut(self) -> 'Option[tuple[int, BytesMut]]':
1401    def split_first_mut(self) -> Option[tuple[int, BytesMut]]:
1402        """Split the first and rest elements of the bytes, If empty, `None` is returned.
1403
1404        Returns a tuple of (first, rest) where rest is a mutable sequence.
1405
1406        Example
1407        -------
1408        ```py
1409        buf = BytesMut.from_bytes([1, 2, 3])
1410        split = buf.split_first_mut()
1411        assert split == Some((1, [2, 3]))
1412        ```
1413        """
1414        if not self._buf:
1415            return _option.NOTHING
1416
1417        if self.len() == 1:
1418            return _option.Some((self[0], BytesMut()))
1419
1420        first = self[0]
1421        rest = self[1:]
1422        return _option.Some((first, rest))

Split the first and rest elements of the bytes, If empty, None is returned.

Returns a tuple of (first, rest) where rest is a mutable sequence.

Example
buf = BytesMut.from_bytes([1, 2, 3])
split = buf.split_first_mut()
assert split == Some((1, [2, 3]))
def split_last_mut(self) -> 'Option[tuple[int, Bytes]]':
1424    def split_last_mut(self) -> Option[tuple[int, Bytes]]:
1425        """Returns the last and rest of the elements of the bytes, If `self` is empty, `None` is returned.
1426
1427        Returns a tuple of (last, rest) where rest is a mutable sequence.
1428
1429        Example
1430        -------
1431        ```py
1432        buf = BytesMut.from_bytes([0, 1, 2])
1433        last, elements = buf.split_last_mut().unwrap()
1434        assert (last, elements) == (2, [0, 1])
1435        ```
1436        """
1437        if not self._buf:
1438            return _option.NOTHING
1439
1440        len_ = self.len()
1441        if len_ == 1:
1442            return _option.Some((self[0], BytesMut()))
1443
1444        last = self[-1]
1445        rest = self[:-1]
1446        return _option.Some((last, rest))

Returns the last and rest of the elements of the bytes, If self is empty, None is returned.

Returns a tuple of (last, rest) where rest is a mutable sequence.

Example
buf = BytesMut.from_bytes([0, 1, 2])
last, elements = buf.split_last_mut().unwrap()
assert (last, elements) == (2, [0, 1])
def split_at_mut( self, mid: int) -> tuple[BytesMut, BytesMut]:
1448    def split_at_mut(self, mid: int) -> tuple[BytesMut, BytesMut]:
1449        """Divide `self` into two at an index.
1450
1451        The first will contain all bytes from `[0:mid]` excluding `mid` itself.
1452        and the second will contain the remaining bytes.
1453
1454        if `mid` > `self.len()`, Then all bytes will be moved to the left,
1455        returning an empty bytes in right.
1456
1457        Example
1458        -------
1459        ```py
1460        buffer = BytesMut.from_bytes((1, 2, 3, 4))
1461        left, right = buffer.split_at_mut(0)
1462        assert left == [] and right == [1, 2, 3, 4]
1463
1464        left, right = buffer.split_at_mut(2)
1465        assert left == [1, 2] and right == [3, 4]
1466        ```
1467
1468        The is roughly the implementation
1469        ```py
1470        self[0:mid], self[mid:]
1471        ```
1472        """
1473        return self[0:mid], self[mid:]

Divide self into two at an index.

The first will contain all bytes from [0:mid] excluding mid itself. and the second will contain the remaining bytes.

if mid > self.len(), Then all bytes will be moved to the left, returning an empty bytes in right.

Example
buffer = BytesMut.from_bytes((1, 2, 3, 4))
left, right = buffer.split_at_mut(0)
assert left == [] and right == [1, 2, 3, 4]

left, right = buffer.split_at_mut(2)
assert left == [1, 2] and right == [3, 4]

The is roughly the implementation

self[0:mid], self[mid:]
def insert(self, index: int, value: int) -> None:
1477    def insert(self, index: int, value: int) -> None:
1478        """Insert a new item with `value` in the buffer before position `index`.
1479
1480        Negative values are treated as being relative to the end of the buffer.
1481        """
1482        if self._buf is None:
1483            return
1484
1485        self._buf.insert(index, value)

Insert a new item with value in the buffer before position index.

Negative values are treated as being relative to the end of the buffer.

def pop(self, i: int = -1) -> 'Option[int]':
1487    def pop(self, i: int = -1) -> Option[int]:
1488        """Removes the last element from the buffer and returns it, `Some(None)` if it is empty.
1489
1490        Example
1491        -------
1492        ```py
1493        buf = BytesMut((21, 32, 44))
1494        assert buf.pop() == Some(44)
1495        ```
1496        """
1497        if not self._buf:
1498            return _option.NOTHING
1499
1500        return _option.Some(self._buf.pop(i))

Removes the last element from the buffer and returns it, Some(None) if it is empty.

Example
buf = BytesMut((21, 32, 44))
assert buf.pop() == Some(44)
def remove(self, i: int) -> None:
1502    def remove(self, i: int) -> None:
1503        """Remove the first appearance of `i` from `self`.
1504
1505        Example
1506        ------
1507        ```py
1508        buf = BytesMut.from_bytes([1, 1, 2, 3, 4])
1509        buf.remove(1)
1510        print(buf) # [1, 2, 3, 4]
1511        ```
1512        """
1513        if not self._buf:
1514            return
1515
1516        self._buf.remove(i)

Remove the first appearance of i from self.

Example
buf = BytesMut.from_bytes([1, 1, 2, 3, 4])
buf.remove(1)
print(buf) # [1, 2, 3, 4]
def clear(self) -> None:
1518    def clear(self) -> None:
1519        """Clear the buffer.
1520
1521        Example
1522        -------
1523        ```py
1524        buf = BytesMut.from_bytes([255])
1525        buf.clear()
1526
1527        assert buf.is_empty()
1528        ```
1529        """
1530        if not self._buf:
1531            return
1532
1533        del self._buf[:]

Clear the buffer.

Example
buf = BytesMut.from_bytes([255])
buf.clear()

assert buf.is_empty()
def byteswap(self) -> None:
1535    def byteswap(self) -> None:
1536        """Swap the byte order of the bytes in `self`."""
1537        if not self._buf:
1538            return
1539
1540        self._buf.byteswap()

Swap the byte order of the bytes in self.

Rawish: TypeAlias = _io.StringIO | _io.BytesIO | _io.BufferedReader

A type hint for some raw data type.

This can be any of:

  • io.StringIO
  • io.BytesIO
  • io.BufferedReader
  • memoryview
Buffer: TypeAlias = bytes | bytearray | Iterable[int]

A type hint for some bytes data type.

This can be any of:

  • bytes
  • Bytes
  • bytearray
  • Iterable[int]
  • memoryview[int]