sain.collections
Dynamically growable collections and containers.
These collections are basic implementations of Rust's standard collections crate. from under the hood, they're an extended
and more rich implementations of the built-in sequences such as list and bytearray.
When Should You Use Which Collection?
This question's answer should be pretty straightforward.
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"""Dynamically growable collections and containers. 31 32These collections are basic implementations of Rust's standard collections crate. from under the hood, they're an extended 33and more rich implementations of the built-in sequences such as `list` and `bytearray`. 34 35### When Should You Use Which Collection? 36This question's answer should be pretty straightforward. 37 38* Use `Vec` when you want to replace `list`. 39* Use `Bytes` when you want to store read-only bytes. The underlying sequence is an `array` of type `u8`. 40* Use `BytesMut` when you want a mutable version of `Bytes`. 41* Use `HashMap` when you want to replace `dict`. 42""" 43 44from __future__ import annotations 45 46__all__ = ("Vec", "Bytes", "BytesMut", "vec", "buf", "slice", "hash_map", "HashMap") 47 48from . import buf 49from . import hash_map 50from . import slice 51from . import vec 52from .buf import Bytes 53from .buf import BytesMut 54from .hash_map import HashMap 55from .vec import Vec
@rustc_diagnostic_item('Vec')
@typing.final
class
Vec70@rustc_diagnostic_item("Vec") 71@typing.final 72class Vec(typing.Generic[T], collections.MutableSequence[T], SpecContains[T]): 73 """A contiguous growable alternative to builtin `list` with extra functionalities. 74 75 The layout of `Vec<T>` is exactly the same as `list<T>`. 76 Which means `list<T>`'s methods are inherited into `Vec<T>`. 77 78 Example 79 ------- 80 ```py 81 names = Vec() 82 names.push('foo') 83 names.push('bar') 84 85 print(names) # ['foo', 'bar'] 86 assert names.len() == 2 87 ``` 88 89 Constructing 90 ------------ 91 * `Vec()`: Create an empty vec, this will not allocate the initial buffer. 92 * `Vec(other_list)`: Create a vec which points to `other_list` 93 * `Vec([1, 2, 3])`: Create a vec with `[1, 2, 3]` pre-allocated. 94 * `Vec(iterable)`: Create a vec from an `iterable`, This is `O(n)` where `n` is the number of elements, 95 since it copies `iterable`'s elements into a new list. 96 * `Vec.with_capacity(5)`: Create a vec that can hold up to 5 elements 97 98 Iterating over `Vec` 99 ------------------- 100 There're two ways to iterate over a `Vec`. The first is to normally use `for` loop. 101 102 ```py 103 for i in names: 104 print(i) 105 106 # foo 107 # bar 108 ``` 109 110 The second is to use `Vec.iter`, which yields all items in this `Vec` from start to end. 111 Then the iterator gets exhausted as usual, See `sain.Iterator`. 112 113 ```py 114 iterator = names.iter() 115 for name in iterator.map(str.upper): 116 print(name) 117 118 # FOO 119 # BAR 120 121 # No more items, The actual vec is left unchanged. 122 assert iterator.next().is_none() 123 ``` 124 125 ## Comparison operators 126 A `Vec` may be compared with another `Vec` or a `list`, any other type will return `False` 127 128 ```py 129 vec = Vec([1, 2, 3]) 130 assert vec == [1, 2, 3] # True 131 assert vec != (1, 2, 3) 132 ``` 133 134 Zero-Copy 135 --------- 136 A vec that gets initialized from a `list` will *point* to it and doesn't copy it. 137 So any element that gets appended to the list will also get pushed into the vec 138 that's pointing to it and vice-versa. 139 140 ```py 141 cells: list[str] = [] 142 vec = Vec(cells) # This DOES NOT copy the `cells`. 143 144 cells.append("foo") 145 vec[0] == "foo" # True 146 ``` 147 148 The opposite of the above is to initialize the vec from either 149 an iterable or args, or copy the list. 150 151 ```py 152 from sain.collections import vec, Vec 153 154 # Copy the list into a vec. 155 vec = Vec(cells[:]) 156 cells.append("bar") 157 158 vec[1] # IndexError: "bar" doesn't exist in vec. 159 ``` 160 """ 161 162 __slots__ = ("_ptr", "_capacity") 163 164 @typing.overload 165 def __init__(self) -> None: ... 166 167 @typing.overload 168 def __init__(self, iterable: collections.Iterable[T]) -> None: ... 169 170 def __init__(self, iterable: collections.Iterable[T] | None = None) -> None: 171 """Create an empty `Vec<T>`. 172 173 The vector will not allocate until elements are pushed onto it. 174 175 Example 176 ------ 177 ```py 178 vec: Vec[float] = Vec() 179 ``` 180 """ 181 # We won't allocate to build the list here. 182 # Instead, On first push or fist indexed set 183 # we allocate if it was None. 184 if isinstance(iterable, list): 185 # Calling `list()` on another list will copy it, So instead we just point to it. 186 self._ptr = iterable 187 elif isinstance(iterable, Vec): 188 self._ptr = iterable._ptr 189 # any other iterable that ain't a list needs to get copied into a new list. 190 else: 191 self._ptr: list[T] | None = list(iterable) if iterable else None 192 193 self._capacity: int | None = None 194 195 @classmethod 196 def with_capacity(cls, capacity: int) -> Vec[T]: 197 """Create a new `Vec` with at least the specified capacity. 198 This vec will be able to hold `capacity` elements without pushing further. 199 200 Check out `Vec.push_within_capacity` as well. 201 202 Example 203 ------- 204 ```py 205 vec = Vec.with_capacity(3) 206 assert vec.len() == 0 and vec.capacity() >= 3 207 208 vec.push(1) 209 vec.push(2) 210 vec.push(3) 211 print(vec.len()) # 3 212 213 # This won't push. 214 vec.push(4) 215 ``` 216 """ 217 v = cls() 218 v._capacity = capacity 219 return v 220 221 def as_ref(self) -> Slice[T]: 222 """Return an immutable view over this vector elements. 223 224 No copying occurs. 225 226 Example 227 ------- 228 ```py 229 def send_bytes(v: Slice[int]) -> None: 230 # access `vec` in a read-only mode. 231 socket.write_all(v) 232 233 buf = Vec([1, 2, 3]) 234 send_bytes(buf.as_ref()) 235 ``` 236 """ 237 return Slice(self) 238 239 def as_mut(self) -> SliceMut[T]: 240 """Return a mutable view over this vector elements. 241 242 No copying occurs. 243 244 Example 245 ------- 246 ```py 247 def read_bytes(buf: SliceMut[int]) -> None: 248 socket.read_to_end(buf.as_mut()) 249 250 buf: Vec[int] = Vec() 251 read_bytes(buf.as_mut()) # or just `buf` 252 assert vec.len() >= 1 253 ``` 254 """ 255 return SliceMut(self) 256 257 def len(self) -> int: 258 """Return the number of elements in this vector. 259 260 Example 261 ------- 262 ```py 263 vec = Vec((1,2,3)) 264 265 assert vec.len() == 3 266 ``` 267 """ 268 return self.__len__() 269 270 def capacity(self) -> int: 271 """Return the capacity of this vector if set, 0 if not . 272 273 The number `0` here has two different Meanings: 274 - The first means the `Vec` doesn't have a specific capacity set. 275 - The second means the `Vec` literally initialized with `0` capacity. 276 277 Example 278 ------- 279 ```py 280 vec_with_cap = Vec.with_capacity(3) 281 assert vec_with_cap.capacity().unwrap() == 3 282 283 vec = Vec([1, 2, 3]) 284 assert vec.capacity() == 0 285 ``` 286 """ 287 return 0 if self._capacity is None else self._capacity 288 289 def iter(self) -> _iter.TrustedIter[T]: 290 """Returns an iterator over this vector elements. 291 292 Example 293 ------- 294 ```py 295 vec = Vec([1 ,2, 3]) 296 for element in vec.iter().map(str): 297 print(element) 298 ``` 299 """ 300 return _iter.TrustedIter(self._ptr or ()) 301 302 def is_empty(self) -> bool: 303 """Returns true if the vector contains no elements. 304 305 Inlined into: 306 ```py 307 not self.vec 308 ``` 309 """ 310 return not self._ptr 311 312 def leak(self) -> list[T]: 313 """Consumes and leaks the Vec, returning a mutable reference to the contents. 314 315 After calling this, this vec will no longer reference the underlying buffer, 316 therefore, it becomes unusable. 317 318 if `self` is uninitialized, an empty list is returned. 319 320 This function is only useful when you want to detach from a `Vec<T>` and get a `list[T]` without 321 any copies. 322 323 Example 324 ------- 325 ```py 326 x = Vec([1, 2, 3]) 327 owned = x.leak() 328 # x is now unusable. 329 owned[0] += 1 330 assert owned == [2, 2, 3] 331 ``` 332 """ 333 if self._ptr is not None: 334 # don't point to `_ptr` anymore. 335 tmp = self._ptr 336 del self._ptr 337 return tmp 338 339 return [] 340 341 def split_off(self, at: int) -> Vec[T]: 342 """Split the vector off at the specified position, returning a new 343 vec at the range of `[at : len]`, leaving `self` at `[at : vec_len]`. 344 345 if this vec is empty, `self` is returned unchanged. 346 347 Example 348 ------- 349 ```py 350 origin = Vec((1, 2, 3, 4)) 351 split = vec.split_off(2) 352 353 print(origin, split) # [1, 2], [3, 4] 354 ``` 355 356 Raises 357 ------ 358 `IndexError` 359 This method will raise if `at` > `len(self)` 360 """ 361 len_ = self.len() 362 if at > len_: 363 raise IndexError( 364 f"Index `at` ({at}) should be <= than len of vector ({len_}) " 365 ) from None 366 367 # Either the list is empty or uninit. 368 if not self._ptr: 369 return self 370 371 split = self[at:len_] # split the items into a new vec. 372 del self._ptr[at:len_] # remove the items from the original list. 373 return split 374 375 def split_first(self) -> _option.Option[tuple[T, collections.Sequence[T]]]: 376 """Split the first and rest elements of the vector, If empty, `None` is returned. 377 378 Example 379 ------- 380 ```py 381 vec = Vec([1, 2, 3]) 382 split = vec.split_first() 383 assert split == Some((1, [2, 3])) 384 385 vec: Vec[int] = Vec() 386 split = vec.split_first() 387 assert split.is_none() 388 ``` 389 """ 390 if not self._ptr: 391 return _option.NOTHING 392 393 # optimized to only one element in the vector. 394 if self.len() == 1: 395 return _option.Some((self[0], ())) 396 397 first, *rest = self._ptr 398 return _option.Some((first, rest)) 399 400 def split_last(self) -> _option.Option[tuple[T, collections.Sequence[T]]]: 401 """Split the last and rest elements of the vector, If empty, `None` is returned. 402 403 Example 404 ------- 405 ```py 406 vec = Vec([1, 2, 3]) 407 last, rest = vec.split_last().unwrap() 408 assert (last, rest) == [3, [1, 2]] 409 ``` 410 """ 411 if not self._ptr: 412 return _option.NOTHING 413 414 # optimized to only one element in the vector. 415 if self.len() == 1: 416 return _option.Some((self[0], ())) 417 418 last, *rest = self._ptr[-1], *self._ptr[:-1] 419 return _option.Some((last, rest)) 420 421 def split_at(self, mid: int) -> tuple[Vec[T], Vec[T]]: 422 """Divide `self` into two at an index. 423 424 The first will contain all elements from `[0:mid]` excluding `mid` it self. 425 and the second will contain the remaining elements. 426 427 if `mid` > `self.len()`, Then all elements will be moved to the left, 428 returning an empty vec in right. 429 430 Example 431 ------- 432 ```py 433 buffer = Vec((1, 2, 3, 4)) 434 left, right = buffer.split_at(0) 435 assert left == [] and right == [1, 2, 3, 4] 436 437 left, right = buffer.split_at(2) 438 assert left == [1, 2] and right == [2, 3] 439 ``` 440 441 The is roughly the implementation 442 ```py 443 vec[0:mid], vec[mid:] 444 ``` 445 """ 446 return self[0:mid], self[mid:] 447 448 def swap(self, a: int, b: int): 449 """Swap two elements in the vec. 450 451 if `a` equals to `b` then it's guaranteed that elements won't change value. 452 453 Example 454 ------- 455 ```py 456 buf = Vec([1, 2, 3, 4]) 457 buf.swap(0, 3) 458 assert buf == [4, 2, 3, 1] 459 ``` 460 461 Raises 462 ------ 463 IndexError 464 If the positions of `a` or `b` are out of index. 465 """ 466 if self[a] == self[b]: 467 return 468 469 self[a], self[b] = self[b], self[a] 470 471 def swap_unchecked(self, a: int, b: int): 472 """Swap two elements in the vec. without checking if `a` == `b`. 473 474 If you care about `a` and `b` equality, see `Vec.swap`. 475 476 Example 477 ------- 478 ```py 479 buf = Vec([1, 2, 3, 1]) 480 buf.swap_unchecked(0, 3) 481 assert buf == [1, 2, 3, 1] 482 ``` 483 484 Raises 485 ------ 486 IndexError 487 If the positions of `a` or `b` are out of index. 488 """ 489 self[a], self[b] = self[b], self[a] 490 491 def first(self) -> _option.Option[T]: 492 """Get the first element in this vec, returning `None` if there's none. 493 494 Example 495 ------- 496 ```py 497 vec = Vec((1,2,3)) 498 first = vec.first() 499 assert ~first == 1 500 ``` 501 """ 502 return self.get(0) 503 504 def last(self) -> _option.Option[T]: 505 """Get the last element in this vec, returning `None` if there's none. 506 507 Example 508 ------- 509 ```py 510 vec = Vec([1, 2, 3, 4]) 511 first = vec.last() 512 assert ~first == 4 513 ``` 514 """ 515 return self.get(-1) 516 517 def truncate(self, size: int) -> None: 518 """Shortens the vec, keeping the first `size` elements and dropping the rest. 519 520 Example 521 ------- 522 ```py 523 vec = Vec([1,2,3]) 524 vec.truncate(1) 525 assert vec == [1] 526 ``` 527 """ 528 if not self._ptr: 529 return 530 531 del self._ptr[size:] 532 533 def retain(self, f: collections.Callable[[T], bool]) -> None: 534 """Retains only the elements specified by the predicate. 535 536 This operation occurs in-place without copying the original list. 537 538 Example 539 ------- 540 ```py 541 vec = Vec([1, 2, 3]) 542 vec.retain(lambda elem: elem > 1) 543 544 assert vec == [2, 3] 545 ``` 546 547 The impl for this method is very simple 548 ```py 549 for idx, e in enumerate(vec): 550 if not f(e): 551 del vec[idx] 552 ``` 553 """ 554 if not self._ptr: 555 return 556 557 idx = 0 558 while idx < len(self._ptr): 559 if not f(self._ptr[idx]): 560 del self._ptr[idx] 561 else: 562 idx += 1 563 564 def swap_remove(self, item: T) -> T: 565 """Remove the first appearance of `item` from this vector and return it. 566 567 Raises 568 ------ 569 * `ValueError`: if `item` is not in this vector. 570 * `MemoryError`: if this vector hasn't allocated, Aka nothing has been pushed to it. 571 572 Example 573 ------- 574 ```py 575 vec = Vec(('a', 'b', 'c')) 576 element = vec.remove('a') 577 assert vec == ['b', 'c'] and element == 'a' 578 ``` 579 """ 580 if self._ptr is None: 581 raise MemoryError("Vec is unallocated.") from None 582 583 return self._ptr.pop(self.index(item)) 584 585 def fill(self, value: T) -> None: 586 """Fill `self` with the given `value`. 587 588 Nothing happens if the vec is empty or unallocated. 589 590 Example 591 ```py 592 a = Vec([0, 1, 2, 3]) 593 a.fill(0) 594 assert a == [0, 0, 0, 0] 595 ``` 596 """ 597 if not self._ptr: 598 return 599 600 for n, _ in enumerate(self): 601 self[n] = value 602 603 def push(self, item: T) -> None: 604 """Push an element at the end of the vector. 605 606 Example 607 ------- 608 ```py 609 vec = Vec() 610 vec.push(1) 611 612 assert vec == [1] 613 ``` 614 """ 615 if self._capacity is not None: 616 self.push_within_capacity(item) 617 return 618 619 if self._ptr is None: 620 self._ptr = [] 621 622 self._ptr.append(item) 623 624 def push_within_capacity(self, x: T) -> Result[None, T]: 625 """Appends an element if there is sufficient spare capacity, otherwise an error is returned with the element. 626 627 Example 628 ------- 629 ```py 630 vec: Vec[int] = Vec.with_capacity(3) 631 for i in range(3): 632 match vec.push_within_capacity(i): 633 case Ok(_): 634 print("All good.") 635 case Err(elem): 636 print("Reached max cap :< can't push", elem) 637 ``` 638 639 Or you can also just call `Vec.push` and it will push if there's is sufficient capacity. 640 ```py 641 vec: Vec[int] = Vec.with_capacity(3) 642 643 vec.extend((1, 2, 3)) 644 vec.push(4) 645 646 assert vec.len() == 3 647 ``` 648 """ 649 if self._ptr is None: 650 self._ptr = [] 651 652 if self.len() == self._capacity: 653 return _result.Err(x) 654 655 self._ptr.append(x) 656 return _result.Ok(None) 657 658 def reserve(self, additional: int) -> None: 659 """Reserves capacity for at least additional more elements to be inserted in the given Vec<T>. 660 661 Example 662 ------- 663 ```py 664 vec = Vec.with_capacity(3) 665 is_vip = random.choice((True, False)) 666 667 for i in range(4): 668 match vec.push_within_capacity(i): 669 case Ok(_): 670 print("All good") 671 case Err(person): 672 # If the person is a VIP, then reserve for one more. 673 if is_vip: 674 vec.reserve(1) 675 continue 676 677 # is_vip was false. 678 print("Can't reserve for non-VIP members...", person) 679 break 680 ``` 681 """ 682 if self._capacity is not None: 683 self._capacity += additional 684 685 def shrink_to_fit(self) -> None: 686 """Shrink the capacity of this `Vec` to match its length. 687 688 If `self` is initialized with no capacity, This is a `NOP`. 689 690 Example 691 ------- 692 ```py 693 s = Vec([1, 2, 3, 4]) 694 695 s.reserve(100) 696 s.shrink_to_fit() 697 assert s.capacity() == 4 698 ``` 699 """ 700 if self._capacity is None: 701 return 702 703 # The capacity is never less than the length. 704 self._capacity = min(self.__len__(), self._capacity) 705 706 def shrink_to(self, min_capacity: int) -> None: 707 """Shrink the capacity of this `Vec` to a minimum specified capacity. 708 709 If `self` is initialized with no capacity or the current capacity is less than the lower limit, 710 This is a `NOP`. 711 712 Example 713 ------- 714 ```py 715 vec = Vec.with_capacity(10) 716 vec.extend([1, 2, 3]) 717 assert vec.capacity() >= 10 718 vec.shrink_to(4) 719 assert vec.capacity() >= 4 720 vec.shrink_to(0) 721 ``` 722 """ 723 if self._capacity is None or self._capacity <= min_capacity: 724 return 725 726 # Ensure the capacity is not reduced below the current length of the vector. 727 self._capacity = max(self.__len__(), min_capacity) 728 729 ########################## 730 # * Builtin Operations * 731 ########################## 732 733 def append(self, value: T) -> None: 734 """An alias to `Vec.push`.""" 735 self.push(value) 736 737 def get(self, index: int) -> _option.Option[T]: 738 """Get the item at the given index, or `Some[None]` if its out of bounds. 739 740 Example 741 ------- 742 ```py 743 vec = Vec((1, 2, 3)) 744 vec.get(0) == Some(1) 745 vec.get(3) == Some(None) 746 ``` 747 """ 748 try: 749 return _option.Some(self[index]) 750 except IndexError: 751 return _option.NOTHING 752 753 def insert(self, index: int, value: T) -> None: 754 """Insert an element at the position `index`. 755 756 Example 757 -------- 758 ```py 759 vec = Vec((2, 3)) 760 vec.insert(0, 1) 761 assert vec == [1, 2, 3] 762 ``` 763 """ 764 self.__setitem__(index, value) 765 766 def pop(self, index: int = -1) -> _option.Option[T]: 767 """Removes the last element from a vector and returns it, or `None` if it is empty. 768 769 Example 770 ------- 771 ```py 772 vec = Vec((1, 2, 3)) 773 assert vec.pop() == Some(3) 774 assert vec == [1, 2] 775 ``` 776 """ 777 if not self._ptr: 778 return _option.NOTHING 779 780 return _option.Some(self._ptr.pop(index)) 781 782 def pop_if(self, pred: collections.Callable[[T], bool]) -> _option.Option[T]: 783 """Removes the last element from a vector and returns it if `f` returns `True`, 784 or `None` if it is empty. 785 786 Example 787 ------- 788 ```py 789 vec = Vec((1, 2, 3)) 790 assert vec.pop_if(lambda num: num * 2 == 6) == Some(3) 791 assert vec == [1, 2] 792 ``` 793 """ 794 if not self._ptr: 795 return _option.NOTHING 796 797 if pred(self[-1]): 798 return self.pop() 799 800 return _option.NOTHING 801 802 def dedup(self) -> None: 803 """Removes consecutive repeated elements in the vector according to their `__eq__` 804 implementation. 805 806 Example 807 ------- 808 ```py 809 vec = Vec([1, 2, 2, 3, 2]) 810 vec.dedup() 811 assert vec == [1, 2, 3, 2] 812 """ 813 self.dedup_by(lambda a, b: a == b) 814 815 def dedup_by(self, same_bucket: collections.Callable[[T, T], bool]) -> None: 816 """Removes all but the first of consecutive elements in the vector satisfying a given equality. 817 818 Example 819 ------- 820 ```py 821 vec = Vec(["foo", "bar", "Bar", "baz", "bar"]) 822 vec.dedup_by(lambda a, b: a.lower() == b.lower()) 823 assert vec == ["foo", "bar", "baz", "bar"] 824 ``` 825 826 Only remove consecutive duplicates. 827 ```py 828 vec = Vec([1, 2, 2, 3, 4, 1]) 829 vec.dedup_by(lambda a, b: a == b) 830 # The final 1 is not adjacent to the first 1, so it is not removed. 831 assert vec == [1, 2, 3, 4, 1] 832 ``` 833 """ 834 835 if not self._ptr or (len_ := len(self._ptr)) <= 1: 836 return 837 838 idx = 1 839 while idx < len_: 840 if same_bucket(self._ptr[idx], self._ptr[idx - 1]): 841 del self._ptr[idx] 842 len_ -= 1 843 else: 844 idx += 1 845 846 def remove(self, item: T) -> None: 847 """Remove the first appearance of `item` from this vector. 848 849 Example 850 ------- 851 ```py 852 vec = Vec(('a', 'b', 'c')) 853 vec.remove('a') 854 assert vec == ['b', 'c'] 855 ``` 856 """ 857 if not self._ptr: 858 return 859 860 self._ptr.remove(item) 861 862 def extend(self, iterable: collections.Iterable[T]) -> None: 863 """Extend this vector from another iterable. 864 865 Example 866 ------- 867 ```py 868 vec = Vec((1, 2, 3)) 869 vec.extend((4, 5, 6)) 870 871 assert vec == [1, 2, 3, 4, 5, 6] 872 ``` 873 """ 874 if self._ptr is None: 875 self._ptr = [] 876 877 self._ptr.extend(iterable) 878 879 def copy(self) -> Vec[T]: 880 """Copy all elements of `self` into a new vector. 881 882 Example 883 ------- 884 ```py 885 original = Vec((1,2,3)) 886 copy = original.copy() 887 copy.push(4) 888 889 print(original) # [1, 2, 3] 890 ``` 891 """ 892 return Vec(self._ptr[:]) if self._ptr else Vec() 893 894 def clear(self) -> None: 895 """Clear all elements of this vector. 896 897 Example 898 ------- 899 ```py 900 vec = Vec((1,2,3)) 901 vec.clear() 902 assert vec.len() == 0 903 ``` 904 """ 905 if not self._ptr: 906 return 907 908 self._ptr.clear() 909 910 def sort( 911 self, 912 *, 913 key: collections.Callable[[T], SupportsRichComparison] | None = None, 914 reverse: bool = False, 915 ) -> None: 916 """This method sorts the list in place, using only < comparisons between items. 917 918 Example 919 ------- 920 ```py 921 vec = Vec((2, 1, 3)) 922 vec.sort() 923 assert vec == [1, 2, 3] 924 ``` 925 """ 926 if not self._ptr: 927 return 928 929 # key can be `None` here just fine, idk why pyright is complaining. 930 self._ptr.sort(key=key, reverse=reverse) # pyright: ignore 931 932 def index( 933 self, item: T, start: typing.SupportsIndex = 0, end: int = _sys.maxsize 934 ) -> int: 935 # << Official documentation >> 936 """Return zero-based index in the vec of the first item whose value is equal to `item`. 937 Raises a ValueError if there is no such item. 938 939 Example 940 ------- 941 ```py 942 vec = Vec((1, 2, 3)) 943 assert vec.index(2) == 1 944 ``` 945 """ 946 if self._ptr is None: 947 raise ValueError from None 948 949 return self._ptr.index(item, start, end) 950 951 def count(self, item: T) -> int: 952 """Return the number of occurrences of `item` in the vec. 953 954 `0` is returned if the vector is empty or hasn't been initialized, as well if them item not found. 955 956 Example 957 -------- 958 ```py 959 vec = Vec((1, 2, 3, 3)) 960 assert vec.count(3) == 2 961 ``` 962 """ 963 if self._ptr is None: 964 return 0 965 966 return self._ptr.count(item) 967 968 def __len__(self) -> int: 969 return len(self._ptr) if self._ptr else 0 970 971 def __setitem__(self, index: int, value: T): 972 if not self._ptr: 973 raise IndexError from None 974 975 self._ptr[index] = value 976 977 @typing.overload 978 def __getitem__(self, index: slice) -> Vec[T]: ... 979 980 @typing.overload 981 def __getitem__(self, index: int) -> T: ... 982 983 def __getitem__(self, index: int | slice) -> T | Vec[T]: 984 if not self._ptr: 985 raise IndexError("Index out of range") 986 987 if isinstance(index, slice): 988 return Vec(self._ptr[index]) 989 990 return self._ptr[index] 991 992 def __delitem__(self, index: int) -> None: 993 if not self._ptr: 994 return 995 996 del self._ptr[index] 997 998 def __contains__(self, element: T) -> bool: 999 return element in self._ptr if self._ptr else False 1000 1001 def __iter__(self) -> collections.Iterator[T]: 1002 if self._ptr is None: 1003 return iter(()) 1004 1005 return self._ptr.__iter__() 1006 1007 def __repr__(self) -> str: 1008 return "[]" if not self._ptr else repr(self._ptr) 1009 1010 def __eq__(self, other: Vec[T] | list[T]) -> bool: 1011 if isinstance(other, Vec): 1012 return self._ptr == other._ptr 1013 1014 return self._ptr == other 1015 1016 def __ne__(self, other: Vec[T] | list[T]) -> bool: 1017 return not self.__eq__(other) 1018 1019 def __le__(self, other: list[T]) -> bool: 1020 if not self._ptr: 1021 return False 1022 1023 return self._ptr <= other 1024 1025 def __ge__(self, other: list[T]) -> bool: 1026 if not self._ptr: 1027 return False 1028 1029 return self._ptr >= other 1030 1031 def __lt__(self, other: list[T]) -> bool: 1032 if not self._ptr: 1033 return False 1034 1035 return self._ptr < other 1036 1037 def __gt__(self, other: list[T]) -> bool: 1038 if not self._ptr: 1039 return False 1040 1041 return self._ptr > other 1042 1043 def __bool__(self) -> bool: 1044 return bool(self._ptr) 1045 1046 def __reversed__(self) -> collections.Iterator[T]: 1047 return reversed(self._ptr or ())
A contiguous growable alternative to builtin list with extra functionalities.
The layout of Vec<T> is exactly the same as list<T>.
Which means list<T>'s methods are inherited into Vec<T>.
Example
names = Vec()
names.push('foo')
names.push('bar')
print(names) # ['foo', 'bar']
assert names.len() == 2
Constructing
Vec(): Create an empty vec, this will not allocate the initial buffer.Vec(other_list): Create a vec which points toother_listVec([1, 2, 3]): Create a vec with[1, 2, 3]pre-allocated.Vec(iterable): Create a vec from aniterable, This isO(n)wherenis the number of elements, since it copiesiterable's elements into a new list.Vec.with_capacity(5): Create a vec that can hold up to 5 elements
Iterating over Vec
There're two ways to iterate over a Vec. The first is to normally use for loop.
for i in names:
print(i)
# foo
# bar
The second is to use Vec.iter, which yields all items in this Vec from start to end.
Then the iterator gets exhausted as usual, See sain.Iterator.
iterator = names.iter()
for name in iterator.map(str.upper):
print(name)
# FOO
# BAR
# No more items, The actual vec is left unchanged.
assert iterator.next().is_none()
Comparison operators
A Vec may be compared with another Vec or a list, any other type will return False
vec = Vec([1, 2, 3])
assert vec == [1, 2, 3] # True
assert vec != (1, 2, 3)
Zero-Copy
A vec that gets initialized from a list will point to it and doesn't copy it.
So any element that gets appended to the list will also get pushed into the vec
that's pointing to it and vice-versa.
cells: list[str] = []
vec = Vec(cells) # This DOES NOT copy the `cells`.
cells.append("foo")
vec[0] == "foo" # True
The opposite of the above is to initialize the vec from either an iterable or args, or copy the list.
from sain.collections import vec, Vec
# Copy the list into a vec.
vec = Vec(cells[:])
cells.append("bar")
vec[1] # IndexError: "bar" doesn't exist in vec.
Implementations
This class implements Vec in Rust.
Vec(iterable: Iterable[~T] | None = None)
170 def __init__(self, iterable: collections.Iterable[T] | None = None) -> None: 171 """Create an empty `Vec<T>`. 172 173 The vector will not allocate until elements are pushed onto it. 174 175 Example 176 ------ 177 ```py 178 vec: Vec[float] = Vec() 179 ``` 180 """ 181 # We won't allocate to build the list here. 182 # Instead, On first push or fist indexed set 183 # we allocate if it was None. 184 if isinstance(iterable, list): 185 # Calling `list()` on another list will copy it, So instead we just point to it. 186 self._ptr = iterable 187 elif isinstance(iterable, Vec): 188 self._ptr = iterable._ptr 189 # any other iterable that ain't a list needs to get copied into a new list. 190 else: 191 self._ptr: list[T] | None = list(iterable) if iterable else None 192 193 self._capacity: int | None = None
Create an empty Vec<T>.
The vector will not allocate until elements are pushed onto it.
Example
vec: Vec[float] = Vec()
195 @classmethod 196 def with_capacity(cls, capacity: int) -> Vec[T]: 197 """Create a new `Vec` with at least the specified capacity. 198 This vec will be able to hold `capacity` elements without pushing further. 199 200 Check out `Vec.push_within_capacity` as well. 201 202 Example 203 ------- 204 ```py 205 vec = Vec.with_capacity(3) 206 assert vec.len() == 0 and vec.capacity() >= 3 207 208 vec.push(1) 209 vec.push(2) 210 vec.push(3) 211 print(vec.len()) # 3 212 213 # This won't push. 214 vec.push(4) 215 ``` 216 """ 217 v = cls() 218 v._capacity = capacity 219 return v
Create a new Vec with at least the specified capacity.
This vec will be able to hold capacity elements without pushing further.
Check out Vec.push_within_capacity as well.
Example
vec = Vec.with_capacity(3)
assert vec.len() == 0 and vec.capacity() >= 3
vec.push(1)
vec.push(2)
vec.push(3)
print(vec.len()) # 3
# This won't push.
vec.push(4)
221 def as_ref(self) -> Slice[T]: 222 """Return an immutable view over this vector elements. 223 224 No copying occurs. 225 226 Example 227 ------- 228 ```py 229 def send_bytes(v: Slice[int]) -> None: 230 # access `vec` in a read-only mode. 231 socket.write_all(v) 232 233 buf = Vec([1, 2, 3]) 234 send_bytes(buf.as_ref()) 235 ``` 236 """ 237 return Slice(self)
Return an immutable view over this vector elements.
No copying occurs.
Example
def send_bytes(v: Slice[int]) -> None:
# access `vec` in a read-only mode.
socket.write_all(v)
buf = Vec([1, 2, 3])
send_bytes(buf.as_ref())
239 def as_mut(self) -> SliceMut[T]: 240 """Return a mutable view over this vector elements. 241 242 No copying occurs. 243 244 Example 245 ------- 246 ```py 247 def read_bytes(buf: SliceMut[int]) -> None: 248 socket.read_to_end(buf.as_mut()) 249 250 buf: Vec[int] = Vec() 251 read_bytes(buf.as_mut()) # or just `buf` 252 assert vec.len() >= 1 253 ``` 254 """ 255 return SliceMut(self)
Return a mutable view over this vector elements.
No copying occurs.
Example
def read_bytes(buf: SliceMut[int]) -> None:
socket.read_to_end(buf.as_mut())
buf: Vec[int] = Vec()
read_bytes(buf.as_mut()) # or just `buf`
assert vec.len() >= 1
def
len(self) -> int:
257 def len(self) -> int: 258 """Return the number of elements in this vector. 259 260 Example 261 ------- 262 ```py 263 vec = Vec((1,2,3)) 264 265 assert vec.len() == 3 266 ``` 267 """ 268 return self.__len__()
Return the number of elements in this vector.
Example
vec = Vec((1,2,3))
assert vec.len() == 3
def
capacity(self) -> int:
270 def capacity(self) -> int: 271 """Return the capacity of this vector if set, 0 if not . 272 273 The number `0` here has two different Meanings: 274 - The first means the `Vec` doesn't have a specific capacity set. 275 - The second means the `Vec` literally initialized with `0` capacity. 276 277 Example 278 ------- 279 ```py 280 vec_with_cap = Vec.with_capacity(3) 281 assert vec_with_cap.capacity().unwrap() == 3 282 283 vec = Vec([1, 2, 3]) 284 assert vec.capacity() == 0 285 ``` 286 """ 287 return 0 if self._capacity is None else self._capacity
Return the capacity of this vector if set, 0 if not .
The number 0 here has two different Meanings:
- The first means the
Vecdoesn't have a specific capacity set. - The second means the
Vecliterally initialized with0capacity.
Example
vec_with_cap = Vec.with_capacity(3)
assert vec_with_cap.capacity().unwrap() == 3
vec = Vec([1, 2, 3])
assert vec.capacity() == 0
289 def iter(self) -> _iter.TrustedIter[T]: 290 """Returns an iterator over this vector elements. 291 292 Example 293 ------- 294 ```py 295 vec = Vec([1 ,2, 3]) 296 for element in vec.iter().map(str): 297 print(element) 298 ``` 299 """ 300 return _iter.TrustedIter(self._ptr or ())
Returns an iterator over this vector elements.
Example
vec = Vec([1 ,2, 3])
for element in vec.iter().map(str):
print(element)
def
is_empty(self) -> bool:
302 def is_empty(self) -> bool: 303 """Returns true if the vector contains no elements. 304 305 Inlined into: 306 ```py 307 not self.vec 308 ``` 309 """ 310 return not self._ptr
Returns true if the vector contains no elements.
Inlined into:
not self.vec
def
leak(self) -> list[~T]:
312 def leak(self) -> list[T]: 313 """Consumes and leaks the Vec, returning a mutable reference to the contents. 314 315 After calling this, this vec will no longer reference the underlying buffer, 316 therefore, it becomes unusable. 317 318 if `self` is uninitialized, an empty list is returned. 319 320 This function is only useful when you want to detach from a `Vec<T>` and get a `list[T]` without 321 any copies. 322 323 Example 324 ------- 325 ```py 326 x = Vec([1, 2, 3]) 327 owned = x.leak() 328 # x is now unusable. 329 owned[0] += 1 330 assert owned == [2, 2, 3] 331 ``` 332 """ 333 if self._ptr is not None: 334 # don't point to `_ptr` anymore. 335 tmp = self._ptr 336 del self._ptr 337 return tmp 338 339 return []
Consumes and leaks the Vec, returning a mutable reference to the contents.
After calling this, this vec will no longer reference the underlying buffer, therefore, it becomes unusable.
if self is uninitialized, an empty list is returned.
This function is only useful when you want to detach from a Vec<T> and get a list[T] without
any copies.
Example
x = Vec([1, 2, 3])
owned = x.leak()
# x is now unusable.
owned[0] += 1
assert owned == [2, 2, 3]
341 def split_off(self, at: int) -> Vec[T]: 342 """Split the vector off at the specified position, returning a new 343 vec at the range of `[at : len]`, leaving `self` at `[at : vec_len]`. 344 345 if this vec is empty, `self` is returned unchanged. 346 347 Example 348 ------- 349 ```py 350 origin = Vec((1, 2, 3, 4)) 351 split = vec.split_off(2) 352 353 print(origin, split) # [1, 2], [3, 4] 354 ``` 355 356 Raises 357 ------ 358 `IndexError` 359 This method will raise if `at` > `len(self)` 360 """ 361 len_ = self.len() 362 if at > len_: 363 raise IndexError( 364 f"Index `at` ({at}) should be <= than len of vector ({len_}) " 365 ) from None 366 367 # Either the list is empty or uninit. 368 if not self._ptr: 369 return self 370 371 split = self[at:len_] # split the items into a new vec. 372 del self._ptr[at:len_] # remove the items from the original list. 373 return split
Split the vector off at the specified position, returning a new
vec at the range of [at : len], leaving self at [at : vec_len].
if this vec is empty, self is returned unchanged.
Example
origin = Vec((1, 2, 3, 4))
split = vec.split_off(2)
print(origin, split) # [1, 2], [3, 4]
Raises
IndexError: This method will raise ifat>len(self)
def
split_first(self) -> '_option.Option[tuple[T, collections.Sequence[T]]]':
375 def split_first(self) -> _option.Option[tuple[T, collections.Sequence[T]]]: 376 """Split the first and rest elements of the vector, If empty, `None` is returned. 377 378 Example 379 ------- 380 ```py 381 vec = Vec([1, 2, 3]) 382 split = vec.split_first() 383 assert split == Some((1, [2, 3])) 384 385 vec: Vec[int] = Vec() 386 split = vec.split_first() 387 assert split.is_none() 388 ``` 389 """ 390 if not self._ptr: 391 return _option.NOTHING 392 393 # optimized to only one element in the vector. 394 if self.len() == 1: 395 return _option.Some((self[0], ())) 396 397 first, *rest = self._ptr 398 return _option.Some((first, rest))
Split the first and rest elements of the vector, If empty, None is returned.
Example
vec = Vec([1, 2, 3])
split = vec.split_first()
assert split == Some((1, [2, 3]))
vec: Vec[int] = Vec()
split = vec.split_first()
assert split.is_none()
def
split_last(self) -> '_option.Option[tuple[T, collections.Sequence[T]]]':
400 def split_last(self) -> _option.Option[tuple[T, collections.Sequence[T]]]: 401 """Split the last and rest elements of the vector, If empty, `None` is returned. 402 403 Example 404 ------- 405 ```py 406 vec = Vec([1, 2, 3]) 407 last, rest = vec.split_last().unwrap() 408 assert (last, rest) == [3, [1, 2]] 409 ``` 410 """ 411 if not self._ptr: 412 return _option.NOTHING 413 414 # optimized to only one element in the vector. 415 if self.len() == 1: 416 return _option.Some((self[0], ())) 417 418 last, *rest = self._ptr[-1], *self._ptr[:-1] 419 return _option.Some((last, rest))
Split the last and rest elements of the vector, If empty, None is returned.
Example
vec = Vec([1, 2, 3])
last, rest = vec.split_last().unwrap()
assert (last, rest) == [3, [1, 2]]
421 def split_at(self, mid: int) -> tuple[Vec[T], Vec[T]]: 422 """Divide `self` into two at an index. 423 424 The first will contain all elements from `[0:mid]` excluding `mid` it self. 425 and the second will contain the remaining elements. 426 427 if `mid` > `self.len()`, Then all elements will be moved to the left, 428 returning an empty vec in right. 429 430 Example 431 ------- 432 ```py 433 buffer = Vec((1, 2, 3, 4)) 434 left, right = buffer.split_at(0) 435 assert left == [] and right == [1, 2, 3, 4] 436 437 left, right = buffer.split_at(2) 438 assert left == [1, 2] and right == [2, 3] 439 ``` 440 441 The is roughly the implementation 442 ```py 443 vec[0:mid], vec[mid:] 444 ``` 445 """ 446 return self[0:mid], self[mid:]
Divide self into two at an index.
The first will contain all elements from [0:mid] excluding mid it self.
and the second will contain the remaining elements.
if mid > self.len(), Then all elements will be moved to the left,
returning an empty vec in right.
Example
buffer = Vec((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
vec[0:mid], vec[mid:]
def
swap(self, a: int, b: int):
448 def swap(self, a: int, b: int): 449 """Swap two elements in the vec. 450 451 if `a` equals to `b` then it's guaranteed that elements won't change value. 452 453 Example 454 ------- 455 ```py 456 buf = Vec([1, 2, 3, 4]) 457 buf.swap(0, 3) 458 assert buf == [4, 2, 3, 1] 459 ``` 460 461 Raises 462 ------ 463 IndexError 464 If the positions of `a` or `b` are out of index. 465 """ 466 if self[a] == self[b]: 467 return 468 469 self[a], self[b] = self[b], self[a]
Swap two elements in the vec.
if a equals to b then it's guaranteed that elements won't change value.
Example
buf = Vec([1, 2, 3, 4])
buf.swap(0, 3)
assert buf == [4, 2, 3, 1]
Raises
- IndexError: If the positions of
aorbare out of index.
def
swap_unchecked(self, a: int, b: int):
471 def swap_unchecked(self, a: int, b: int): 472 """Swap two elements in the vec. without checking if `a` == `b`. 473 474 If you care about `a` and `b` equality, see `Vec.swap`. 475 476 Example 477 ------- 478 ```py 479 buf = Vec([1, 2, 3, 1]) 480 buf.swap_unchecked(0, 3) 481 assert buf == [1, 2, 3, 1] 482 ``` 483 484 Raises 485 ------ 486 IndexError 487 If the positions of `a` or `b` are out of index. 488 """ 489 self[a], self[b] = self[b], self[a]
Swap two elements in the vec. without checking if a == b.
If you care about a and b equality, see Vec.swap.
Example
buf = Vec([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.
def
first(self) -> '_option.Option[T]':
491 def first(self) -> _option.Option[T]: 492 """Get the first element in this vec, returning `None` if there's none. 493 494 Example 495 ------- 496 ```py 497 vec = Vec((1,2,3)) 498 first = vec.first() 499 assert ~first == 1 500 ``` 501 """ 502 return self.get(0)
Get the first element in this vec, returning None if there's none.
Example
vec = Vec((1,2,3))
first = vec.first()
assert ~first == 1
def
last(self) -> '_option.Option[T]':
504 def last(self) -> _option.Option[T]: 505 """Get the last element in this vec, returning `None` if there's none. 506 507 Example 508 ------- 509 ```py 510 vec = Vec([1, 2, 3, 4]) 511 first = vec.last() 512 assert ~first == 4 513 ``` 514 """ 515 return self.get(-1)
Get the last element in this vec, returning None if there's none.
Example
vec = Vec([1, 2, 3, 4])
first = vec.last()
assert ~first == 4
def
truncate(self, size: int) -> None:
517 def truncate(self, size: int) -> None: 518 """Shortens the vec, keeping the first `size` elements and dropping the rest. 519 520 Example 521 ------- 522 ```py 523 vec = Vec([1,2,3]) 524 vec.truncate(1) 525 assert vec == [1] 526 ``` 527 """ 528 if not self._ptr: 529 return 530 531 del self._ptr[size:]
Shortens the vec, keeping the first size elements and dropping the rest.
Example
vec = Vec([1,2,3])
vec.truncate(1)
assert vec == [1]
def
retain(self, f: Callable[[~T], bool]) -> None:
533 def retain(self, f: collections.Callable[[T], bool]) -> None: 534 """Retains only the elements specified by the predicate. 535 536 This operation occurs in-place without copying the original list. 537 538 Example 539 ------- 540 ```py 541 vec = Vec([1, 2, 3]) 542 vec.retain(lambda elem: elem > 1) 543 544 assert vec == [2, 3] 545 ``` 546 547 The impl for this method is very simple 548 ```py 549 for idx, e in enumerate(vec): 550 if not f(e): 551 del vec[idx] 552 ``` 553 """ 554 if not self._ptr: 555 return 556 557 idx = 0 558 while idx < len(self._ptr): 559 if not f(self._ptr[idx]): 560 del self._ptr[idx] 561 else: 562 idx += 1
Retains only the elements specified by the predicate.
This operation occurs in-place without copying the original list.
Example
vec = Vec([1, 2, 3])
vec.retain(lambda elem: elem > 1)
assert vec == [2, 3]
The impl for this method is very simple
for idx, e in enumerate(vec):
if not f(e):
del vec[idx]
def
swap_remove(self, item: ~T) -> ~T:
564 def swap_remove(self, item: T) -> T: 565 """Remove the first appearance of `item` from this vector and return it. 566 567 Raises 568 ------ 569 * `ValueError`: if `item` is not in this vector. 570 * `MemoryError`: if this vector hasn't allocated, Aka nothing has been pushed to it. 571 572 Example 573 ------- 574 ```py 575 vec = Vec(('a', 'b', 'c')) 576 element = vec.remove('a') 577 assert vec == ['b', 'c'] and element == 'a' 578 ``` 579 """ 580 if self._ptr is None: 581 raise MemoryError("Vec is unallocated.") from None 582 583 return self._ptr.pop(self.index(item))
Remove the first appearance of item from this vector and return it.
Raises
*
ValueError(ifitemis not in this vector.):*
MemoryError(if this vector hasn't allocated, Aka nothing has been pushed to it.):
Example
vec = Vec(('a', 'b', 'c'))
element = vec.remove('a')
assert vec == ['b', 'c'] and element == 'a'
def
fill(self, value: ~T) -> None:
585 def fill(self, value: T) -> None: 586 """Fill `self` with the given `value`. 587 588 Nothing happens if the vec is empty or unallocated. 589 590 Example 591 ```py 592 a = Vec([0, 1, 2, 3]) 593 a.fill(0) 594 assert a == [0, 0, 0, 0] 595 ``` 596 """ 597 if not self._ptr: 598 return 599 600 for n, _ in enumerate(self): 601 self[n] = value
Fill self with the given value.
Nothing happens if the vec is empty or unallocated.
Example
a = Vec([0, 1, 2, 3])
a.fill(0)
assert a == [0, 0, 0, 0]
def
push(self, item: ~T) -> None:
603 def push(self, item: T) -> None: 604 """Push an element at the end of the vector. 605 606 Example 607 ------- 608 ```py 609 vec = Vec() 610 vec.push(1) 611 612 assert vec == [1] 613 ``` 614 """ 615 if self._capacity is not None: 616 self.push_within_capacity(item) 617 return 618 619 if self._ptr is None: 620 self._ptr = [] 621 622 self._ptr.append(item)
Push an element at the end of the vector.
Example
vec = Vec()
vec.push(1)
assert vec == [1]
def
push_within_capacity(self, x: ~T) -> 'Result[None, T]':
624 def push_within_capacity(self, x: T) -> Result[None, T]: 625 """Appends an element if there is sufficient spare capacity, otherwise an error is returned with the element. 626 627 Example 628 ------- 629 ```py 630 vec: Vec[int] = Vec.with_capacity(3) 631 for i in range(3): 632 match vec.push_within_capacity(i): 633 case Ok(_): 634 print("All good.") 635 case Err(elem): 636 print("Reached max cap :< can't push", elem) 637 ``` 638 639 Or you can also just call `Vec.push` and it will push if there's is sufficient capacity. 640 ```py 641 vec: Vec[int] = Vec.with_capacity(3) 642 643 vec.extend((1, 2, 3)) 644 vec.push(4) 645 646 assert vec.len() == 3 647 ``` 648 """ 649 if self._ptr is None: 650 self._ptr = [] 651 652 if self.len() == self._capacity: 653 return _result.Err(x) 654 655 self._ptr.append(x) 656 return _result.Ok(None)
Appends an element if there is sufficient spare capacity, otherwise an error is returned with the element.
Example
vec: Vec[int] = Vec.with_capacity(3)
for i in range(3):
match vec.push_within_capacity(i):
case Ok(_):
print("All good.")
case Err(elem):
print("Reached max cap :< can't push", elem)
Or you can also just call Vec.push and it will push if there's is sufficient capacity.
vec: Vec[int] = Vec.with_capacity(3)
vec.extend((1, 2, 3))
vec.push(4)
assert vec.len() == 3
def
reserve(self, additional: int) -> None:
658 def reserve(self, additional: int) -> None: 659 """Reserves capacity for at least additional more elements to be inserted in the given Vec<T>. 660 661 Example 662 ------- 663 ```py 664 vec = Vec.with_capacity(3) 665 is_vip = random.choice((True, False)) 666 667 for i in range(4): 668 match vec.push_within_capacity(i): 669 case Ok(_): 670 print("All good") 671 case Err(person): 672 # If the person is a VIP, then reserve for one more. 673 if is_vip: 674 vec.reserve(1) 675 continue 676 677 # is_vip was false. 678 print("Can't reserve for non-VIP members...", person) 679 break 680 ``` 681 """ 682 if self._capacity is not None: 683 self._capacity += additional
Reserves capacity for at least additional more elements to be inserted in the given Vec
Example
vec = Vec.with_capacity(3)
is_vip = random.choice((True, False))
for i in range(4):
match vec.push_within_capacity(i):
case Ok(_):
print("All good")
case Err(person):
# If the person is a VIP, then reserve for one more.
if is_vip:
vec.reserve(1)
continue
# is_vip was false.
print("Can't reserve for non-VIP members...", person)
break
def
shrink_to_fit(self) -> None:
685 def shrink_to_fit(self) -> None: 686 """Shrink the capacity of this `Vec` to match its length. 687 688 If `self` is initialized with no capacity, This is a `NOP`. 689 690 Example 691 ------- 692 ```py 693 s = Vec([1, 2, 3, 4]) 694 695 s.reserve(100) 696 s.shrink_to_fit() 697 assert s.capacity() == 4 698 ``` 699 """ 700 if self._capacity is None: 701 return 702 703 # The capacity is never less than the length. 704 self._capacity = min(self.__len__(), self._capacity)
Shrink the capacity of this Vec to match its length.
If self is initialized with no capacity, This is a NOP.
Example
s = Vec([1, 2, 3, 4])
s.reserve(100)
s.shrink_to_fit()
assert s.capacity() == 4
def
shrink_to(self, min_capacity: int) -> None:
706 def shrink_to(self, min_capacity: int) -> None: 707 """Shrink the capacity of this `Vec` to a minimum specified capacity. 708 709 If `self` is initialized with no capacity or the current capacity is less than the lower limit, 710 This is a `NOP`. 711 712 Example 713 ------- 714 ```py 715 vec = Vec.with_capacity(10) 716 vec.extend([1, 2, 3]) 717 assert vec.capacity() >= 10 718 vec.shrink_to(4) 719 assert vec.capacity() >= 4 720 vec.shrink_to(0) 721 ``` 722 """ 723 if self._capacity is None or self._capacity <= min_capacity: 724 return 725 726 # Ensure the capacity is not reduced below the current length of the vector. 727 self._capacity = max(self.__len__(), min_capacity)
Shrink the capacity of this Vec to a minimum specified capacity.
If self is initialized with no capacity or the current capacity is less than the lower limit,
This is a NOP.
Example
vec = Vec.with_capacity(10)
vec.extend([1, 2, 3])
assert vec.capacity() >= 10
vec.shrink_to(4)
assert vec.capacity() >= 4
vec.shrink_to(0)
def
get(self, index: int) -> '_option.Option[T]':
737 def get(self, index: int) -> _option.Option[T]: 738 """Get the item at the given index, or `Some[None]` if its out of bounds. 739 740 Example 741 ------- 742 ```py 743 vec = Vec((1, 2, 3)) 744 vec.get(0) == Some(1) 745 vec.get(3) == Some(None) 746 ``` 747 """ 748 try: 749 return _option.Some(self[index]) 750 except IndexError: 751 return _option.NOTHING
Get the item at the given index, or Some[None] if its out of bounds.
Example
vec = Vec((1, 2, 3))
vec.get(0) == Some(1)
vec.get(3) == Some(None)
def
insert(self, index: int, value: ~T) -> None:
753 def insert(self, index: int, value: T) -> None: 754 """Insert an element at the position `index`. 755 756 Example 757 -------- 758 ```py 759 vec = Vec((2, 3)) 760 vec.insert(0, 1) 761 assert vec == [1, 2, 3] 762 ``` 763 """ 764 self.__setitem__(index, value)
Insert an element at the position index.
Example
vec = Vec((2, 3))
vec.insert(0, 1)
assert vec == [1, 2, 3]
def
pop(self, index: int = -1) -> '_option.Option[T]':
766 def pop(self, index: int = -1) -> _option.Option[T]: 767 """Removes the last element from a vector and returns it, or `None` if it is empty. 768 769 Example 770 ------- 771 ```py 772 vec = Vec((1, 2, 3)) 773 assert vec.pop() == Some(3) 774 assert vec == [1, 2] 775 ``` 776 """ 777 if not self._ptr: 778 return _option.NOTHING 779 780 return _option.Some(self._ptr.pop(index))
Removes the last element from a vector and returns it, or None if it is empty.
Example
vec = Vec((1, 2, 3))
assert vec.pop() == Some(3)
assert vec == [1, 2]
def
pop_if(self, pred: Callable[[~T], bool]) -> '_option.Option[T]':
782 def pop_if(self, pred: collections.Callable[[T], bool]) -> _option.Option[T]: 783 """Removes the last element from a vector and returns it if `f` returns `True`, 784 or `None` if it is empty. 785 786 Example 787 ------- 788 ```py 789 vec = Vec((1, 2, 3)) 790 assert vec.pop_if(lambda num: num * 2 == 6) == Some(3) 791 assert vec == [1, 2] 792 ``` 793 """ 794 if not self._ptr: 795 return _option.NOTHING 796 797 if pred(self[-1]): 798 return self.pop() 799 800 return _option.NOTHING
Removes the last element from a vector and returns it if f returns True,
or None if it is empty.
Example
vec = Vec((1, 2, 3))
assert vec.pop_if(lambda num: num * 2 == 6) == Some(3)
assert vec == [1, 2]
def
dedup(self) -> None:
802 def dedup(self) -> None: 803 """Removes consecutive repeated elements in the vector according to their `__eq__` 804 implementation. 805 806 Example 807 ------- 808 ```py 809 vec = Vec([1, 2, 2, 3, 2]) 810 vec.dedup() 811 assert vec == [1, 2, 3, 2] 812 """ 813 self.dedup_by(lambda a, b: a == b)
Removes consecutive repeated elements in the vector according to their __eq__
implementation.
Example
```py vec = Vec([1, 2, 2, 3, 2]) vec.dedup() assert vec == [1, 2, 3, 2]
def
dedup_by(self, same_bucket: Callable[[~T, ~T], bool]) -> None:
815 def dedup_by(self, same_bucket: collections.Callable[[T, T], bool]) -> None: 816 """Removes all but the first of consecutive elements in the vector satisfying a given equality. 817 818 Example 819 ------- 820 ```py 821 vec = Vec(["foo", "bar", "Bar", "baz", "bar"]) 822 vec.dedup_by(lambda a, b: a.lower() == b.lower()) 823 assert vec == ["foo", "bar", "baz", "bar"] 824 ``` 825 826 Only remove consecutive duplicates. 827 ```py 828 vec = Vec([1, 2, 2, 3, 4, 1]) 829 vec.dedup_by(lambda a, b: a == b) 830 # The final 1 is not adjacent to the first 1, so it is not removed. 831 assert vec == [1, 2, 3, 4, 1] 832 ``` 833 """ 834 835 if not self._ptr or (len_ := len(self._ptr)) <= 1: 836 return 837 838 idx = 1 839 while idx < len_: 840 if same_bucket(self._ptr[idx], self._ptr[idx - 1]): 841 del self._ptr[idx] 842 len_ -= 1 843 else: 844 idx += 1
Removes all but the first of consecutive elements in the vector satisfying a given equality.
Example
vec = Vec(["foo", "bar", "Bar", "baz", "bar"])
vec.dedup_by(lambda a, b: a.lower() == b.lower())
assert vec == ["foo", "bar", "baz", "bar"]
Only remove consecutive duplicates.
vec = Vec([1, 2, 2, 3, 4, 1])
vec.dedup_by(lambda a, b: a == b)
# The final 1 is not adjacent to the first 1, so it is not removed.
assert vec == [1, 2, 3, 4, 1]
def
remove(self, item: ~T) -> None:
846 def remove(self, item: T) -> None: 847 """Remove the first appearance of `item` from this vector. 848 849 Example 850 ------- 851 ```py 852 vec = Vec(('a', 'b', 'c')) 853 vec.remove('a') 854 assert vec == ['b', 'c'] 855 ``` 856 """ 857 if not self._ptr: 858 return 859 860 self._ptr.remove(item)
Remove the first appearance of item from this vector.
Example
vec = Vec(('a', 'b', 'c'))
vec.remove('a')
assert vec == ['b', 'c']
def
extend(self, iterable: Iterable[~T]) -> None:
862 def extend(self, iterable: collections.Iterable[T]) -> None: 863 """Extend this vector from another iterable. 864 865 Example 866 ------- 867 ```py 868 vec = Vec((1, 2, 3)) 869 vec.extend((4, 5, 6)) 870 871 assert vec == [1, 2, 3, 4, 5, 6] 872 ``` 873 """ 874 if self._ptr is None: 875 self._ptr = [] 876 877 self._ptr.extend(iterable)
Extend this vector from another iterable.
Example
vec = Vec((1, 2, 3))
vec.extend((4, 5, 6))
assert vec == [1, 2, 3, 4, 5, 6]
879 def copy(self) -> Vec[T]: 880 """Copy all elements of `self` into a new vector. 881 882 Example 883 ------- 884 ```py 885 original = Vec((1,2,3)) 886 copy = original.copy() 887 copy.push(4) 888 889 print(original) # [1, 2, 3] 890 ``` 891 """ 892 return Vec(self._ptr[:]) if self._ptr else Vec()
Copy all elements of self into a new vector.
Example
original = Vec((1,2,3))
copy = original.copy()
copy.push(4)
print(original) # [1, 2, 3]
def
clear(self) -> None:
894 def clear(self) -> None: 895 """Clear all elements of this vector. 896 897 Example 898 ------- 899 ```py 900 vec = Vec((1,2,3)) 901 vec.clear() 902 assert vec.len() == 0 903 ``` 904 """ 905 if not self._ptr: 906 return 907 908 self._ptr.clear()
Clear all elements of this vector.
Example
vec = Vec((1,2,3))
vec.clear()
assert vec.len() == 0
def
sort( self, *, key: 'collections.Callable[[T], SupportsRichComparison] | None' = None, reverse: bool = False) -> None:
910 def sort( 911 self, 912 *, 913 key: collections.Callable[[T], SupportsRichComparison] | None = None, 914 reverse: bool = False, 915 ) -> None: 916 """This method sorts the list in place, using only < comparisons between items. 917 918 Example 919 ------- 920 ```py 921 vec = Vec((2, 1, 3)) 922 vec.sort() 923 assert vec == [1, 2, 3] 924 ``` 925 """ 926 if not self._ptr: 927 return 928 929 # key can be `None` here just fine, idk why pyright is complaining. 930 self._ptr.sort(key=key, reverse=reverse) # pyright: ignore
This method sorts the list in place, using only < comparisons between items.
Example
vec = Vec((2, 1, 3))
vec.sort()
assert vec == [1, 2, 3]
def
index( self, item: ~T, start: <class 'SupportsIndex'> = 0, end: int = 9223372036854775807) -> int:
932 def index( 933 self, item: T, start: typing.SupportsIndex = 0, end: int = _sys.maxsize 934 ) -> int: 935 # << Official documentation >> 936 """Return zero-based index in the vec of the first item whose value is equal to `item`. 937 Raises a ValueError if there is no such item. 938 939 Example 940 ------- 941 ```py 942 vec = Vec((1, 2, 3)) 943 assert vec.index(2) == 1 944 ``` 945 """ 946 if self._ptr is None: 947 raise ValueError from None 948 949 return self._ptr.index(item, start, end)
Return zero-based index in the vec of the first item whose value is equal to item.
Raises a ValueError if there is no such item.
Example
vec = Vec((1, 2, 3))
assert vec.index(2) == 1
def
count(self, item: ~T) -> int:
951 def count(self, item: T) -> int: 952 """Return the number of occurrences of `item` in the vec. 953 954 `0` is returned if the vector is empty or hasn't been initialized, as well if them item not found. 955 956 Example 957 -------- 958 ```py 959 vec = Vec((1, 2, 3, 3)) 960 assert vec.count(3) == 2 961 ``` 962 """ 963 if self._ptr is None: 964 return 0 965 966 return self._ptr.count(item)
Return the number of occurrences of item in the vec.
0 is returned if the vector is empty or hasn't been initialized, as well if them item not found.
Example
vec = Vec((1, 2, 3, 3))
assert vec.count(3) == 2
Inherited Members
@rustc_diagnostic_item('&[u8]')
@typing.final
class
Bytes109@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.
@classmethod
def
from_ptr(cls, arr: array.array[int]) -> Self:
170 @classmethod 171 def from_ptr(cls, arr: array.array[int]) -> Self: 172 """Create a new `Bytes` from an array. 173 174 The returned `Bytes` will directly point to `arr` without copying. 175 176 Example 177 ------- 178 ```py 179 arr = array.array("B", b"Hello") 180 buffer = Bytes.from_ptr(arr) 181 ``` 182 """ 183 # this is technically an `assert` line 184 # but Python isn't smart enough to inline and opt-out 185 # this out of the generated bytecode. 186 # so we'll just leave this under `if` statement. 187 if __debug__: 188 assert_precondition( 189 arr.typecode == "B", 190 f"array type must be `B`, not `{arr.typecode}`", 191 TypeError, 192 ) 193 194 b = cls() 195 b._buf = arr 196 return b
Create a new Bytes from an array.
The returned Bytes will directly point to arr without copying.
Example
arr = array.array("B", b"Hello")
buffer = Bytes.from_ptr(arr)
@classmethod
@unsafe
def
from_ptr_unchecked(cls, arr: array.array[int]) -> Self:
198 @classmethod 199 @unsafe 200 def from_ptr_unchecked(cls, arr: array.array[int]) -> Self: 201 """Create a new `Bytes` from an array, without checking the type code. 202 203 The returned `Bytes` will directly point to `arr` without copying. 204 205 ## Safety 206 207 The caller must ensure that `arr` is of type `array.array[int]` with type code `B`. 208 209 Example 210 ------- 211 ```py 212 arr = array.array("B", b"Hello") 213 buffer = Bytes.from_ptr_unchecked(arr) 214 ``` 215 """ 216 b = cls() 217 b._buf = arr 218 return b
Create a new Bytes from an array, without checking the type code.
The returned Bytes will directly point to arr without copying.
Safety
The caller must ensure that arr is of type array.array[int] with type code B.
Example
arr = array.array("B", b"Hello")
buffer = Bytes.from_ptr_unchecked(arr)
Safety ⚠️
Calling this method without knowing the output is considered undefined behavior.
@classmethod
def
from_bytes(cls, buf: bytes | bytearray | Iterable[int]) -> Self:
@classmethod
def
from_raw(cls, raw: _io.StringIO | _io.BytesIO | _io.BufferedReader) -> Self:
234 @classmethod 235 def from_raw(cls, raw: Rawish) -> Self: 236 """Initialize a new `Bytes` from a `Rawish` data type. 237 238 Example 239 ------- 240 ```py 241 with open('file.txt', 'rb') as file: 242 buff = Bytes.from_raw(file) 243 244 # in memory bytes io 245 bytes_io = io.BytesIO(b"data") 246 buffer1 = Bytes.from_raw(bytes_io) 247 # in memory string io 248 string_io = io.StringIO("data") 249 buffer2 = Bytes.from_raw(string_io) 250 ``` 251 """ 252 c = cls() 253 c._buf = array.array("B", unwrap_bytes(raw)) 254 return c
Initialize a new Bytes from a Rawish data type.
Example
with open('file.txt', 'rb') as file:
buff = Bytes.from_raw(file)
# in memory bytes io
bytes_io = io.BytesIO(b"data")
buffer1 = Bytes.from_raw(bytes_io)
# in memory string io
string_io = io.StringIO("data")
buffer2 = Bytes.from_raw(string_io)
@classmethod
def
zeroed(cls, count: int) -> Self:
256 @classmethod 257 def zeroed(cls, count: int) -> Self: 258 """Initialize a new `Bytes` filled with `0 * count`. 259 260 Example 261 ------- 262 ```py 263 ALLOC_SIZE = 1024 * 2 264 buffer = Bytes.zeros(ALLOC_SIZE) 265 assert buffer.len() == ALLOC_SIZE 266 ``` 267 """ 268 c = cls() 269 c._buf = array.array("B", [0] * count) 270 return c
Initialize a new Bytes filled with 0 * count.
Example
ALLOC_SIZE = 1024 * 2
buffer = Bytes.zeros(ALLOC_SIZE)
assert buffer.len() == ALLOC_SIZE
@deprecated(since='1.3.0', removed_in='2.0.0', use_instead='Bytes.to_bytes().decode("utf8")', hint='Converting a bytes object to string is fairly trivial.')
def
try_to_str(self) -> 'Result[str, bytes]':
291 @deprecated( 292 since="1.3.0", 293 removed_in="2.0.0", 294 use_instead='Bytes.to_bytes().decode("utf8")', 295 hint="Converting a bytes object to string is fairly trivial.", 296 ) 297 def try_to_str(self) -> Result[str, bytes]: 298 """A safe method to convert `self` into a string. 299 300 This may fail if the `self` contains invalid bytes. strings 301 needs to be valid utf-8. 302 303 Example 304 ------- 305 ```py 306 buf = Bytes() 307 sparkles_heart = [240, 159, 146, 150] 308 buf.put_bytes(sparkles_heart) 309 310 assert buf.try_to_str().unwrap() == "💖" 311 ``` 312 313 Incorrect bytes 314 --------------- 315 ```py 316 invalid_bytes = Bytes.from_bytes([0, 159, 146, 150]) 317 invalid_bytes.try_to_str().is_err() 318 ``` 319 320 Returns 321 ------- 322 `Result[str, bytes]` 323 If successful, returns the decoded string, otherwise the original bytes that failed 324 to get decoded. 325 """ 326 try: 327 return _result.Ok(self.to_bytes().decode(ENCODING)) 328 except UnicodeDecodeError as e: 329 return _result.Err(e.object)
A safe method to convert self into a string.
This may fail if the self contains invalid bytes. strings
needs to be valid utf-8.
Example
buf = Bytes()
sparkles_heart = [240, 159, 146, 150]
buf.put_bytes(sparkles_heart)
assert buf.try_to_str().unwrap() == "💖"
Incorrect bytes
invalid_bytes = Bytes.from_bytes([0, 159, 146, 150])
invalid_bytes.try_to_str().is_err()
Returns
Result[str, bytes]: If successful, returns the decoded string, otherwise the original bytes that failed to get decoded.
@deprecated(since='1.3.0', removed_in='2.0.0', use_instead='str(Bytes, encoding="utf-8")', hint='Converting a bytes object to string is fairly trivial.')
def
to_str(self) -> str:
331 @deprecated( 332 since="1.3.0", 333 removed_in="2.0.0", 334 use_instead='str(Bytes, encoding="utf-8")', 335 hint="Converting a bytes object to string is fairly trivial.", 336 ) 337 def to_str(self) -> str: 338 r"""Convert `self` to a utf-8 string. 339 340 During the conversion process, any invalid bytes will get converted to 341 [REPLACEMENT_CHARACTER](https://en.wikipedia.org/wiki/Specials_(Unicode_block)) 342 which looks like this `�`, so be careful on what you're trying to convert. 343 344 Use `.try_to_str` try attempt the conversion in case of failure. 345 346 Example 347 ------- 348 ```py 349 buf = Bytes() 350 sparkles_heart = [240, 159, 146, 150] 351 buf.put_bytes(sparkles_heart) 352 353 assert buf.to_str() == "💖" 354 ``` 355 356 Incorrect bytes 357 --------------- 358 ```py 359 invalid_bytes = Bytes.from_bytes(b"Hello \xf0\x90\x80World") 360 assert invalid_bytes.to_str() == "Hello �World" 361 ``` 362 """ 363 if not self._buf: 364 return "" 365 366 return self._buf.tobytes().decode(ENCODING, errors="replace")
Convert self to a utf-8 string.
During the conversion process, any invalid bytes will get converted to
REPLACEMENT_CHARACTER
which looks like this �, so be careful on what you're trying to convert.
Use .try_to_str try attempt the conversion in case of failure.
Example
buf = Bytes()
sparkles_heart = [240, 159, 146, 150]
buf.put_bytes(sparkles_heart)
assert buf.to_str() == "💖"
Incorrect bytes
invalid_bytes = Bytes.from_bytes(b"Hello \xf0\x90\x80World")
assert invalid_bytes.to_str() == "Hello �World"
def
to_bytes(self) -> bytes:
368 def to_bytes(self) -> bytes: 369 """Convert `self` into `bytes`, copying the underlying array into a new buffer. 370 371 Example 372 ------- 373 ```py 374 buf = Bytes.from_str("Hello") 375 assert buf.to_bytes() == b'Hello' 376 ``` 377 """ 378 if not self._buf: 379 return b"" 380 381 return self._buf.tobytes()
Convert self into bytes, copying the underlying array into a new buffer.
Example
buf = Bytes.from_str("Hello")
assert buf.to_bytes() == b'Hello'
383 def to_vec(self) -> _vec.Vec[int]: 384 """Copies `self` into a new `Vec`. 385 386 Example 387 ------- 388 ```py 389 buffer = Bytes.from_str([1, 2, 3, 4]) 390 # buffer and x can be modified independently. 391 x = buffer.to_vec() 392 """ 393 return _vec.Vec(self.copy())
Copies self into a new Vec.
Example
```py buffer = Bytes.from_str([1, 2, 3, 4])
buffer and x can be modified independently.
x = buffer.to_vec()
def
leak(self) -> array.array[int]:
395 def leak(self) -> array.array[int]: 396 """Consumes and leaks the `Bytes`, returning the contents as an `array[int]`, 397 398 A new empty array is returned if the underlying buffer is not initialized. 399 400 `self` will deallocate the underlying array, therefore it becomes unusable. 401 402 Safety 403 ------ 404 It is unsafe to access the leaked array from `self` after calling this function. 405 406 Example 407 ------- 408 ```py 409 bytes = Bytes.from_str("chunks of data") 410 consumed = bytes.leak() 411 # `bytes` doesn't point to anything, this is undefined behavior. 412 bytes.put(0) 413 # access the array directly instead. 414 consumed.tobytes() == b"chunks of data" 415 ``` 416 """ 417 if self._buf is None: 418 return array.array("B") 419 420 arr = self._buf 421 # We don't need to reference this anymore since the caller will own the array. 422 del self._buf 423 return arr
Consumes and leaks the Bytes, returning the contents as an array[int],
A new empty array is returned if the underlying buffer is not initialized.
self will deallocate the underlying array, therefore it becomes unusable.
Safety
It is unsafe to access the leaked array from self after calling this function.
Example
bytes = Bytes.from_str("chunks of data")
consumed = bytes.leak()
# `bytes` doesn't point to anything, this is undefined behavior.
bytes.put(0)
# access the array directly instead.
consumed.tobytes() == b"chunks of data"
def
as_ptr(self) -> 'memoryview[int]':
425 def as_ptr(self) -> memoryview[int]: 426 """Returns a read-only pointer to the buffer data. 427 428 `pointer` here refers to a `memoryview` object. 429 430 A `BufferError` is raised if the underlying sequence is not initialized. 431 432 Example 433 ------- 434 ```py 435 buffer = Bytes.from_bytes(b"data") 436 ptr = buffer.as_ptr() 437 ptr[0] = 1 # TypeError: cannot modify read-only memory 438 ``` 439 """ 440 return self.__buffer__(256).toreadonly()
Returns a read-only pointer to the buffer data.
pointer here refers to a memoryview object.
A BufferError is raised if the underlying sequence is not initialized.
Example
buffer = Bytes.from_bytes(b"data")
ptr = buffer.as_ptr()
ptr[0] = 1 # TypeError: cannot modify read-only memory
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]
def
raw_parts(self) -> tuple[int, int]:
493 def raw_parts( 494 self, 495 ) -> tuple[int, int]: 496 """Return `self` as tuple containing the memory address to the buffer and how many bytes it currently contains. 497 498 An alias to `array.buffer_into` 499 """ 500 if not self._buf: 501 return (0x0, 0) 502 503 return self._buf.buffer_info()
Return self as tuple containing the memory address to the buffer and how many bytes it currently contains.
An alias to array.buffer_into
def
len(self) -> int:
505 def len(self) -> int: 506 """Return the number of bytes in this buffer. 507 508 Example 509 ------- 510 ```py 511 buf = Bytes.from_bytes([240, 159, 146, 150]) 512 assert buf.len() == 4 513 ``` 514 """ 515 return self.__len__()
Return the number of bytes in this buffer.
Example
buf = Bytes.from_bytes([240, 159, 146, 150])
assert buf.len() == 4
def
size(self) -> int:
517 def size(self) -> int: 518 """The length in bytes of one array item in the internal representation. 519 520 521 An alias to `array.itemsize` 522 523 Example 524 ------- 525 ```py 526 buf = Bytes.from_bytes([240, 159, 146, 150]) 527 assert buf.size() == 1 528 ``` 529 """ 530 if not self._buf: 531 return 0 532 return self._buf.itemsize
The length in bytes of one array item in the internal representation.
An alias to array.itemsize
Example
buf = Bytes.from_bytes([240, 159, 146, 150])
assert buf.size() == 1
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')
def
is_empty(self) -> bool:
577 def is_empty(self) -> bool: 578 """Check whether `self` contains any bytes or not. 579 580 Example 581 ------- 582 ```py 583 buffer = Bytes() 584 assert buffer.is_empty() 585 ``` 586 """ 587 return not self._buf
Check whether self contains any bytes or not.
Example
buffer = Bytes()
assert buffer.is_empty()
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)
def
split_first(self) -> 'Option[tuple[int, Bytes]]':
623 def split_first(self) -> Option[tuple[int, Bytes]]: 624 """Split the first and rest elements of the bytes, If empty, `None` is returned. 625 626 Example 627 ------- 628 ```py 629 buf = Bytes.from_bytes([1, 2, 3]) 630 split = buf.split_first() 631 assert split == Some((1, [2, 3])) 632 ``` 633 """ 634 if not self._buf: 635 return _option.NOTHING 636 637 # optimized to only one element in the buffer. 638 if self.len() == 1: 639 return _option.Some((self[0], Bytes())) 640 641 first, rest = self[0], self[1:] 642 return _option.Some((first, rest))
Split the first and rest elements of the bytes, If empty, None is returned.
Example
buf = Bytes.from_bytes([1, 2, 3])
split = buf.split_first()
assert split == Some((1, [2, 3]))
def
split_last(self) -> 'Option[tuple[int, Bytes]]':
644 def split_last(self) -> Option[tuple[int, Bytes]]: 645 """Returns the last and rest of the elements of the bytes, If `self` is empty, `None` is returned. 646 647 Example 648 ------- 649 ```py 650 buf = Bytes.from_bytes([0, 1, 2]) 651 last, elements = buf.split_last().unwrap() 652 assert (last, elements) == (3, [1, 2]) 653 ``` 654 """ 655 if not self._buf: 656 return _option.NOTHING 657 658 len_ = self.len() 659 # optimized to only one element in the buffer. 660 if len_ == 1: 661 return _option.Some((self[0], Bytes())) 662 663 last, rest = self[-1], self[:-1] 664 return _option.Some((last, rest))
Returns the last and rest of the elements of the bytes, If self is empty, None is returned.
Example
buf = Bytes.from_bytes([0, 1, 2])
last, elements = buf.split_last().unwrap()
assert (last, elements) == (3, [1, 2])
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()
def
index(self, v: int, start: int = 0, stop: int = 9223372036854775807) -> int:
712 def index(self, v: int, start: int = 0, stop: int = _sys.maxsize) -> int: 713 """Return the smallest `i` such that `i` is the index of the first occurrence of `v` in the buffer. 714 715 The optional arguments start and stop can be specified to search for x within a 716 subsection of the array. Raise ValueError if x is not found 717 """ 718 if not self._buf: 719 raise ValueError from None 720 721 return self._buf.index(v, start, stop)
Return the smallest i such that i is the index of the first occurrence of v in the buffer.
The optional arguments start and stop can be specified to search for x within a subsection of the array. Raise ValueError if x is not found
def
count(self, x: int) -> int:
723 def count(self, x: int) -> int: 724 """Return the number of occurrences of `x` in the buffer. 725 726 Example 727 -------- 728 ```py 729 buf = Bytes([32, 32, 31]) 730 assert buf.count(32) == 2 731 ``` 732 """ 733 if self._buf is None: 734 return 0 735 736 return self._buf.count(x)
Return the number of occurrences of x in the buffer.
Example
buf = Bytes([32, 32, 31])
assert buf.count(32) == 2
Inherited Members
@rustc_diagnostic_item('&mut [u8]')
@typing.final
class
BytesMut878@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.
def
extend(self, src: bytes | bytearray | Iterable[int]) -> None:
927 def extend(self, src: Buffer) -> None: 928 """Extend `self` from a `src`. 929 930 Example 931 ------- 932 ```py 933 buf = Bytes() 934 buf.extend([1, 2, 3]) 935 assert buf == [1, 2, 3] 936 ``` 937 938 Parameters 939 ---------- 940 src : `Buffer` 941 Can be one of `Bytes`, `bytes`, `bytearray` or `Sequence[int]` 942 943 Raises 944 ------ 945 `OverflowError` 946 If `src` not in range of `0..255` 947 """ 948 if self._buf is None: 949 # If it was `None`, we initialize it with a source instead of extending. 950 self._buf = array.array("B", src) 951 else: 952 self._buf.extend(src)
Extend self from a src.
Example
buf = Bytes()
buf.extend([1, 2, 3])
assert buf == [1, 2, 3]
Parameters
- src (
Buffer): Can be one ofBytes,bytes,bytearrayorSequence[int]
Raises
OverflowError: Ifsrcnot in range of0..255
def
put(self, src: int) -> None:
954 def put(self, src: int) -> None: 955 """Append a byte at the end of the array. 956 957 unlike `.put_bytes`, this method appends instead of extending the array 958 which is faster if you're putting a single byte in a single call. 959 960 Example 961 ------- 962 ```py 963 buf = Bytes() 964 buf.put(32) # append a space to the end of the buffer 965 assert buf.to_bytes() == b' ' 966 ``` 967 968 Parameters 969 ---------- 970 src : `int` 971 An unsigned integer, also known as `u8` to put. 972 973 Raises 974 ------ 975 `OverflowError` 976 If `src` not in range of `0..255` 977 """ 978 if self._buf is None: 979 # If it was `None`, we initialize it with a source instead of appending. 980 self._buf = array.array("B", [src]) 981 else: 982 self._buf.append(src)
Append a byte at the end of the array.
unlike .put_bytes, this method appends instead of extending the array
which is faster if you're putting a single byte in a single call.
Example
buf = Bytes()
buf.put(32) # append a space to the end of the buffer
assert buf.to_bytes() == b' '
Parameters
- src (
int): An unsigned integer, also known asu8to put.
Raises
OverflowError: Ifsrcnot in range of0..255
def
put_float(self, src: float) -> None:
984 def put_float(self, src: float) -> None: 985 r"""Writes a single-precision (4 bytes) floating point number to `self` in big-endian byte order. 986 987 The valid range for the float value is `-3.402823466e38` to `3.402823466e38`. 988 989 Example 990 ------- 991 ```py 992 buf = BytesMut() 993 buf.put_float(1.2) 994 assert buf.to_bytes() == b'\x3f\x99\x99\x9a' 995 ``` 996 997 Raises 998 ------ 999 `OverflowError` 1000 If `src` is out of range. 1001 """ 1002 assert_precondition( 1003 -3.402823466e38 <= src <= 3.402823466e38, 1004 f"Float {src} is out of range for a single-precision float", 1005 OverflowError, 1006 ) 1007 bits = struct.pack(">f", src) 1008 1009 if self._buf is None: 1010 self._buf = array.array("B", bits) 1011 else: 1012 self._buf.extend(bits)
Writes a single-precision (4 bytes) floating point number to self in big-endian byte order.
The valid range for the float value is -3.402823466e38 to 3.402823466e38.
Example
buf = BytesMut()
buf.put_float(1.2)
assert buf.to_bytes() == b'\x3f\x99\x99\x9a'
Raises
OverflowError: Ifsrcis out of range.
def
put_char(self, char: str) -> None:
1014 def put_char(self, char: str) -> None: 1015 """Append a single character to the buffer. 1016 1017 This is the same as `self.put(ord(char))`. 1018 1019 Example 1020 ------- 1021 ```py 1022 buf = BytesMut() 1023 buf.put_char('a') 1024 assert buf == b"a" 1025 ``` 1026 1027 Parameters 1028 ---------- 1029 char : `str` 1030 The character to put. 1031 """ 1032 assert (ln := len(char)) == 1, f"Expected a single character, got {ln}" 1033 self.put(ord(char))
Append a single character to the buffer.
This is the same as self.put(ord(char)).
Example
buf = BytesMut()
buf.put_char('a')
assert buf == b"a"
Parameters
- char (
str): The character to put.
def
put_raw(self, src: _io.StringIO | _io.BytesIO | _io.BufferedReader) -> None:
1035 def put_raw(self, src: Rawish) -> None: 1036 """Extend `self` from a raw data type source. 1037 1038 Example 1039 ------- 1040 ```py 1041 buffer = BytesMut() 1042 # A file descriptor's contents 1043 with open('file.txt', 'rb') as file: 1044 buffer.put_raw(file) 1045 1046 # bytes io 1047 buffer.put(io.BytesIO(b"data")) 1048 # string io 1049 buffer.put(io.StringIO("data")) 1050 ``` 1051 1052 Parameters 1053 ---------- 1054 src : `Rawish` 1055 A valid raw data type. See `Rawish` for more details. 1056 1057 Raises 1058 ------ 1059 `OverflowError` 1060 If `src` not in range of `0..255` 1061 """ 1062 if self._buf is None: 1063 # If it was `None`, we initialize it with a source instead of extending. 1064 self._buf = array.array("B", unwrap_bytes(src)) 1065 else: 1066 self._buf.extend(unwrap_bytes(src))
Extend self from a raw data type source.
Example
buffer = BytesMut()
# A file descriptor's contents
with open('file.txt', 'rb') as file:
buffer.put_raw(file)
# bytes io
buffer.put(io.BytesIO(b"data"))
# string io
buffer.put(io.StringIO("data"))
Parameters
- src (
Rawish): A valid raw data type. SeeRawishfor more details.
Raises
OverflowError: Ifsrcnot in range of0..255
def
put_bytes(self, src: bytes | bytearray | Iterable[int]) -> None:
1068 def put_bytes(self, src: Buffer) -> None: 1069 """Put `bytes` into `self`. 1070 1071 Example 1072 ------- 1073 ```py 1074 buf = BytesMut.from_bytes(b"hello") 1075 buf.put_bytes([32, 119, 111, 114, 108, 100]) 1076 assert buf == b"hello world" 1077 ``` 1078 1079 Parameters 1080 ---------- 1081 src : `Buffer` 1082 Can be one of `Bytes`, `bytes`, `bytearray` or `Sequence[int]` 1083 1084 Raises 1085 ------ 1086 `OverflowError` 1087 If `src` not in range of `0..255` 1088 """ 1089 if self._buf is None: 1090 # If it was `None`, we initialize it with a source instead of extending. 1091 self._buf = array.array("B", src) 1092 else: 1093 self._buf.extend(src)
Put bytes into self.
Example
buf = BytesMut.from_bytes(b"hello")
buf.put_bytes([32, 119, 111, 114, 108, 100])
assert buf == b"hello world"
Parameters
- src (
Buffer): Can be one ofBytes,bytes,bytearrayorSequence[int]
Raises
OverflowError: Ifsrcnot in range of0..255
def
put_str(self, s: str) -> None:
1095 def put_str(self, s: str) -> None: 1096 """Put a `utf-8` encoded bytes from a string. 1097 1098 Example 1099 ------- 1100 ```py 1101 buffer = BytesMut() 1102 buffer.put_str("hello") 1103 1104 assert buffer == b"hello" 1105 ``` 1106 1107 Parameters 1108 ---------- 1109 src: `str` 1110 The string 1111 """ 1112 self.put_bytes(s.encode(ENCODING))
Put a utf-8 encoded bytes from a string.
Example
buffer = BytesMut()
buffer.put_str("hello")
assert buffer == b"hello"
Parameters
- src (
str): The string
def
replace(self, index: int, byte: int) -> None:
1114 def replace(self, index: int, byte: int) -> None: 1115 """Replace the byte at `index` with `byte`. 1116 1117 This method is `NOOP` if: 1118 ------------------------- 1119 * `self` is empty or unallocated. 1120 * `index` is out of range. 1121 1122 Example 1123 ------- 1124 ```py 1125 buf = BytesMut.from_bytes([1, 2, 3]) 1126 buf.replace(1, 4) 1127 assert buf == [1, 4, 3] 1128 ``` 1129 """ 1130 if not self._buf or index < 0 or index >= self.len(): 1131 return 1132 1133 self._buf[index] = byte
Replace the byte at index with byte.
This method is NOOP if:
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]
def
replace_with(self, index: int, f: Callable[[int], int]) -> None:
1135 def replace_with(self, index: int, f: collections.Callable[[int], int]) -> None: 1136 """Replace the byte at `index` with a byte returned from `f`. 1137 1138 The signature of `f` is `Fn(int) -> int`, where the argument is the old byte. 1139 1140 ## This method is `NOOP` if: 1141 * `self` is empty or unallocated. 1142 * `index` is out of range. 1143 1144 Example 1145 ------- 1146 ```py 1147 buf = BytesMut.from_bytes([1, 2, 3]) 1148 buf.replace_with(1, lambda prev: prev * 2) 1149 assert buf == [1, 4, 3] 1150 ``` 1151 """ 1152 if not self._buf or index < 0 or index >= self.len(): 1153 return 1154 1155 old = self._buf[index] 1156 self._buf[index] = f(old)
Replace the byte at index with a byte returned from f.
The signature of f is Fn(int) -> int, where the argument is the old byte.
This method is NOOP if:
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]
def
offset(self, f: Callable[[int], int]) -> None:
1158 def offset(self, f: collections.Callable[[int], int]) -> None: 1159 """Modify each byte in the buffer with a new byte returned from `f`. 1160 1161 The signature of `f` is `Fn(int) -> int`, where the argument is the previous byte. 1162 1163 Example 1164 ------- 1165 ```py 1166 buf = BytesMut.from_bytes([1, 2, 3]) 1167 buf.offset(lambda prev: prev * 2) 1168 assert buf == [2, 4, 6] 1169 ``` 1170 1171 This method is `NOOP` if `self` is empty or unallocated. 1172 """ 1173 1174 if not self._buf: 1175 return 1176 1177 for index in range(len(self._buf)): 1178 self._buf[index] = f(self._buf[index])
Modify each byte in the buffer with a new byte returned from f.
The signature of f is Fn(int) -> int, where the argument is the previous byte.
Example
buf = BytesMut.from_bytes([1, 2, 3])
buf.offset(lambda prev: prev * 2)
assert buf == [2, 4, 6]
This method is NOOP if self is empty or unallocated.
def
fill(self, value: int) -> None:
1180 def fill(self, value: int) -> None: 1181 """Fills `self` with the given byte. 1182 1183 Nothing happens if the buffer is empty or unallocated. 1184 1185 Example 1186 ------- 1187 ```py 1188 a = Bytes.from_bytes([0, 1, 2, 3]) 1189 a.fill(0) 1190 assert a == [0, 0, 0, 0] 1191 ``` 1192 """ 1193 if not self._buf: 1194 return 1195 1196 self.as_mut_ptr()[:] = bytearray([value] * self.len())
Fills self with the given byte.
Nothing happens if the buffer is empty or unallocated.
Example
a = Bytes.from_bytes([0, 1, 2, 3])
a.fill(0)
assert a == [0, 0, 0, 0]
def
fill_with(self, f: Callable[[], int]) -> None:
1198 def fill_with(self, f: collections.Callable[[], int]) -> None: 1199 """Fills `self` with the given byte returned from `f`. 1200 1201 Nothing happens if the buffer is empty or unallocated. 1202 1203 Example 1204 ------- 1205 ```py 1206 def default() -> int: 1207 return 0 1208 1209 a = Bytes.from_bytes([0, 1, 2, 3]) 1210 a.fill_with(default) 1211 assert a == [0, 0, 0, 0] 1212 ``` 1213 """ 1214 if not self._buf: 1215 return 1216 1217 self.as_mut_ptr()[:] = bytearray([f()] * self.len())
Fills self with the given byte returned from f.
Nothing happens if the buffer is empty or unallocated.
Example
def default() -> int:
return 0
a = Bytes.from_bytes([0, 1, 2, 3])
a.fill_with(default)
assert a == [0, 0, 0, 0]
def
swap(self, a: int, b: int):
1219 def swap(self, a: int, b: int): 1220 """Swap two bytes in the buffer. 1221 1222 if `a` equals to `b` then it's guaranteed that elements won't change value. 1223 1224 Example 1225 ------- 1226 ```py 1227 buf = Bytes.from_bytes([1, 2, 3, 4]) 1228 buf.swap(0, 3) 1229 assert buf == [4, 2, 3, 1] 1230 ``` 1231 1232 Raises 1233 ------ 1234 IndexError 1235 If the positions of `a` or `b` are out of index. 1236 """ 1237 if self[a] == self[b]: 1238 return 1239 1240 self[a], self[b] = self[b], self[a]
Swap two bytes in the buffer.
if a equals to b then it's guaranteed that elements won't change value.
Example
buf = Bytes.from_bytes([1, 2, 3, 4])
buf.swap(0, 3)
assert buf == [4, 2, 3, 1]
Raises
- IndexError: If the positions of
aorbare out of index.
def
swap_unchecked(self, a: int, b: int):
1242 def swap_unchecked(self, a: int, b: int): 1243 """Swap two bytes in the buffer. without checking if `a` == `b`. 1244 1245 If you care about `a` and `b` equality, see `Bytes.swap`. 1246 1247 Example 1248 ------- 1249 ```py 1250 buf = Bytes.from_bytes([1, 2, 3, 1]) 1251 buf.swap_unchecked(0, 3) 1252 assert buf == [1, 2, 3, 1] 1253 ``` 1254 1255 Raises 1256 ------ 1257 IndexError 1258 If the positions of `a` or `b` are out of index. 1259 """ 1260 self[a], self[b] = self[b], self[a]
Swap two bytes in the buffer. without checking if a == b.
If you care about a and b equality, see Bytes.swap.
Example
buf = Bytes.from_bytes([1, 2, 3, 1])
buf.swap_unchecked(0, 3)
assert buf == [1, 2, 3, 1]
Raises
- IndexError: If the positions of
aorbare out of index.
def
as_mut_ptr(self) -> 'memoryview[int]':
1262 def as_mut_ptr(self) -> memoryview[int]: 1263 """Returns a mutable pointer to the buffer data. 1264 1265 `pointer` here refers to a `memoryview` object. 1266 1267 A `BufferError` is raised if the underlying sequence is not initialized. 1268 1269 Example 1270 ------- 1271 ```py 1272 buffer = BytesMut.from_str("ouv") 1273 ptr = buffer.as_mut_ptr() 1274 ptr[0] = ord(b'L') 1275 assert buffer.to_bytes() == b"Luv" 1276 ``` 1277 """ 1278 return self.__buffer__(512)
Returns a mutable pointer to the buffer data.
pointer here refers to a memoryview object.
A BufferError is raised if the underlying sequence is not initialized.
Example
buffer = BytesMut.from_str("ouv")
ptr = buffer.as_mut_ptr()
ptr[0] = ord(b'L')
assert buffer.to_bytes() == b"Luv"
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]
def
swap_remove(self, byte: int) -> int:
1331 def swap_remove(self, byte: int) -> int: 1332 """Remove the first appearance of `item` from this buffer and return it. 1333 1334 Raises 1335 ------ 1336 * `ValueError`: if `item` is not in this buffer. 1337 * `MemoryError`: if this buffer hasn't allocated, Aka nothing has been pushed to it. 1338 1339 Example 1340 ------- 1341 ```py 1342 buf = BytesMut.from_bytes([1, 2, 3, 4]) 1343 assert 1 == buf.swap_remove(1) 1344 assert buf == [2, 3, 4] 1345 ``` 1346 """ 1347 if self._buf is None: 1348 raise MemoryError("`self` is unallocated.") from None 1349 1350 return self._buf.pop(self.index(byte))
Remove the first appearance of item from this buffer and return it.
Raises
*
ValueError(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]
def
truncate(self, size: int) -> None:
1352 def truncate(self, size: int) -> None: 1353 """Shortens the bytes, keeping the first `size` elements and dropping the rest. 1354 1355 Example 1356 ------- 1357 ```py 1358 buf = BytesMut.from_bytes([0, 0, 0, 0]) 1359 buf.truncate(1) 1360 assert buf.len() == 1 1361 ``` 1362 """ 1363 if not self._buf: 1364 return 1365 1366 del self._buf[size:]
Shortens the bytes, keeping the first size elements and dropping the rest.
Example
buf = BytesMut.from_bytes([0, 0, 0, 0])
buf.truncate(1)
assert buf.len() == 1
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)
def
split_first_mut(self) -> 'Option[tuple[int, BytesMut]]':
1401 def split_first_mut(self) -> Option[tuple[int, BytesMut]]: 1402 """Split the first and rest elements of the bytes, If empty, `None` is returned. 1403 1404 Returns a tuple of (first, rest) where rest is a mutable sequence. 1405 1406 Example 1407 ------- 1408 ```py 1409 buf = BytesMut.from_bytes([1, 2, 3]) 1410 split = buf.split_first_mut() 1411 assert split == Some((1, [2, 3])) 1412 ``` 1413 """ 1414 if not self._buf: 1415 return _option.NOTHING 1416 1417 if self.len() == 1: 1418 return _option.Some((self[0], BytesMut())) 1419 1420 first = self[0] 1421 rest = self[1:] 1422 return _option.Some((first, rest))
Split the first and rest elements of the bytes, If empty, None is returned.
Returns a tuple of (first, rest) where rest is a mutable sequence.
Example
buf = BytesMut.from_bytes([1, 2, 3])
split = buf.split_first_mut()
assert split == Some((1, [2, 3]))
def
split_last_mut(self) -> 'Option[tuple[int, Bytes]]':
1424 def split_last_mut(self) -> Option[tuple[int, Bytes]]: 1425 """Returns the last and rest of the elements of the bytes, If `self` is empty, `None` is returned. 1426 1427 Returns a tuple of (last, rest) where rest is a mutable sequence. 1428 1429 Example 1430 ------- 1431 ```py 1432 buf = BytesMut.from_bytes([0, 1, 2]) 1433 last, elements = buf.split_last_mut().unwrap() 1434 assert (last, elements) == (2, [0, 1]) 1435 ``` 1436 """ 1437 if not self._buf: 1438 return _option.NOTHING 1439 1440 len_ = self.len() 1441 if len_ == 1: 1442 return _option.Some((self[0], BytesMut())) 1443 1444 last = self[-1] 1445 rest = self[:-1] 1446 return _option.Some((last, rest))
Returns the last and rest of the elements of the bytes, If self is empty, None is returned.
Returns a tuple of (last, rest) where rest is a mutable sequence.
Example
buf = BytesMut.from_bytes([0, 1, 2])
last, elements = buf.split_last_mut().unwrap()
assert (last, elements) == (2, [0, 1])
1448 def split_at_mut(self, mid: int) -> tuple[BytesMut, BytesMut]: 1449 """Divide `self` into two at an index. 1450 1451 The first will contain all bytes from `[0:mid]` excluding `mid` itself. 1452 and the second will contain the remaining bytes. 1453 1454 if `mid` > `self.len()`, Then all bytes will be moved to the left, 1455 returning an empty bytes in right. 1456 1457 Example 1458 ------- 1459 ```py 1460 buffer = BytesMut.from_bytes((1, 2, 3, 4)) 1461 left, right = buffer.split_at_mut(0) 1462 assert left == [] and right == [1, 2, 3, 4] 1463 1464 left, right = buffer.split_at_mut(2) 1465 assert left == [1, 2] and right == [3, 4] 1466 ``` 1467 1468 The is roughly the implementation 1469 ```py 1470 self[0:mid], self[mid:] 1471 ``` 1472 """ 1473 return self[0:mid], self[mid:]
Divide self into two at an index.
The first will contain all bytes from [0:mid] excluding mid itself.
and the second will contain the remaining bytes.
if mid > self.len(), Then all bytes will be moved to the left,
returning an empty bytes in right.
Example
buffer = BytesMut.from_bytes((1, 2, 3, 4))
left, right = buffer.split_at_mut(0)
assert left == [] and right == [1, 2, 3, 4]
left, right = buffer.split_at_mut(2)
assert left == [1, 2] and right == [3, 4]
The is roughly the implementation
self[0:mid], self[mid:]
def
insert(self, index: int, value: int) -> None:
1477 def insert(self, index: int, value: int) -> None: 1478 """Insert a new item with `value` in the buffer before position `index`. 1479 1480 Negative values are treated as being relative to the end of the buffer. 1481 """ 1482 if self._buf is None: 1483 return 1484 1485 self._buf.insert(index, value)
Insert a new item with value in the buffer before position index.
Negative values are treated as being relative to the end of the buffer.
def
pop(self, i: int = -1) -> 'Option[int]':
1487 def pop(self, i: int = -1) -> Option[int]: 1488 """Removes the last element from the buffer and returns it, `Some(None)` if it is empty. 1489 1490 Example 1491 ------- 1492 ```py 1493 buf = BytesMut((21, 32, 44)) 1494 assert buf.pop() == Some(44) 1495 ``` 1496 """ 1497 if not self._buf: 1498 return _option.NOTHING 1499 1500 return _option.Some(self._buf.pop(i))
Removes the last element from the buffer and returns it, Some(None) if it is empty.
Example
buf = BytesMut((21, 32, 44))
assert buf.pop() == Some(44)
def
remove(self, i: int) -> None:
1502 def remove(self, i: int) -> None: 1503 """Remove the first appearance of `i` from `self`. 1504 1505 Example 1506 ------ 1507 ```py 1508 buf = BytesMut.from_bytes([1, 1, 2, 3, 4]) 1509 buf.remove(1) 1510 print(buf) # [1, 2, 3, 4] 1511 ``` 1512 """ 1513 if not self._buf: 1514 return 1515 1516 self._buf.remove(i)
Remove the first appearance of i from self.
Example
buf = BytesMut.from_bytes([1, 1, 2, 3, 4])
buf.remove(1)
print(buf) # [1, 2, 3, 4]
def
clear(self) -> None:
def
byteswap(self) -> None:
1535 def byteswap(self) -> None: 1536 """Swap the byte order of the bytes in `self`.""" 1537 if not self._buf: 1538 return 1539 1540 self._buf.byteswap()
Swap the byte order of the bytes in self.
Inherited Members
@rustc_diagnostic_item('HashMap')
class
HashMap398@rustc_diagnostic_item("HashMap") 399class HashMap(_RawMap[K, V], Default["HashMap[K, V]"]): 400 """An immutable key-value dictionary. 401 402 But default, `HashMap` is immutable, it cannot change its values after initializing it. however, 403 you can return a mutable reference to this hashmap via `HashMap.as_mut` method, it returns a reference to the underlying map. 404 405 Example 406 ------- 407 ```py 408 # initialize it with a source. after that, item insertion is not allowed. 409 books: HashMap[str, str] = HashMap({ 410 "Adventures of Huckleberry Finn": "My favorite book." 411 }) 412 # get a mutable reference to it to be able to. 413 books_mut = books.as_mut() 414 # You can either call `.insert`, which is similar to Rust's. 415 books_mut.insert( 416 "Grimms Fairy Tales", 417 "Masterpiece.", 418 ) 419 # Or via item assignments 420 books_mut["Pride and Prejudice"] = "Very enjoyable." 421 print(books) 422 423 for book, review in books.items(): 424 print(book, review) 425 ``` 426 427 Parameters 428 ---------- 429 source: `dict[K, V]` 430 A dictionary to point to. this will be used as the underlying source. 431 """ 432 433 __slots__ = ("_source",) 434 435 def __init__(self, source: dict[K, V] | None = None, /) -> None: 436 super().__init__(source) 437 438 @staticmethod 439 def default() -> HashMap[K, V]: 440 """Creates an empty `HashMap<K, V>`.""" 441 return HashMap() 442 443 @classmethod 444 def from_mut(cls, source: dict[K, V] | None = None, /) -> RefMut[K, V]: 445 """Create a new mutable `HashMap`, with the given source if available. 446 447 Example 448 ------- 449 ```py 450 books = HashMap.from_mut() 451 books[0] = "Twilight" 452 ``` 453 """ 454 return RefMut(source or {}) 455 456 @staticmethod 457 def from_keys_mut(*iterable: tuple[K, V]) -> RefMut[K, V]: 458 """Create a new mutable `HashMap` from a sequence of key-value pairs. 459 460 Example 461 ------- 462 ```py 463 tier_list = [ 464 ("S", "Top tier"), 465 ("A", "Very Good"), 466 ("B", "Good"), 467 ("C", "Average"), 468 ("D", "dodo") 469 ] 470 mapping: RefMut[str, str] = HashMap.from_keys_mut(*tier_list) 471 ``` 472 """ 473 return RefMut({k: v for k, v in iterable}) 474 475 def as_mut(self) -> RefMut[K, V]: 476 """Get a mutable reference to this hash map. 477 478 Example 479 ------- 480 ```py 481 map: HashMap[str, float] = HashMap() 482 483 # Get a reference to map 484 mut = map.as_mut() 485 mut.insert("sqrt", 1.0) 486 del mut # not needed anymore 487 488 assert map == {"sqrt": 1.0} 489 ``` 490 """ 491 return RefMut(self._source)
An immutable key-value dictionary.
But default, HashMap is immutable, it cannot change its values after initializing it. however,
you can return a mutable reference to this hashmap via HashMap.as_mut method, it returns a reference to the underlying map.
Example
# initialize it with a source. after that, item insertion is not allowed.
books: HashMap[str, str] = HashMap({
"Adventures of Huckleberry Finn": "My favorite book."
})
# get a mutable reference to it to be able to.
books_mut = books.as_mut()
# You can either call `.insert`, which is similar to Rust's.
books_mut.insert(
"Grimms Fairy Tales",
"Masterpiece.",
)
# Or via item assignments
books_mut["Pride and Prejudice"] = "Very enjoyable."
print(books)
for book, review in books.items():
print(book, review)
Parameters
- source (
dict[K, V]): A dictionary to point to. this will be used as the underlying source. - # Implementations
- **This class implements HashMap:
438 @staticmethod 439 def default() -> HashMap[K, V]: 440 """Creates an empty `HashMap<K, V>`.""" 441 return HashMap()
Creates an empty HashMap<K, V>.
@classmethod
def
from_mut( cls, source: dict[~K, ~V] | None = None, /) -> sain.collections.hash_map.RefMut[~K, ~V]:
443 @classmethod 444 def from_mut(cls, source: dict[K, V] | None = None, /) -> RefMut[K, V]: 445 """Create a new mutable `HashMap`, with the given source if available. 446 447 Example 448 ------- 449 ```py 450 books = HashMap.from_mut() 451 books[0] = "Twilight" 452 ``` 453 """ 454 return RefMut(source or {})
Create a new mutable HashMap, with the given source if available.
Example
books = HashMap.from_mut()
books[0] = "Twilight"
@staticmethod
def
from_keys_mut(*iterable: tuple[~K, ~V]) -> sain.collections.hash_map.RefMut[~K, ~V]:
456 @staticmethod 457 def from_keys_mut(*iterable: tuple[K, V]) -> RefMut[K, V]: 458 """Create a new mutable `HashMap` from a sequence of key-value pairs. 459 460 Example 461 ------- 462 ```py 463 tier_list = [ 464 ("S", "Top tier"), 465 ("A", "Very Good"), 466 ("B", "Good"), 467 ("C", "Average"), 468 ("D", "dodo") 469 ] 470 mapping: RefMut[str, str] = HashMap.from_keys_mut(*tier_list) 471 ``` 472 """ 473 return RefMut({k: v for k, v in iterable})
Create a new mutable HashMap from a sequence of key-value pairs.
Example
tier_list = [
("S", "Top tier"),
("A", "Very Good"),
("B", "Good"),
("C", "Average"),
("D", "dodo")
]
mapping: RefMut[str, str] = HashMap.from_keys_mut(*tier_list)
475 def as_mut(self) -> RefMut[K, V]: 476 """Get a mutable reference to this hash map. 477 478 Example 479 ------- 480 ```py 481 map: HashMap[str, float] = HashMap() 482 483 # Get a reference to map 484 mut = map.as_mut() 485 mut.insert("sqrt", 1.0) 486 del mut # not needed anymore 487 488 assert map == {"sqrt": 1.0} 489 ``` 490 """ 491 return RefMut(self._source)
Get a mutable reference to this hash map.
Example
map: HashMap[str, float] = HashMap()
# Get a reference to map
mut = map.as_mut()
mut.insert("sqrt", 1.0)
del mut # not needed anymore
assert map == {"sqrt": 1.0}