Skip to content

session

Track the reachable parsed-module graph, import edges, diagnostics, and visible bindings for one multi-module analysis run.

Classes:

CompilationSession dataclass

CompilationSession(
    root: ParsedModule,
    resolver: ImportResolver,
    modules: dict[ModuleKey, ParsedModule] = dict(),
    graph: dict[ModuleKey, set[ModuleKey]] = dict(),
    load_order: list[ModuleKey] = list(),
    diagnostics: DiagnosticBag = DiagnosticBag(),
    visible_bindings: dict[
        ModuleKey, dict[str, SemanticBinding]
    ] = dict(),
    _resolution_cache: dict[
        tuple[ModuleKey, str], ParsedModule | None
    ] = dict(),
)

Own the loaded module graph and cross-module binding state that analysis and lowering share for one compilation. attributes: root: type: ParsedModule resolver: type: ImportResolver modules: type: dict[ModuleKey, ParsedModule] graph: type: dict[ModuleKey, set[ModuleKey]] load_order: type: list[ModuleKey] diagnostics: type: DiagnosticBag visible_bindings: type: dict[ModuleKey, dict[str, SemanticBinding]] _resolution_cache: type: dict[tuple[ModuleKey, str], ParsedModule | None]

Methods:

expand_graph

expand_graph() -> None

Walk top-level imports from the root module, load every reachable dependency, and reject cycles.

Source code in src/irx/analysis/session.py
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def expand_graph(self) -> None:
    """
    title: Expand the reachable import graph from the root module.
    summary: >-
      Walk top-level imports from the root module, load every reachable
      dependency, and reject cycles.
    """
    self.load_order.clear()
    temporary: list[ModuleKey] = []
    temporary_lookup: set[ModuleKey] = set()
    permanent: set[ModuleKey] = set()

    def dfs(module_key: ModuleKey) -> None:
        """
        title: Visit one reachable module during graph expansion.
        summary: >-
          Depth-first walk one module, record its outgoing edges, and
          append it to the stable load order after its dependencies.
        parameters:
          module_key:
            type: ModuleKey
        """
        if module_key in permanent:
            return
        if module_key in temporary_lookup:
            cycle_start = temporary.index(module_key)
            cycle_path = [*temporary[cycle_start:], module_key]
            cycle_str = " -> ".join(str(item) for item in cycle_path)
            self.diagnostics.add(
                f"Cyclic import detected: {cycle_str}",
                node=self.modules[module_key].ast,
                module_key=module_key,
            )
            return

        temporary.append(module_key)
        temporary_lookup.add(module_key)

        parsed_module = self.modules[module_key]
        dependencies: list[ModuleKey] = []
        for node in parsed_module.ast.nodes:
            if isinstance(node, astx.ImportStmt):
                for alias in node.names:
                    resolved = self.resolve_import_specifier(
                        module_key,
                        node,
                        alias.name,
                    )
                    if resolved is None:
                        continue
                    self.graph.setdefault(module_key, set()).add(
                        resolved.key
                    )
                    dependencies.append(resolved.key)
            elif isinstance(node, astx.ImportFromStmt):
                resolved = self.resolve_import_specifier(
                    module_key,
                    node,
                    _module_import_specifier(node),
                )
                if resolved is None:
                    continue
                self.graph.setdefault(module_key, set()).add(resolved.key)
                dependencies.append(resolved.key)

        for dependency_key in dependencies:
            dfs(dependency_key)

        temporary.pop()
        temporary_lookup.remove(module_key)
        permanent.add(module_key)
        self.load_order.append(module_key)

    dfs(self.root.key)

module

module(module_key: ModuleKey) -> ParsedModule

Look up a previously-registered parsed module by its canonical host key. parameters: module_key: type: ModuleKey returns: type: ParsedModule

Source code in src/irx/analysis/session.py
111
112
113
114
115
116
117
118
119
120
121
122
123
def module(self, module_key: ModuleKey) -> ParsedModule:
    """
    title: Return a parsed module by key.
    summary: >-
      Look up a previously-registered parsed module by its canonical host
      key.
    parameters:
      module_key:
        type: ModuleKey
    returns:
      type: ParsedModule
    """
    return self.modules[module_key]

ordered_modules

ordered_modules() -> list[ParsedModule]

Materialize the dependency-ordered module list used by later semantic and lowering passes. returns: type: list[ParsedModule]

Source code in src/irx/analysis/session.py
125
126
127
128
129
130
131
132
133
134
def ordered_modules(self) -> list[ParsedModule]:
    """
    title: Return parsed modules in stable dependency order.
    summary: >-
      Materialize the dependency-ordered module list used by later semantic
      and lowering passes.
    returns:
      type: list[ParsedModule]
    """
    return [self.modules[module_key] for module_key in self.load_order]

register_module

register_module(
    parsed_module: ParsedModule,
) -> ParsedModule

Cache a parsed module once and initialize its graph and visible binding slots. parameters: parsed_module: type: ParsedModule returns: type: ParsedModule

Source code in src/irx/analysis/session.py
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def register_module(self, parsed_module: ParsedModule) -> ParsedModule:
    """
    title: Register one parsed module in the session cache.
    summary: >-
      Cache a parsed module once and initialize its graph and visible
      binding slots.
    parameters:
      parsed_module:
        type: ParsedModule
    returns:
      type: ParsedModule
    """
    existing = self.modules.get(parsed_module.key)
    if existing is not None:
        return existing
    self.modules[parsed_module.key] = parsed_module
    self.graph.setdefault(parsed_module.key, set())
    self.visible_bindings.setdefault(parsed_module.key, {})
    return parsed_module

resolve_import_specifier

resolve_import_specifier(
    requesting_module_key: ModuleKey,
    import_node: ImportStmt | ImportFromStmt,
    requested_specifier: str,
) -> ParsedModule | None

Call the host resolver once per import request, memoizing both successes and failures. parameters: requesting_module_key: type: ModuleKey import_node: type: astx.ImportStmt | astx.ImportFromStmt requested_specifier: type: str returns: type: ParsedModule | None

Source code in src/irx/analysis/session.py
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def resolve_import_specifier(
    self,
    requesting_module_key: ModuleKey,
    import_node: astx.ImportStmt | astx.ImportFromStmt,
    requested_specifier: str,
) -> ParsedModule | None:
    """
    title: Resolve one import request through the host resolver.
    summary: >-
      Call the host resolver once per import request, memoizing both
      successes and failures.
    parameters:
      requesting_module_key:
        type: ModuleKey
      import_node:
        type: astx.ImportStmt | astx.ImportFromStmt
      requested_specifier:
        type: str
    returns:
      type: ParsedModule | None
    """
    cache_key = (requesting_module_key, requested_specifier)
    if cache_key in self._resolution_cache:
        return self._resolution_cache[cache_key]

    try:
        resolved = self.resolver(
            requesting_module_key,
            import_node,
            requested_specifier,
        )
    except Exception as exc:
        self.diagnostics.add(
            f"Unable to resolve module '{requested_specifier}': {exc}",
            node=import_node,
            module_key=requesting_module_key,
        )
        self._resolution_cache[cache_key] = None
        return None

    self.register_module(resolved)
    self._resolution_cache[cache_key] = resolved
    return resolved