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))
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.
Bytes(): Initialize an emptyBytesobjectfrom_str: CreateBytesfromstrfrom_bytes: CreateBytesfrom aBufferbytes-like typefrom_raw: CreateBytesfrom aRawishtypefrom_ptr: CreateBytesthat points to anarray.array[int]without copying itBytes.zeroed(count): CreateBytesfilled withzeroes * count.
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.
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)
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.
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)
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
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.
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"
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'
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()
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"
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
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.
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
selfis not initialized, a new emptyBytesMutis returned. selfwill 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]
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
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
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
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)
# ☺
# ☻
# ♥
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')
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()
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 ifat>len(self)
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]))
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])
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:]
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()
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
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
Inherited Members
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.
BytesMut(): Initialize an emptyBytesMutobjectfrom_str: CreateBytesMutfromstrfrom_bytes: CreateBytesMutfrom aBufferbytes-like typefrom_raw: CreateBytesMutfrom aRawishtypefrom_ptr: CreateBytesMutthat points to anarray.array[int]without copying itBytesMut.zeroed(count): CreateBytesMutfilled withzeroes * count.
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.
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)
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 asu8to put.
Raises
OverflowError: Ifsrcnot in range of0..255
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: Ifsrcis out of range.
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.
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
Raises
OverflowError: Ifsrcnot in range of0..255
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
Raises
OverflowError: Ifsrcnot in range of0..255
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
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:
selfis empty or unallocated.indexis out of range.
Example
buf = BytesMut.from_bytes([1, 2, 3])
buf.replace(1, 4)
assert buf == [1, 4, 3]
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:
selfis empty or unallocated.indexis out of range.
Example
buf = BytesMut.from_bytes([1, 2, 3])
buf.replace_with(1, lambda prev: prev * 2)
assert buf == [1, 4, 3]
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.
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]
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]
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
aorbare out of index.
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
aorbare out of index.
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"
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 "
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
selfis not initialized, a new emptyBytesis returned. selfwill 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]
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(ifitemis 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]
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
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 ifat>len(self)
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]))
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])
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:]
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.
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)
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]
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.
Inherited Members
A type hint for some raw data type.
This can be any of:
io.StringIOio.BytesIOio.BufferedReadermemoryview
A type hint for some bytes data type.
This can be any of:
bytesBytesbytearrayIterable[int]memoryview[int]