Module

parsing.shared_tokens

Shared token stream via ContextVar.

Eliminates re-tokenization by sharing tokens across sub-parsers. Each parser gets a cursor into the shared stream.

Architecture:

  • TokenStream: Holds all tokens for a document (stored in ContextVar)
  • TokenCursor: Lightweight view into stream with position tracking
  • Sub-parsers create cursors, not new streams

Performance:

  • Zero re-tokenization for nested content
  • O(1) cursor creation vs O(n) tokenization
  • Memory: Single token list vs copies

Thread Safety:

  • ContextVar ensures thread-local token streams
  • Multiple threads can parse different documents safely
  • Cursors are not shared across threads
3Classes3Functions

Classes

TokenStream 4 ▼
Shared token stream for a document. Stored in ContextVar for thread-local access. Sub-parsers crea…

Shared token stream for a document.

Stored in ContextVar for thread-local access. Sub-parsers create cursors instead of re-tokenizing.

Methods

types 0 frozenset[TokenType] ▼
Get unique token types (cached).
property
def types(self) -> frozenset[TokenType]
Returns
frozenset[TokenType]
cursor 1 TokenCursor ▼
Create a cursor into this stream.
def cursor(self, start: int = 0) -> TokenCursor
Parameters
Name Type Description
start — Default:0
Returns
TokenCursor
slice_cursor 2 TokenCursor ▼
Create a cursor for a slice of the stream.
def slice_cursor(self, start: int, end: int) -> TokenCursor
Parameters
Name Type Description
start —
end —
Returns
TokenCursor
Internal Methods 1 ▼
__init__ 1 ▼
def __init__(self, tokens: list[PatitasToken]) -> None
Parameters
Name Type Description
tokens —
TokenCursor 8 ▼
Lightweight cursor into a TokenStream. Provides iterator interface without copying tokens. Used by…

Lightweight cursor into a TokenStream.

Provides iterator interface without copying tokens. Used by sub-parsers to process nested content.

Methods

current 0 PatitasToken | None ▼
Current token (cached for performance).
property
def current(self) -> PatitasToken | None
Returns
PatitasToken | None
at_end 0 bool ▼
Check if cursor is at end of range.
def at_end(self) -> bool
Returns
bool
advance 0 PatitasToken | None ▼
Move to next token, return previous.
def advance(self) -> PatitasToken | None
Returns
PatitasToken | None
peek 1 PatitasToken | None ▼
Look ahead without advancing.
def peek(self, offset: int = 0) -> PatitasToken | None
Parameters
Name Type Description
offset — Default:0
Returns
PatitasToken | None
fork 0 TokenCursor ▼
Create a new cursor at current position.
def fork(self) -> TokenCursor
Returns
TokenCursor
remaining 0 int ▼
Number of tokens remaining.
def remaining(self) -> int
Returns
int
Internal Methods 2 ▼
__init__ 3 ▼
def __init__(self, stream: TokenStream, start: int = 0, end: int | None = None) -> None
Parameters
Name Type Description
stream —
start — Default:0
end — Default:None
__iter__ 0 Iterator[PatitasToken] ▼
Iterate over remaining tokens.
def __iter__(self) -> Iterator[PatitasToken]
Returns
Iterator[PatitasToken]
TokenStreamContext 5 ▼
Context manager for token stream lifecycle. Usage: with TokenStreamContext(tokens) as stream: …

Context manager for token stream lifecycle.

Usage:

with TokenStreamContext(tokens) as stream:
    cursor = stream.cursor()
    # parse with cursor

Attributes

Name Type Description
stream TokenStream —
_token ContextVarToken[TokenStream | None] | None —

Methods

Internal Methods 3 ▼
__init__ 1 ▼
def __init__(self, tokens: list[PatitasToken]) -> None
Parameters
Name Type Description
tokens —
__enter__ 0 TokenStream ▼
def __enter__(self) -> TokenStream
Returns
TokenStream
__exit__ 1 ▼
def __exit__(self, *args: object) -> None
Parameters
Name Type Description
*args —

Functions

set_token_stream 1 None ▼
Set the current thread's token stream.
def set_token_stream(stream: TokenStream) -> None
Parameters
Name Type Description
stream TokenStream
get_token_stream 0 TokenStream | None ▼
Get the current thread's token stream.
def get_token_stream() -> TokenStream | None
Returns
TokenStream | None
clear_token_stream 0 None ▼
Clear the current thread's token stream.
def clear_token_stream() -> None