Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -20,3 +20,4 @@ uv.lock
.ropeproject
node_modules
mutants
AGENTS.md
11 changes: 8 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -168,9 +168,14 @@ def change_add(node: Add, context: Context): # <- The function takes a second a
)
```

The context object has two main fields and one useful method:

- `coordinate` with fields `start_line: int`, `start_column: int`, `end_line: int`, `end_column: int` and some others — identifies the current location in the code.
The frozen `Context` dataclass provides the following attributes and method:

- `position: SourcePosition` — the node’s position in the original source. It provides:
- `coordinate` with fields `start_line: int`, `start_column: int`, `end_line: int`, `end_column: int` and some others — identifies the current syntactic location in the code;
- `source` — the complete original source passed to `Changer`;
- `node_range` — the node’s [`WhitespaceInclusivePositionProvider`](https://libcst.readthedocs.io/en/latest/metadata.html#libcst.metadata.WhitespaceInclusivePositionProvider) range, including whitespace owned by that node;
- `start_offset: int` and `end_offset: int` — the lazily computed inclusive start and exclusive end indices of that range in Python characters;
- `code_before` and `code_after` — the lazily computed source text before and after that range.
- `comment` — the comment on the node’s first line, if there is one, without the leading `#`, or `None` if there is no comment.
- `get_metacodes(key: Union[str, List[str]]) -> List[ParsedComment]` — a method that returns a list of parsed comments in [metacode format](https://github.com/mutating/metacode) associated with the current line of code.

Expand Down
7 changes: 5 additions & 2 deletions cstvis/changer.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@

from cstvis.collector import Collector
from cstvis.dto import Context, Coordinate
from cstvis.source_offsets import SourceOffsetResolver
from cstvis.transformers.super_transformer import SuperTransformer
from cstvis.visitors.bloodhound import Bloodhound
from cstvis.visitors.comments_aggregator import CommentsAggregator
Expand Down Expand Up @@ -66,12 +67,14 @@ def converter(self, function: Optional[Union[Callable[[CSTNode], CSTNode], Calla

def iterate_coordinates(self) -> Generator[Coordinate, None, None]:
wrapper = metadata.MetadataWrapper(self.module)
printer = Bloodhound(self.converters_by_types, self._comments_by_lines, self.filters_by_types)
node_ranges = wrapper.resolve(metadata.WhitespaceInclusivePositionProvider)
printer = Bloodhound(self.converters_by_types, self._comments_by_lines, self.filters_by_types, SourceOffsetResolver(wrapper.module, self.source, node_ranges.values()))

wrapper.visit(printer)
yield from printer.coordinates

def apply_coordinate(self, coordinate: Coordinate) -> str:
wrapper = metadata.MetadataWrapper(self.module)
modified = wrapper.visit(SuperTransformer(coordinate, self.converters_by_types, self._comments_by_lines, set()))
node_ranges = wrapper.resolve(metadata.WhitespaceInclusivePositionProvider)
modified = wrapper.visit(SuperTransformer(coordinate, self.converters_by_types, self._comments_by_lines, set(), SourceOffsetResolver(wrapper.module, self.source, node_ranges.values())))
return modified.code
95 changes: 91 additions & 4 deletions cstvis/dto.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,11 @@
from dataclasses import dataclass
from dataclasses import dataclass, field
from functools import cached_property
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
from typing import Any, Callable, Dict, List, Optional, Tuple, Union

from libcst.metadata import CodeRange
from metacode import ParsedComment, parse
from printo import describe_call


@dataclass
Expand All @@ -15,12 +18,96 @@ class Coordinate:
end_column: int
converter_id: Optional[str] = None

@dataclass
class Context:

@dataclass(frozen=True, repr=False)
class SourcePosition:
"""
Describe and lazily partition a node's span in the original source.

``coordinate`` is the ordinary syntactic position, while ``node_range`` is
the node's whitespace-inclusive LibCST range. The required resolver maps
that range to absolute Python-character offsets in ``source``. It is
normally shared by every position produced during one traversal, so the
first access performs one alignment pass and later positions reuse its
results.

The range identifies a contextual source span rather than
``Module.code_for_node()`` output, which may lose ambient indentation or
add a final newline. For the exact source slice ``node_source_span``::

code_before + node_source_span + code_after == source

Offsets and the two surrounding source slices are cached independently.
"""

coordinate: Coordinate
source: str
node_range: CodeRange
offset_resolver: Callable[[CodeRange], Tuple[int, int]] = field(repr=False, compare=False)

@cached_property
def start_offset(self) -> int:
"""Lazily resolve and cache the node span's starting source index."""
return self.offset_resolver(self.node_range)[0]

@cached_property
def end_offset(self) -> int:
"""Lazily resolve and cache the node span's ending source index."""
return self.offset_resolver(self.node_range)[1]

@cached_property
def code_before(self) -> str:
"""Lazily slice and cache the exact source prefix before node_range."""
return self.source[:self.start_offset]

@cached_property
def code_after(self) -> str:
"""Lazily slice and cache the exact source suffix after node_range."""
return self.source[self.end_offset:]

def __repr__(self) -> str:
"""Describe public source data and lazy offsets with an item limit of 80."""
return describe_call(
type(self),
[],
{
'coordinate': self.coordinate,
'source': self.source,
'node_range': self.node_range,
'start_offset': self.start_offset,
'end_offset': self.end_offset,
},
item_limit=80,
)


@dataclass(frozen=True, repr=False)
class Context:
"""
Describe a callback invocation and its position in the original source.

Source-related data and lazy fragments are exposed through ``position``.
The dataclass is frozen, although a dictionary supplied as ``meta`` remains
mutable.
"""

position: SourcePosition
comment: Optional[str]
meta: Optional[Dict[str, Any]] = None

def __repr__(self) -> str:
"""Describe public callback data with an item limit of 80."""
return describe_call(
type(self),
[],
{
'position': self.position,
'comment': self.comment,
'meta': self.meta,
},
item_limit=80,
)

def get_metacodes(self, key: Union[str, List[str]]) -> List[ParsedComment]:
if self.comment is None:
return []
Expand Down
101 changes: 101 additions & 0 deletions cstvis/source_offsets.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
from typing import Dict, Iterable, Optional, Set, Tuple

from libcst import Module
from libcst._nodes.internal import CodegenState
from libcst.metadata import CodePosition, CodeRange


class _SourceOffsetCodegenState(CodegenState):
"""
Align LibCST code-generation positions with the original source string.

Generated positions can diverge from original-source offsets when LibCST
omits source text such as an initial BOM, form-feed prefixes, or explicit
line continuations. Token-by-token alignment preserves those characters as
well as Unicode and mixed LF, CRLF, and CR line endings in the resulting
source partition.
"""

def __init__(self, module: Module, source: str, target_positions: Set[CodePosition]) -> None:
super().__init__(default_indent=module.default_indent, default_newline=module.default_newline)
self.source = source
self.source_offset = int(source.startswith('\ufeff'))
self.line = 1
self.column = 0
self.target_positions = target_positions
self.right_position_offsets: Dict[CodePosition, int] = {}

def _record_right_position(self, source_offset: int) -> None:
position = CodePosition(self.line, self.column)
if position in self.target_positions:
self.right_position_offsets[position] = source_offset

def _consume_source_token(self, value: str, search_forward: bool) -> Optional[int]:
if self.source.startswith(value, self.source_offset):
match_start = self.source_offset
elif search_forward:
match_start = self.source.find(value, self.source_offset)
else:
match_start = -1

if match_start >= 0:
self.source_offset = match_start + len(value)
return match_start
return None

def _add_generated_token(self, value: str, search_forward: bool) -> None:
source_start = self._consume_source_token(value, search_forward)
aligned_source_offset = self.source_offset if source_start is None else source_start
self._record_right_position(aligned_source_offset)

cursor = 0
while cursor < len(value):
if value[cursor] == '\r' and cursor + 1 < len(value) and value[cursor + 1] == '\n':
cursor += 2
self.line += 1
self.column = 0
elif value[cursor] in {'\r', '\n'}:
cursor += 1
self.line += 1
self.column = 0
else:
cursor += 1
self.column += 1

position_source_offset = self.source_offset if source_start is None else source_start + cursor
self._record_right_position(position_source_offset)

def add_indent_tokens(self) -> None:
for token in self.indent_tokens:
self._add_generated_token(token, search_forward=False)
self.tokens.extend(self.indent_tokens)

def add_token(self, value: str) -> None:
self._add_generated_token(value, search_forward=True)
self.tokens.append(value)


class SourceOffsetResolver:
"""
Lazily map registered whitespace-inclusive ranges to source offsets.

The first requested range triggers one shared code-generation pass for all
positions registered at construction. Later ranges reuse the cached
character offsets. Alignment reads the original source directly without
copying it in full or splitting it into lines.
"""

def __init__(self, module: Module, source: str, node_ranges: Iterable[CodeRange]) -> None:
self.module = module
self.source = source
self.target_positions = {position for node_range in node_ranges for position in (node_range.start, node_range.end)}
self._offsets: Dict[CodePosition, int] = {}

def __call__(self, node_range: CodeRange) -> Tuple[int, int]:
"""Return character offsets for a range registered at construction."""
if not self._offsets:
state = _SourceOffsetCodegenState(self.module, self.source, self.target_positions)
self.module._codegen(state)
state._record_right_position(state.source_offset)
self._offsets.update(state.right_position_offsets)
return self._offsets[node_range.start], self._offsets[node_range.end]
29 changes: 23 additions & 6 deletions cstvis/transformers/super_transformer.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,8 @@
leave,
)

from cstvis.dto import Context, Coordinate
from cstvis.dto import Context, Coordinate, SourcePosition
from cstvis.source_offsets import SourceOffsetResolver
from cstvis.wrapper import CallableWrapper


Expand All @@ -34,19 +35,29 @@ def leave_all(function: Callable[[Any, CSTNode, CSTNode], CSTNode]) -> Callable[


class SuperTransformer(MatcherDecoratableTransformer):
METADATA_DEPENDENCIES = (metadata.PositionProvider,)
"""
Apply one conversion with positions from the original node.

Public coordinates use ``PositionProvider``; contextual ranges use the
additional ``WhitespaceInclusivePositionProvider`` metadata pass. Reading
``Context.position`` later resolves the original node's source offsets.
"""

METADATA_DEPENDENCIES = (metadata.PositionProvider, metadata.WhitespaceInclusivePositionProvider)

def __init__(
self,
target_coordinate: Coordinate,
nodes_mapping: Dict[Type[CSTNode], List[CallableWrapper[CSTNode]]],
comments: Dict[int, str],
nodes_ids: Set[int],
source_offsets: SourceOffsetResolver,
):
self.target_coordinate = target_coordinate
self.nodes_mapping = nodes_mapping
self.comments = comments
self.nodes_ids = nodes_ids
self.source_offsets = source_offsets

super().__init__()

Expand All @@ -56,6 +67,10 @@ def leave(self, original_node, updated_node): # type: ignore[no-untyped-def]
return updated_node
self.nodes_ids.add(id(original_node))

converters = self.nodes_mapping.get(type(original_node), []) + self.nodes_mapping.get(CSTNode, []) # type: ignore[type-abstract]
if not converters:
return updated_node

position = self.get_metadata(metadata.PositionProvider, original_node)
coordinate = Coordinate(
file=None,
Expand All @@ -74,11 +89,13 @@ def leave(self, original_node, updated_node): # type: ignore[no-untyped-def]
end_column=self.target_coordinate.end_column,
)

converters = self.nodes_mapping.get(type(original_node), []) + self.nodes_mapping.get(CSTNode, []) # type: ignore[type-abstract]

if coordinate == target_coordinate_without_converter_id and converters:
context = Context(coordinate, self.comments.get(coordinate.start_line))
if coordinate == target_coordinate_without_converter_id:
for converter in converters: # pragma: no branch
if converter.get_function_id() == self.target_coordinate.converter_id:
node_range = self.get_metadata(metadata.WhitespaceInclusivePositionProvider, original_node)
context = Context(
SourcePosition(coordinate, self.source_offsets.source, node_range, self.source_offsets),
self.comments.get(coordinate.start_line),
)
return converter(updated_node, context)
return updated_node
Loading
Loading