Mercurial > public > mercurial-scm > hg-stable

comparison mercurial/utils/cborutil.py @ 52709:279e217d6041

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
typing: lock in the new type annotations detected with the pyupgrade changes After the changes culminating in 70a75d379daf, pytype was able to detect these types better (typically changing from something like `Generator[Any, Any, None]` to `Generator[bytes, Any, None]`). Let's make these explicit so they don't disappear because of other changes.
author Matt Harbison <matt_harbison@yahoo.com>
date Mon, 06 Jan 2025 20:02:17 -0500
parents e627cc25b6f3
children 5e09c6b5b795
comparison
equal deleted inserted replaced
52708:efac197c6cff 52709:279e217d6041
6 # GNU General Public License version 2 or any later version. 6 # GNU General Public License version 2 or any later version.
7 7
8 from __future__ import annotations 8 from __future__ import annotations
9 9
10 import struct 10 import struct
11 11 import typing
12
13 if typing.TYPE_CHECKING:
14 from typing import (
15 Iterator,
16 )
12 17
13 # Very short very of RFC 7049... 18 # Very short very of RFC 7049...
14 # 19 #
15 # Each item begins with a byte. The 3 high bits of that byte denote the 20 # Each item begins with a byte. The 3 high bits of that byte denote the
16 # "major type." The lower 5 bits denote the "subtype." Each major type 21 # "major type." The lower 5 bits denote the "subtype." Each major type
129 yield encodelength(MAJOR_TYPE_UINT, v) 134 yield encodelength(MAJOR_TYPE_UINT, v)
130 else: 135 else:
131 yield encodelength(MAJOR_TYPE_NEGINT, abs(v) - 1) 136 yield encodelength(MAJOR_TYPE_NEGINT, abs(v) - 1)
132 137
133 138
134 def streamencodearray(l): 139 def streamencodearray(l) -> Iterator[bytes]:
135 """Encode a known size iterable to an array.""" 140 """Encode a known size iterable to an array."""
136 141
137 yield encodelength(MAJOR_TYPE_ARRAY, len(l)) 142 yield encodelength(MAJOR_TYPE_ARRAY, len(l))
138 143
139 for i in l: 144 for i in l:
140 yield from streamencode(i) 145 yield from streamencode(i)
141 146
142 147
143 def streamencodearrayfromiter(it): 148 def streamencodearrayfromiter(it) -> Iterator[bytes]:
144 """Encode an iterator of items to an indefinite length array.""" 149 """Encode an iterator of items to an indefinite length array."""
145 150
146 yield BEGIN_INDEFINITE_ARRAY 151 yield BEGIN_INDEFINITE_ARRAY
147 152
148 for i in it: 153 for i in it:
161 yield encodelength(MAJOR_TYPE_SEMANTIC, SEMANTIC_TAG_FINITE_SET) 166 yield encodelength(MAJOR_TYPE_SEMANTIC, SEMANTIC_TAG_FINITE_SET)
162 167
163 yield from streamencodearray(sorted(s, key=_mixedtypesortkey)) 168 yield from streamencodearray(sorted(s, key=_mixedtypesortkey))
164 169
165 170
166 def streamencodemap(d): 171 def streamencodemap(d) -> Iterator[bytes]:
167 """Encode dictionary to a generator. 172 """Encode dictionary to a generator.
168 173
169 Does not supporting indefinite length dictionaries. 174 Does not supporting indefinite length dictionaries.
170 """ 175 """
171 yield encodelength(MAJOR_TYPE_MAP, len(d)) 176 yield encodelength(MAJOR_TYPE_MAP, len(d))
173 for key, value in sorted(d.items(), key=lambda x: _mixedtypesortkey(x[0])): 178 for key, value in sorted(d.items(), key=lambda x: _mixedtypesortkey(x[0])):
174 yield from streamencode(key) 179 yield from streamencode(key)
175 yield from streamencode(value) 180 yield from streamencode(value)
176 181
177 182
178 def streamencodemapfromiter(it): 183 def streamencodemapfromiter(it) -> Iterator[bytes]:
179 """Given an iterable of (key, value), encode to an indefinite length map.""" 184 """Given an iterable of (key, value), encode to an indefinite length map."""
180 yield BEGIN_INDEFINITE_MAP 185 yield BEGIN_INDEFINITE_MAP
181 186
182 for key, value in it: 187 for key, value in it:
183 yield from streamencode(key) 188 yield from streamencode(key)