Implement _tokenize and update tokenize from v3.14.3 (#7392)

* Base implementation of _tokenize module

Port from PR #6240 by ShaharNaveh, adapted to current codebase.
Uses ruff_python_parser for tokenization via TokenizerIter.

* Update tokenize from v3.14.3

* Rewrite _tokenize with 2-phase model

Replace per-line reparsing with single-pass tokenization:
- Read all lines via readline, parse once, yield tokens
- Fix token type values (COMMENT=65, NL=66, OP=55)
- Fix NEWLINE/NL end positions and implicit newline handling
- Fix DEDENT positions via look-ahead to next non-DEDENT token
- Handle FSTRING_MIDDLE brace unescaping ({{ → {, }} → })
- Emit implicit NL before ENDMARKER when source lacks trailing newline
- Raise IndentationError from lexer errors
- Remove 13 expectedFailure marks for now-passing tests

---------

Co-authored-by: ShaharNaveh <shaharnaveh@users.noreply.github.com>
Co-authored-by: CPython Developers <>
This commit is contained in:
Jeong, YunWonShaharNavehCPython Developers
authored and GitHub committed 2026-03-10 12:27:09 +09:00
commit e645761142
7 files changed
+2873 -379

No files matched your search

Generated
+4
View File
@@ -3283,6 +3283,10 @@ dependencies = [
"pkcs8",
"pymath",
"rand_core 0.9.5",
"ruff_python_ast",
"ruff_python_parser",
"ruff_source_file",
"ruff_text_size",
"rustix",
"rustls",
"rustls-native-certs",
-1
View File
@@ -316,7 +316,6 @@ class TestCommandLine(TestCase):
self.assertListEqual(out.splitlines(), stdout.splitlines())
self.assertListEqual(err.splitlines(), stderr.splitlines())
@unittest.expectedFailure # TODO: RUSTPYTHON; Should displays error when errored python file is given.
def test_with_errored_file(self):
"""Should displays error when errored python file is given."""
with TemporaryPyFile(SOURCE_CODES["wrong_indented"]) as file_path:
+1982 -157
View File
File diff suppressed because it is too large. Load diff
+132 -221
View File
@@ -24,10 +24,7 @@ __author__ = 'Ka-Ping Yee <ping@lfw.org>'
__credits__ = ('GvR, ESR, Tim Peters, Thomas Wouters, Fred Drake, '
'Skip Montanaro, Raymond Hettinger, Trent Nelson, '
'Michael Foord')
try:
from builtins import open as _builtin_open
except ImportError:
pass
from builtins import open as _builtin_open
from codecs import lookup, BOM_UTF8
import collections
import functools
@@ -37,13 +34,14 @@ import re
import sys
from token import *
from token import EXACT_TOKEN_TYPES
import _tokenize
cookie_re = re.compile(r'^[ \t\f]*#.*?coding[:=][ \t]*([-\w.]+)', re.ASCII)
cookie_re = re.compile(br'^[ \t\f]*#.*?coding[:=][ \t]*([-\w.]+)', re.ASCII)
blank_re = re.compile(br'^[ \t\f]*(?:[#\r\n]|$)', re.ASCII)
import token
__all__ = token.__all__ + ["tokenize", "generate_tokens", "detect_encoding",
"untokenize", "TokenInfo"]
"untokenize", "TokenInfo", "open", "TokenError"]
del token
class TokenInfo(collections.namedtuple('TokenInfo', 'type string start end line')):
@@ -88,7 +86,7 @@ def _all_string_prefixes():
# The valid string prefixes. Only contain the lower case versions,
# and don't contain any permutations (include 'fr', but not
# 'rf'). The various permutations will be generated.
_valid_string_prefixes = ['b', 'r', 'u', 'f', 'br', 'fr']
_valid_string_prefixes = ['b', 'r', 'u', 'f', 't', 'br', 'fr', 'tr']
# if we add binary f-strings, add: ['fb', 'fbr']
result = {''}
for prefix in _valid_string_prefixes:
@@ -134,7 +132,7 @@ ContStr = group(StringPrefix + r"'[^\n'\\]*(?:\\.[^\n'\\]*)*" +
group("'", r'\\\r?\n'),
StringPrefix + r'"[^\n"\\]*(?:\\.[^\n"\\]*)*' +
group('"', r'\\\r?\n'))
PseudoExtras = group(r'\\\r?\n|\Z', Comment, Triple)
PseudoExtras = group(r'\\\r?\n|\z', Comment, Triple)
PseudoToken = Whitespace + group(PseudoExtras, Number, Funny, ContStr, Name)
# For a given string prefix plus quotes, endpats maps it to a regex
@@ -146,6 +144,7 @@ for _prefix in _all_string_prefixes():
endpats[_prefix + '"'] = Double
endpats[_prefix + "'''"] = Single3
endpats[_prefix + '"""'] = Double3
del _prefix
# A set of all of the single and triple quoted string prefixes,
# including the opening quotes.
@@ -156,13 +155,12 @@ for t in _all_string_prefixes():
single_quoted.add(u)
for u in (t + '"""', t + "'''"):
triple_quoted.add(u)
del t, u
tabsize = 8
class TokenError(Exception): pass
class StopTokenizing(Exception): pass
class Untokenizer:
@@ -170,6 +168,8 @@ class Untokenizer:
self.tokens = []
self.prev_row = 1
self.prev_col = 0
self.prev_type = None
self.prev_line = ""
self.encoding = None
def add_whitespace(self, start):
@@ -177,14 +177,51 @@ class Untokenizer:
if row < self.prev_row or row == self.prev_row and col < self.prev_col:
raise ValueError("start ({},{}) precedes previous end ({},{})"
.format(row, col, self.prev_row, self.prev_col))
row_offset = row - self.prev_row
if row_offset:
self.tokens.append("\\\n" * row_offset)
self.prev_col = 0
self.add_backslash_continuation(start)
col_offset = col - self.prev_col
if col_offset:
self.tokens.append(" " * col_offset)
def add_backslash_continuation(self, start):
"""Add backslash continuation characters if the row has increased
without encountering a newline token.
This also inserts the correct amount of whitespace before the backslash.
"""
row = start[0]
row_offset = row - self.prev_row
if row_offset == 0:
return
newline = '\r\n' if self.prev_line.endswith('\r\n') else '\n'
line = self.prev_line.rstrip('\\\r\n')
ws = ''.join(_itertools.takewhile(str.isspace, reversed(line)))
self.tokens.append(ws + f"\\{newline}" * row_offset)
self.prev_col = 0
def escape_brackets(self, token):
characters = []
consume_until_next_bracket = False
for character in token:
if character == "}":
if consume_until_next_bracket:
consume_until_next_bracket = False
else:
characters.append(character)
if character == "{":
n_backslashes = sum(
1 for char in _itertools.takewhile(
"\\".__eq__,
characters[-2::-1]
)
)
if n_backslashes % 2 == 0 or characters[-1] != "N":
characters.append(character)
else:
consume_until_next_bracket = True
characters.append(character)
return "".join(characters)
def untokenize(self, iterable):
it = iter(iterable)
indents = []
@@ -214,12 +251,22 @@ class Untokenizer:
self.tokens.append(indent)
self.prev_col = len(indent)
startline = False
elif tok_type in {FSTRING_MIDDLE, TSTRING_MIDDLE}:
if '{' in token or '}' in token:
token = self.escape_brackets(token)
last_line = token.splitlines()[-1]
end_line, end_col = end
extra_chars = last_line.count("{{") + last_line.count("}}")
end = (end_line, end_col + extra_chars)
self.add_whitespace(start)
self.tokens.append(token)
self.prev_row, self.prev_col = end
if tok_type in (NEWLINE, NL):
self.prev_row += 1
self.prev_col = 0
self.prev_type = tok_type
self.prev_line = line
return "".join(self.tokens)
def compat(self, token, iterable):
@@ -227,6 +274,7 @@ class Untokenizer:
toks_append = self.tokens.append
startline = token[0] in (NEWLINE, NL)
prevstring = False
in_fstring_or_tstring = 0
for tok in _itertools.chain([token], iterable):
toknum, tokval = tok[:2]
@@ -245,6 +293,10 @@ class Untokenizer:
else:
prevstring = False
if toknum in {FSTRING_START, TSTRING_START}:
in_fstring_or_tstring += 1
elif toknum in {FSTRING_END, TSTRING_END}:
in_fstring_or_tstring -= 1
if toknum == INDENT:
indents.append(tokval)
continue
@@ -256,7 +308,19 @@ class Untokenizer:
elif startline and indents:
toks_append(indents[-1])
startline = False
elif toknum in {FSTRING_MIDDLE, TSTRING_MIDDLE}:
tokval = self.escape_brackets(tokval)
# Insert a space between two consecutive brackets if we are in an f-string or t-string
if tokval in {"{", "}"} and self.tokens and self.tokens[-1] == tokval and in_fstring_or_tstring:
tokval = ' ' + tokval
# Insert a space between two consecutive f-strings
if toknum in (STRING, FSTRING_START) and self.prev_type in (STRING, FSTRING_END):
self.tokens.append(" ")
toks_append(tokval)
self.prev_type = toknum
def untokenize(iterable):
@@ -268,16 +332,10 @@ def untokenize(iterable):
with at least two elements, a token number and token value. If
only two tokens are passed, the resulting output is poor.
Round-trip invariant for full input:
Untokenized source will match input source exactly
Round-trip invariant for limited input:
# Output bytes will tokenize back to the input
t1 = [tok[:2] for tok in tokenize(f.readline)]
newcode = untokenize(t1)
readline = BytesIO(newcode).readline
t2 = [tok[:2] for tok in tokenize(readline)]
assert t1 == t2
The result is guaranteed to tokenize back to match the input so
that the conversion is lossless and round-trips are assured.
The guarantee applies only to the token type and token string as
the spacing between tokens (column positions) may change.
"""
ut = Untokenizer()
out = ut.untokenize(iterable)
@@ -287,7 +345,7 @@ def untokenize(iterable):
def _get_normal_name(orig_enc):
"""Imitates get_normal_name in tokenizer.c."""
"""Imitates get_normal_name in Parser/tokenizer/helpers.c."""
# Only care about the first 12 characters.
enc = orig_enc[:12].lower().replace("_", "-")
if enc == "utf-8" or enc.startswith("utf-8-"):
@@ -327,22 +385,23 @@ def detect_encoding(readline):
except StopIteration:
return b''
def find_cookie(line):
def check(line, encoding):
# Check if the line matches the encoding.
if 0 in line:
raise SyntaxError("source code cannot contain null bytes")
try:
# Decode as UTF-8. Either the line is an encoding declaration,
# in which case it should be pure ASCII, or it must be UTF-8
# per default encoding.
line_string = line.decode('utf-8')
line.decode(encoding)
except UnicodeDecodeError:
msg = "invalid or missing encoding declaration"
if filename is not None:
msg = '{} for {!r}'.format(msg, filename)
raise SyntaxError(msg)
match = cookie_re.match(line_string)
def find_cookie(line):
match = cookie_re.match(line)
if not match:
return None
encoding = _get_normal_name(match.group(1))
encoding = _get_normal_name(match.group(1).decode())
try:
codec = lookup(encoding)
except LookupError:
@@ -375,18 +434,23 @@ def detect_encoding(readline):
encoding = find_cookie(first)
if encoding:
check(first, encoding)
return encoding, [first]
if not blank_re.match(first):
check(first, default)
return default, [first]
second = read_or_stop()
if not second:
check(first, default)
return default, [first]
encoding = find_cookie(second)
if encoding:
check(first + second, encoding)
return encoding, [first, second]
check(first + second, default)
return default, [first, second]
@@ -405,7 +469,6 @@ def open(filename):
buffer.close()
raise
def tokenize(readline):
"""
The tokenize() generator requires one argument, readline, which
@@ -426,193 +489,13 @@ def tokenize(readline):
which tells you which encoding was used to decode the bytes stream.
"""
encoding, consumed = detect_encoding(readline)
empty = _itertools.repeat(b"")
rl_gen = _itertools.chain(consumed, iter(readline, b""), empty)
return _tokenize(rl_gen.__next__, encoding)
def _tokenize(readline, encoding):
lnum = parenlev = continued = 0
numchars = '0123456789'
contstr, needcont = '', 0
contline = None
indents = [0]
rl_gen = _itertools.chain(consumed, iter(readline, b""))
if encoding is not None:
if encoding == "utf-8-sig":
# BOM will already have been stripped.
encoding = "utf-8"
yield TokenInfo(ENCODING, encoding, (0, 0), (0, 0), '')
last_line = b''
line = b''
while True: # loop over lines in stream
try:
# We capture the value of the line variable here because
# readline uses the empty string '' to signal end of input,
# hence `line` itself will always be overwritten at the end
# of this loop.
last_line = line
line = readline()
except StopIteration:
line = b''
if encoding is not None:
line = line.decode(encoding)
lnum += 1
pos, max = 0, len(line)
if contstr: # continued string
if not line:
raise TokenError("EOF in multi-line string", strstart)
endmatch = endprog.match(line)
if endmatch:
pos = end = endmatch.end(0)
yield TokenInfo(STRING, contstr + line[:end],
strstart, (lnum, end), contline + line)
contstr, needcont = '', 0
contline = None
elif needcont and line[-2:] != '\\\n' and line[-3:] != '\\\r\n':
yield TokenInfo(ERRORTOKEN, contstr + line,
strstart, (lnum, len(line)), contline)
contstr = ''
contline = None
continue
else:
contstr = contstr + line
contline = contline + line
continue
elif parenlev == 0 and not continued: # new statement
if not line: break
column = 0
while pos < max: # measure leading whitespace
if line[pos] == ' ':
column += 1
elif line[pos] == '\t':
column = (column//tabsize + 1)*tabsize
elif line[pos] == '\f':
column = 0
else:
break
pos += 1
if pos == max:
break
if line[pos] in '#\r\n': # skip comments or blank lines
if line[pos] == '#':
comment_token = line[pos:].rstrip('\r\n')
yield TokenInfo(COMMENT, comment_token,
(lnum, pos), (lnum, pos + len(comment_token)), line)
pos += len(comment_token)
yield TokenInfo(NL, line[pos:],
(lnum, pos), (lnum, len(line)), line)
continue
if column > indents[-1]: # count indents or dedents
indents.append(column)
yield TokenInfo(INDENT, line[:pos], (lnum, 0), (lnum, pos), line)
while column < indents[-1]:
if column not in indents:
raise IndentationError(
"unindent does not match any outer indentation level",
("<tokenize>", lnum, pos, line))
indents = indents[:-1]
yield TokenInfo(DEDENT, '', (lnum, pos), (lnum, pos), line)
else: # continued statement
if not line:
raise TokenError("EOF in multi-line statement", (lnum, 0))
continued = 0
while pos < max:
pseudomatch = _compile(PseudoToken).match(line, pos)
if pseudomatch: # scan for tokens
start, end = pseudomatch.span(1)
spos, epos, pos = (lnum, start), (lnum, end), end
if start == end:
continue
token, initial = line[start:end], line[start]
if (initial in numchars or # ordinary number
(initial == '.' and token != '.' and token != '...')):
yield TokenInfo(NUMBER, token, spos, epos, line)
elif initial in '\r\n':
if parenlev > 0:
yield TokenInfo(NL, token, spos, epos, line)
else:
yield TokenInfo(NEWLINE, token, spos, epos, line)
elif initial == '#':
assert not token.endswith("\n")
yield TokenInfo(COMMENT, token, spos, epos, line)
elif token in triple_quoted:
endprog = _compile(endpats[token])
endmatch = endprog.match(line, pos)
if endmatch: # all on one line
pos = endmatch.end(0)
token = line[start:pos]
yield TokenInfo(STRING, token, spos, (lnum, pos), line)
else:
strstart = (lnum, start) # multiple lines
contstr = line[start:]
contline = line
break
# Check up to the first 3 chars of the token to see if
# they're in the single_quoted set. If so, they start
# a string.
# We're using the first 3, because we're looking for
# "rb'" (for example) at the start of the token. If
# we switch to longer prefixes, this needs to be
# adjusted.
# Note that initial == token[:1].
# Also note that single quote checking must come after
# triple quote checking (above).
elif (initial in single_quoted or
token[:2] in single_quoted or
token[:3] in single_quoted):
if token[-1] == '\n': # continued string
strstart = (lnum, start)
# Again, using the first 3 chars of the
# token. This is looking for the matching end
# regex for the correct type of quote
# character. So it's really looking for
# endpats["'"] or endpats['"'], by trying to
# skip string prefix characters, if any.
endprog = _compile(endpats.get(initial) or
endpats.get(token[1]) or
endpats.get(token[2]))
contstr, needcont = line[start:], 1
contline = line
break
else: # ordinary string
yield TokenInfo(STRING, token, spos, epos, line)
elif initial.isidentifier(): # ordinary name
yield TokenInfo(NAME, token, spos, epos, line)
elif initial == '\\': # continued stmt
continued = 1
else:
if initial in '([{':
parenlev += 1
elif initial in ')]}':
parenlev -= 1
yield TokenInfo(OP, token, spos, epos, line)
else:
yield TokenInfo(ERRORTOKEN, line[pos],
(lnum, pos), (lnum, pos+1), line)
pos += 1
# Add an implicit NEWLINE if the input doesn't end in one
if last_line and last_line[-1] not in '\r\n' and not last_line.strip().startswith("#"):
yield TokenInfo(NEWLINE, '', (lnum - 1, len(last_line)), (lnum - 1, len(last_line) + 1), '')
for indent in indents[1:]: # pop remaining indent levels
yield TokenInfo(DEDENT, '', (lnum, 0), (lnum, 0), '')
yield TokenInfo(ENDMARKER, '', (lnum, 0), (lnum, 0), '')
yield from _generate_tokens_from_c_tokenizer(rl_gen.__next__, encoding, extra_tokens=True)
def generate_tokens(readline):
"""Tokenize a source reading Python code as unicode strings.
@@ -620,9 +503,9 @@ def generate_tokens(readline):
This has the same API as tokenize(), except that it expects the *readline*
callable to return str objects instead of bytes.
"""
return _tokenize(readline, None)
return _generate_tokens_from_c_tokenizer(readline, extra_tokens=True)
def main():
def _main(args=None):
import argparse
# Helper error handling routines
@@ -641,13 +524,13 @@ def main():
sys.exit(1)
# Parse the arguments and options
parser = argparse.ArgumentParser(prog='python -m tokenize')
parser = argparse.ArgumentParser(color=True)
parser.add_argument(dest='filename', nargs='?',
metavar='filename.py',
help='the file to tokenize; defaults to stdin')
parser.add_argument('-e', '--exact', dest='exact', action='store_true',
help='display token names using the exact type')
args = parser.parse_args()
args = parser.parse_args(args)
try:
# Tokenize the input
@@ -657,7 +540,9 @@ def main():
tokens = list(tokenize(f.readline))
else:
filename = "<stdin>"
tokens = _tokenize(sys.stdin.readline, None)
tokens = _generate_tokens_from_c_tokenizer(
sys.stdin.readline, extra_tokens=True)
# Output the tokenization
for token in tokens:
@@ -683,5 +568,31 @@ def main():
perror("unexpected error: %s" % err)
raise
def _transform_msg(msg):
"""Transform error messages from the C tokenizer into the Python tokenize
The C tokenizer is more picky than the Python one, so we need to massage
the error messages a bit for backwards compatibility.
"""
if "unterminated triple-quoted string literal" in msg:
return "EOF in multi-line string"
return msg
def _generate_tokens_from_c_tokenizer(source, encoding=None, extra_tokens=False):
"""Tokenize a source reading Python code as unicode strings using the internal C tokenizer"""
if encoding is None:
it = _tokenize.TokenizerIter(source, extra_tokens=extra_tokens)
else:
it = _tokenize.TokenizerIter(source, encoding=encoding, extra_tokens=extra_tokens)
try:
for info in it:
yield TokenInfo._make(info)
except SyntaxError as e:
if type(e) != SyntaxError:
raise e from None
msg = _transform_msg(e.msg)
raise TokenError(msg, (e.lineno, e.offset)) from None
if __name__ == "__main__":
main()
_main()
+5
View File
@@ -31,6 +31,11 @@ rustpython-derive = { workspace = true }
rustpython-vm = { workspace = true, default-features = false, features = ["compiler"]}
rustpython-common = { workspace = true }
ruff_python_parser = { workspace = true }
ruff_python_ast = { workspace = true }
ruff_text_size = { workspace = true }
ruff_source_file = { workspace = true }
ahash = { workspace = true }
ascii = { workspace = true }
cfg-if = { workspace = true }
+747
View File
@@ -0,0 +1,747 @@
pub(crate) use _tokenize::module_def;
#[pymodule]
mod _tokenize {
use crate::{
common::lock::PyRwLock,
vm::{
AsObject, Py, PyObjectRef, PyPayload, PyResult, VirtualMachine,
builtins::{PyBytes, PyStr, PyType},
convert::ToPyObject,
function::ArgCallable,
protocol::PyIterReturn,
types::{Constructor, IterNext, Iterable, SelfIter},
},
};
use ruff_python_ast::PySourceType;
use ruff_python_ast::token::{Token, TokenKind};
use ruff_python_parser::{
LexicalErrorType, ParseError, ParseErrorType, parse_unchecked_source,
};
use ruff_source_file::{LineIndex, LineRanges};
use ruff_text_size::{Ranged, TextSize};
use core::fmt;
const TOKEN_ENDMARKER: u8 = 0;
const TOKEN_DEDENT: u8 = 6;
const TOKEN_OP: u8 = 55;
const TOKEN_COMMENT: u8 = 65;
const TOKEN_NL: u8 = 66;
#[pyattr]
#[pyclass(name = "TokenizerIter")]
#[derive(PyPayload)]
pub struct PyTokenizerIter {
readline: ArgCallable,
extra_tokens: bool,
encoding: Option<String>,
state: PyRwLock<TokenizerState>,
}
impl PyTokenizerIter {
fn readline(&self, vm: &VirtualMachine) -> PyResult<String> {
let raw_line = match self.readline.invoke((), vm) {
Ok(v) => v,
Err(err) => {
if err.fast_isinstance(vm.ctx.exceptions.stop_iteration) {
return Ok(String::new());
}
return Err(err);
}
};
Ok(match &self.encoding {
Some(encoding) => {
let bytes = raw_line
.downcast::<PyBytes>()
.map_err(|_| vm.new_type_error("readline() returned a non-bytes object"))?;
vm.state
.codec_registry
.decode_text(bytes.into(), encoding, None, vm)
.map(|s| s.to_string())?
}
None => raw_line
.downcast::<PyStr>()
.map(|s| s.to_string())
.map_err(|_| vm.new_type_error("readline() returned a non-string object"))?,
})
}
}
impl fmt::Debug for PyTokenizerIter {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("PyTokenizerIter")
.field("extra_tokens", &self.extra_tokens)
.field("encoding", &self.encoding)
.finish()
}
}
#[pyclass(with(Constructor, Iterable, IterNext))]
impl PyTokenizerIter {}
impl Constructor for PyTokenizerIter {
type Args = PyTokenizerIterArgs;
fn py_new(_cls: &Py<PyType>, args: Self::Args, _vm: &VirtualMachine) -> PyResult<Self> {
let Self::Args {
readline,
extra_tokens,
encoding,
} = args;
Ok(Self {
readline,
extra_tokens,
encoding: encoding.map(|s| s.to_string()),
state: PyRwLock::new(TokenizerState {
phase: TokenizerPhase::Reading {
source: String::new(),
},
}),
})
}
}
impl SelfIter for PyTokenizerIter {}
impl IterNext for PyTokenizerIter {
fn next(zelf: &Py<Self>, vm: &VirtualMachine) -> PyResult<PyIterReturn> {
let mut state = zelf.state.read().clone();
loop {
match &mut state.phase {
TokenizerPhase::Reading { source } => {
let line = zelf.readline(vm)?;
if line.is_empty() {
let accumulated = core::mem::take(source);
let parsed =
parse_unchecked_source(&accumulated, PySourceType::Python);
let tokens: Vec<Token> = parsed.tokens().iter().copied().collect();
let errors: Vec<ParseError> = parsed.errors().to_vec();
let line_index = LineIndex::from_source_text(&accumulated);
let implicit_nl = !accumulated.ends_with('\n');
state.phase = TokenizerPhase::Yielding {
source: accumulated,
tokens,
errors,
index: 0,
line_index,
need_implicit_nl: implicit_nl,
pending_fstring_parts: Vec::new(),
pending_empty_fstring_middle: None,
};
} else {
source.push_str(&line);
}
}
TokenizerPhase::Yielding { .. } => {
let result =
emit_next_token(&mut state, zelf.extra_tokens, vm)?;
*zelf.state.write() = state;
return Ok(result);
}
TokenizerPhase::Done => {
return Ok(PyIterReturn::StopIteration(None));
}
}
}
}
}
/// Emit the next token from the Yielding phase.
fn emit_next_token(
state: &mut TokenizerState,
extra_tokens: bool,
vm: &VirtualMachine,
) -> PyResult<PyIterReturn> {
let TokenizerPhase::Yielding {
source,
tokens,
errors,
index,
line_index,
need_implicit_nl,
pending_fstring_parts,
pending_empty_fstring_middle,
} = &mut state.phase
else {
unreachable!()
};
// Emit pending empty FSTRING_MIDDLE (for format spec nesting)
if let Some((mid_type, mid_line, mid_col, mid_line_str)) =
pending_empty_fstring_middle.take()
{
return Ok(PyIterReturn::Return(make_token_tuple(
vm,
mid_type,
"",
mid_line,
mid_col as isize,
mid_line,
mid_col as isize,
&mid_line_str,
)));
}
// Emit any pending fstring sub-tokens first
if let Some((tok_type, tok_str, sl, sc, el, ec)) = pending_fstring_parts.pop() {
let offset: usize = source
.lines()
.take(sl.saturating_sub(1))
.map(|l| l.len() + 1)
.sum();
let full_line =
source.full_line_str(TextSize::from(offset.min(source.len()) as u32));
return Ok(PyIterReturn::Return(make_token_tuple(
vm, tok_type, &tok_str, sl, sc as isize, el, ec as isize, full_line,
)));
}
let source_len = TextSize::from(source.len() as u32);
while *index < tokens.len() {
let token = tokens[*index];
*index += 1;
let kind = token.kind();
let range = token.range();
// Check for lexical indentation errors.
// Skip when source has tabs — ruff and CPython handle tab
// indentation differently (CPython uses tabsize=8), so ruff may
// report false IndentationErrors for valid mixed-tab code.
if !source.contains('\t') {
for err in errors.iter() {
if !matches!(
err.error,
ParseErrorType::Lexical(LexicalErrorType::IndentationError)
) {
continue;
}
if err.location.start() <= range.start()
&& range.start() < err.location.end()
{
return Err(raise_indentation_error(vm, err, source, line_index));
}
}
}
if kind == TokenKind::EndOfFile {
continue;
}
if !extra_tokens
&& matches!(kind, TokenKind::Comment | TokenKind::NonLogicalNewline)
{
continue;
}
let raw_type = token_kind_value(kind);
let token_type = if extra_tokens && raw_type > TOKEN_DEDENT && raw_type < TOKEN_OP
{
TOKEN_OP
} else {
raw_type
};
let (token_str, start_line, start_col, end_line, end_col, line_str) =
if kind == TokenKind::Dedent {
let last_line = source.lines().count();
let default_pos = if extra_tokens {
(last_line + 1, 0)
} else {
(last_line, 0)
};
let (pos, dedent_line) =
next_non_dedent_info(tokens, *index, source, line_index, default_pos);
("", pos.0, pos.1, pos.0, pos.1, dedent_line)
} else {
let start_lc = line_index.line_column(range.start(), source);
let start_line = start_lc.line.get();
let start_col = start_lc.column.to_zero_indexed();
let implicit_newline = range.start() >= source_len;
let in_source = range.end() <= source_len;
let (s, el, ec) = if kind == TokenKind::Newline {
if extra_tokens {
if implicit_newline {
("", start_line, start_col + 1)
} else {
let s = if source[range].starts_with('\r') {
"\r\n"
} else {
"\n"
};
(s, start_line, start_col + s.len())
}
} else {
("", start_line, start_col)
}
} else if kind == TokenKind::NonLogicalNewline {
let s = if in_source { &source[range] } else { "" };
(s, start_line, start_col + s.len())
} else {
let end_lc = line_index.line_column(range.end(), source);
let s = if in_source { &source[range] } else { "" };
(s, end_lc.line.get(), end_lc.column.to_zero_indexed())
};
let line_str = source.full_line_str(range.start());
(s, start_line, start_col, el, ec, line_str)
};
// Handle FSTRING_MIDDLE/TSTRING_MIDDLE brace unescaping
if matches!(kind, TokenKind::FStringMiddle | TokenKind::TStringMiddle)
&& (token_str.contains("{{") || token_str.contains("}}"))
{
let mut parts =
split_fstring_middle(token_str, token_type, start_line, start_col)
.into_iter();
let (tt, ts, sl, sc, el, ec) = parts.next().unwrap();
let rest: Vec<_> = parts.collect();
for p in rest.into_iter().rev() {
pending_fstring_parts.push(p);
}
return Ok(PyIterReturn::Return(make_token_tuple(
vm, tt, &ts, sl, sc as isize, el, ec as isize, line_str,
)));
}
// After emitting a Rbrace inside an fstring, check if the
// next token is also Rbrace without an intervening FStringMiddle.
// CPython emits an empty FSTRING_MIDDLE in that position.
if kind == TokenKind::Rbrace
&& tokens
.get(*index)
.is_some_and(|t| t.kind() == TokenKind::Rbrace)
{
let mid_type = find_fstring_middle_type(tokens, *index);
*pending_empty_fstring_middle = Some((
mid_type,
end_line,
end_col,
line_str.to_string(),
));
}
return Ok(PyIterReturn::Return(make_token_tuple(
vm, token_type, token_str, start_line, start_col as isize, end_line,
end_col as isize, line_str,
)));
}
// Emit implicit NL before ENDMARKER if source
// doesn't end with newline and last token is Comment
if extra_tokens && core::mem::take(need_implicit_nl) {
let last_tok = tokens
.iter()
.rev()
.find(|t| t.kind() != TokenKind::EndOfFile);
if let Some(last) = last_tok.filter(|t| t.kind() == TokenKind::Comment) {
let end_lc = line_index.line_column(last.range().end(), source);
let nl_line = end_lc.line.get();
let nl_col = end_lc.column.to_zero_indexed();
return Ok(PyIterReturn::Return(make_token_tuple(
vm,
TOKEN_NL,
"",
nl_line,
nl_col as isize,
nl_line,
nl_col as isize + 1,
source.full_line_str(last.range().start()),
)));
}
}
// Check for unclosed brackets before ENDMARKER — CPython's tokenizer
// raises SyntaxError("EOF in multi-line statement") in this case.
{
let bracket_count: i32 = tokens
.iter()
.map(|t| match t.kind() {
TokenKind::Lpar | TokenKind::Lsqb | TokenKind::Lbrace => 1,
TokenKind::Rpar | TokenKind::Rsqb | TokenKind::Rbrace => -1,
_ => 0,
})
.sum();
if bracket_count > 0 {
let last_line = source.lines().count();
return Err(raise_syntax_error(
vm,
"EOF in multi-line statement",
last_line + 1,
0,
));
}
}
// All tokens consumed — emit ENDMARKER
let last_line = source.lines().count();
let (em_line, em_col, em_line_str): (usize, isize, &str) = if extra_tokens {
(last_line + 1, 0, "")
} else {
let last_line_text = source.full_line_str(TextSize::from(
source.len().saturating_sub(1) as u32,
));
(last_line, -1, last_line_text)
};
let result = make_token_tuple(
vm, TOKEN_ENDMARKER, "", em_line, em_col, em_line, em_col, em_line_str,
);
state.phase = TokenizerPhase::Done;
Ok(PyIterReturn::Return(result))
}
/// Determine whether to emit FSTRING_MIDDLE (60) or TSTRING_MIDDLE (63)
/// by looking back for the most recent FStringStart/TStringStart.
fn find_fstring_middle_type(tokens: &[Token], index: usize) -> u8 {
let mut depth = 0i32;
for i in (0..index).rev() {
match tokens[i].kind() {
TokenKind::FStringEnd | TokenKind::TStringEnd => depth += 1,
TokenKind::FStringStart => {
if depth == 0 {
return 60; // FSTRING_MIDDLE
}
depth -= 1;
}
TokenKind::TStringStart => {
if depth == 0 {
return 63; // TSTRING_MIDDLE
}
depth -= 1;
}
_ => {}
}
}
60 // default to FSTRING_MIDDLE
}
/// Find the next non-DEDENT token's position and source line.
/// Returns ((line, col), line_str).
fn next_non_dedent_info<'a>(
tokens: &[Token],
index: usize,
source: &'a str,
line_index: &LineIndex,
default_pos: (usize, usize),
) -> ((usize, usize), &'a str) {
for future in &tokens[index..] {
match future.kind() {
TokenKind::Dedent => continue,
TokenKind::EndOfFile => return (default_pos, ""),
_ => {
let flc = line_index.line_column(future.range().start(), source);
let pos = (flc.line.get(), flc.column.to_zero_indexed());
return (pos, source.full_line_str(future.range().start()));
}
}
}
(default_pos, "")
}
/// Raise a SyntaxError with the given message and position.
fn raise_syntax_error(
vm: &VirtualMachine,
msg: &str,
lineno: usize,
offset: usize,
) -> rustpython_vm::builtins::PyBaseExceptionRef {
let exc = vm.new_exception_msg(
vm.ctx.exceptions.syntax_error.to_owned(),
msg.into(),
);
let obj = exc.as_object();
let _ = obj.set_attr("msg", vm.ctx.new_str(msg), vm);
let _ = obj.set_attr("lineno", vm.ctx.new_int(lineno), vm);
let _ = obj.set_attr("offset", vm.ctx.new_int(offset), vm);
let _ = obj.set_attr("filename", vm.ctx.new_str("<string>"), vm);
let _ = obj.set_attr("text", vm.ctx.none(), vm);
exc
}
/// Raise an IndentationError from a parse error.
fn raise_indentation_error(
vm: &VirtualMachine,
err: &ParseError,
source: &str,
line_index: &LineIndex,
) -> rustpython_vm::builtins::PyBaseExceptionRef {
let err_lc = line_index.line_column(err.location.start(), source);
let err_line_text = source.full_line_str(err.location.start());
let err_text = err_line_text.trim_end_matches('\n').trim_end_matches('\r');
let msg = format!("{}", err.error);
let exc = vm.new_exception_msg(
vm.ctx.exceptions.indentation_error.to_owned(),
msg.clone().into(),
);
let obj = exc.as_object();
let _ = obj.set_attr("lineno", vm.ctx.new_int(err_lc.line.get()), vm);
let _ = obj.set_attr("offset", vm.ctx.new_int(err_text.len() as i64 + 1), vm);
let _ = obj.set_attr("msg", vm.ctx.new_str(msg), vm);
let _ = obj.set_attr("filename", vm.ctx.new_str("<string>"), vm);
let _ = obj.set_attr("text", vm.ctx.new_str(err_text), vm);
exc
}
/// Split an FSTRING_MIDDLE/TSTRING_MIDDLE token containing `{{`/`}}`
/// into multiple unescaped sub-tokens.
/// Returns vec of (type, string, start_line, start_col, end_line, end_col).
fn split_fstring_middle(
raw: &str,
token_type: u8,
start_line: usize,
start_col: usize,
) -> Vec<(u8, String, usize, usize, usize, usize)> {
let mut parts = Vec::new();
let mut current = String::new();
// Track source position (line, col) — these correspond to the
// original source positions (with {{ and }} still doubled)
let mut cur_line = start_line;
let mut cur_col = start_col;
// Track the start position of the current accumulating part
let mut part_start_line = cur_line;
let mut part_start_col = cur_col;
let mut chars = raw.chars().peekable();
// Compute end position of the current accumulated text
let end_pos = |current: &str, start_line: usize, start_col: usize| -> (usize, usize) {
let mut el = start_line;
let mut ec = start_col;
for ch in current.chars() {
if ch == '\n' {
el += 1;
ec = 0;
} else {
ec += ch.len_utf8();
}
}
(el, ec)
};
while let Some(ch) = chars.next() {
if ch == '{' && chars.peek() == Some(&'{') {
chars.next();
current.push('{');
cur_col += 2; // skip both {{ in source
} else if ch == '}' && chars.peek() == Some(&'}') {
chars.next();
// Flush accumulated text before }}
if !current.is_empty() {
let (el, ec) = end_pos(&current, part_start_line, part_start_col);
parts.push((
token_type,
core::mem::take(&mut current),
part_start_line,
part_start_col,
el,
ec,
));
}
// Emit unescaped '}' at source position of }}
parts.push((
token_type,
"}".to_string(),
cur_line,
cur_col,
cur_line,
cur_col + 1,
));
cur_col += 2; // skip both }} in source
part_start_line = cur_line;
part_start_col = cur_col;
} else {
if current.is_empty() {
part_start_line = cur_line;
part_start_col = cur_col;
}
current.push(ch);
if ch == '\n' {
cur_line += 1;
cur_col = 0;
} else {
cur_col += ch.len_utf8();
}
}
}
if !current.is_empty() {
let (el, ec) = end_pos(&current, part_start_line, part_start_col);
parts.push((token_type, current, part_start_line, part_start_col, el, ec));
}
parts
}
#[allow(clippy::too_many_arguments)]
fn make_token_tuple(
vm: &VirtualMachine,
token_type: u8,
string: &str,
start_line: usize,
start_col: isize,
end_line: usize,
end_col: isize,
line: &str,
) -> PyObjectRef {
vm.ctx
.new_tuple(vec![
token_type.to_pyobject(vm),
vm.ctx.new_str(string).into(),
vm.ctx
.new_tuple(vec![start_line.to_pyobject(vm), start_col.to_pyobject(vm)])
.into(),
vm.ctx
.new_tuple(vec![end_line.to_pyobject(vm), end_col.to_pyobject(vm)])
.into(),
vm.ctx.new_str(line).into(),
])
.into()
}
#[derive(FromArgs)]
pub struct PyTokenizerIterArgs {
#[pyarg(positional)]
readline: ArgCallable,
#[pyarg(named)]
extra_tokens: bool,
#[pyarg(named, optional)]
encoding: Option<rustpython_vm::PyRef<PyStr>>,
}
#[derive(Clone, Debug)]
struct TokenizerState {
phase: TokenizerPhase,
}
#[derive(Clone, Debug)]
enum TokenizerPhase {
Reading {
source: String,
},
Yielding {
source: String,
tokens: Vec<Token>,
errors: Vec<ParseError>,
index: usize,
line_index: LineIndex,
need_implicit_nl: bool,
/// Pending sub-tokens from FSTRING_MIDDLE splitting
pending_fstring_parts: Vec<(u8, String, usize, usize, usize, usize)>,
/// Pending empty FSTRING_MIDDLE for format spec nesting:
/// (type, line, col, line_str)
pending_empty_fstring_middle: Option<(u8, usize, usize, String)>,
},
Done,
}
const fn token_kind_value(kind: TokenKind) -> u8 {
match kind {
TokenKind::EndOfFile => 0,
TokenKind::Name
| TokenKind::For
| TokenKind::In
| TokenKind::Pass
| TokenKind::Class
| TokenKind::And
| TokenKind::Is
| TokenKind::Raise
| TokenKind::True
| TokenKind::False
| TokenKind::Assert
| TokenKind::Try
| TokenKind::While
| TokenKind::Yield
| TokenKind::Lambda
| TokenKind::None
| TokenKind::Not
| TokenKind::Or
| TokenKind::Break
| TokenKind::Continue
| TokenKind::Global
| TokenKind::Nonlocal
| TokenKind::Return
| TokenKind::Except
| TokenKind::Import
| TokenKind::Case
| TokenKind::Match
| TokenKind::Type
| TokenKind::Await
| TokenKind::With
| TokenKind::Del
| TokenKind::Finally
| TokenKind::From
| TokenKind::Def
| TokenKind::If
| TokenKind::Else
| TokenKind::Elif
| TokenKind::As
| TokenKind::Async => 1,
TokenKind::Int | TokenKind::Complex | TokenKind::Float => 2,
TokenKind::String => 3,
TokenKind::Newline => 4,
TokenKind::NonLogicalNewline => TOKEN_NL,
TokenKind::Indent => 5,
TokenKind::Dedent => 6,
TokenKind::Lpar => 7,
TokenKind::Rpar => 8,
TokenKind::Lsqb => 9,
TokenKind::Rsqb => 10,
TokenKind::Colon => 11,
TokenKind::Comma => 12,
TokenKind::Semi => 13,
TokenKind::Plus => 14,
TokenKind::Minus => 15,
TokenKind::Star => 16,
TokenKind::Slash => 17,
TokenKind::Vbar => 18,
TokenKind::Amper => 19,
TokenKind::Less => 20,
TokenKind::Greater => 21,
TokenKind::Equal => 22,
TokenKind::Dot => 23,
TokenKind::Percent => 24,
TokenKind::Lbrace => 25,
TokenKind::Rbrace => 26,
TokenKind::EqEqual => 27,
TokenKind::NotEqual => 28,
TokenKind::LessEqual => 29,
TokenKind::GreaterEqual => 30,
TokenKind::Tilde => 31,
TokenKind::CircumFlex => 32,
TokenKind::LeftShift => 33,
TokenKind::RightShift => 34,
TokenKind::DoubleStar => 35,
TokenKind::PlusEqual => 36,
TokenKind::MinusEqual => 37,
TokenKind::StarEqual => 38,
TokenKind::SlashEqual => 39,
TokenKind::PercentEqual => 40,
TokenKind::AmperEqual => 41,
TokenKind::VbarEqual => 42,
TokenKind::CircumflexEqual => 43,
TokenKind::LeftShiftEqual => 44,
TokenKind::RightShiftEqual => 45,
TokenKind::DoubleStarEqual => 46,
TokenKind::DoubleSlash => 47,
TokenKind::DoubleSlashEqual => 48,
TokenKind::At => 49,
TokenKind::AtEqual => 50,
TokenKind::Rarrow => 51,
TokenKind::Ellipsis => 52,
TokenKind::ColonEqual => 53,
TokenKind::Exclamation => 54,
TokenKind::FStringStart => 59,
TokenKind::FStringMiddle => 60,
TokenKind::FStringEnd => 61,
TokenKind::Comment => TOKEN_COMMENT,
TokenKind::TStringStart => 62,
TokenKind::TStringMiddle => 63,
TokenKind::TStringEnd => 64,
TokenKind::IpyEscapeCommand
| TokenKind::Question
| TokenKind::Unknown => 67, // ERRORTOKEN
}
}
}
+3
View File
@@ -49,6 +49,8 @@ mod pystruct;
mod random;
mod statistics;
mod suggestions;
#[path = "_tokenize.rs"]
mod _tokenize;
// TODO: maybe make this an extension module, if we ever get those
// mod re;
#[cfg(all(feature = "host_env", not(target_arch = "wasm32")))]
@@ -225,6 +227,7 @@ pub fn stdlib_module_defs(ctx: &Context) -> Vec<&'static builtins::PyModuleDef>
ssl::module_def(ctx),
statistics::module_def(ctx),
suggestions::module_def(ctx),
_tokenize::module_def(ctx),
#[cfg(all(feature = "host_env", unix, not(target_os = "redox")))]
syslog::module_def(ctx),
#[cfg(all(