graph

CodeGraph: call-graph layer for kosha, backed by graph.db. Augments semantic search with structural traversal.

---

CodeGraph

def CodeGraph(
    path:str | pathlib.Path=':memory:'
):

Call graph builder and query engine. Uses pyan3 for static analysis and custom AST parsing for dynamic edges. Stores everything in a SQLite database.

cg = CodeGraph(':memory:')

tables = {r[0] for r in cg.db.execute("SELECT name FROM sqlite_master WHERE type='table'")}
assert {'graph_edges','graph_nodes','co_dispatch','file_index'} <= tables, f"Missing tables: {tables}"
print("schema ok:", tables)

schema ok: {'graph_edges', 'file_index', 'co_dispatch', 'graph_nodes', 'pending_attr_calls'}

---

dyn_edges

def dyn_edges(
    src:str, module:str
)->list:

All dynamic dispatch edges: decorator, connect_body, registration, co_dispatch.

from fastcore.test import test_eq

_src = "from litesearch import *\ndef f(self): self.db.search(q)\ndef g(self): self.st.insert(x)"
_meta = {'m.f': {}, 'm.g': {}}
_hints = _attr_call_hints(sources={'m': _src}, meta=_meta)
test_eq(len(_hints), 2)
test_eq({h['method'] for h in _hints}, {'search','insert'})
assert all('litesearch' in h['pkgs'].split(',') for h in _hints)
# fn not in meta → excluded
test_eq(_attr_call_hints(sources={'m': _src}, meta={}), [])
# depth-1 self.method() and non-self calls → excluded
_src2 = "def f(self): self.direct(); other.thing()"
test_eq(_attr_call_hints(sources={'m2': _src2}, meta={'m2.f':{}}), [])

src="""
def f(self):
    self.db.search(q)
    self.store.insert(x)
    self.direct()        # depth 1 — skip
    other.thing()        # not self — skip
"""
_fn = [n for n in ast.walk(ast.parse(_src)) if isinstance(n, ast.FunctionDef)]
test_eq(_attr_calls(_fn[0]), {'search'})
test_eq(_attr_calls(_fn[1]), {'insert'})

_SRC = '''
import flask
app = flask.Flask(__name__)

@app.route('/hello')
def hello(): pass

def setup(other_app):
    other_app.register(hello)
'''
edges = dyn_edges(_SRC, 'mymod')
kinds = {e['kind'] for e in edges}
assert 'decorator' in kinds, f"Expected decorator edge, got: {edges}"
assert 'registration' in kinds, f"Expected registration edge, got: {edges}"
assert dyn_edges('def (broken', 'bad') == L()
print("_dyn_edges ok:", edges)

_dyn_edges ok: [{'caller': 'mymod.app', 'callee': 'mymod.hello', 'kind': 'decorator', 'confidence': 0.75}, {'caller': 'mymod.setup', 'callee': 'mymod.hello', 'kind': 'connect_body', 'confidence': 0.9}, {'caller': 'mymod.other_app', 'callee': 'mymod.hello', 'kind': 'registration', 'confidence': 0.75}]

_DELEGATES_SRC = '''
from fastcore.all import delegates

def target(x, y, z=1): pass

@delegates(target)
def wrapper_pos(**kw): pass

@delegates(to=target)
def wrapper_kw(**kw): pass

@delegates
def bare(**kw): pass

@delegates()
def empty(**kw): pass
'''
dyn_edges(_DELEGATES_SRC, 'mymod')

[{'caller': 'mymod.wrapper_pos',
  'callee': 'mymod.target',
  'kind': 'delegates',
  'confidence': 0.85},
 {'caller': 'mymod.wrapper_kw',
  'callee': 'mymod.to=target',
  'kind': 'delegates',
  'confidence': 0.85}]

_PATCH_SRC = '''
from fastcore.all import patch, delegates
from apswutils.db import Database, Table, View

@patch
def basic(self:Database, x): pass

@patch(as_prop=True)
def prop(self:Database): pass

@patch(cls_method=True)
def cls_m(cls:Database): pass

@patch
def no_ann(self): pass

@patch
def union_m(self:Table|View): pass
'''
dyn_edges(_PATCH_SRC, 'mymod')

[{'caller': 'mymod.Database',
  'callee': 'mymod.basic',
  'kind': 'patch',
  'confidence': 0.9},
 {'caller': 'mymod.Database',
  'callee': 'mymod.prop',
  'kind': 'patch',
  'confidence': 0.9},
 {'caller': 'mymod.Database',
  'callee': 'mymod.cls_m',
  'kind': 'patch',
  'confidence': 0.9},
 {'caller': 'mymod.Table',
  'callee': 'mymod.union_m',
  'kind': 'patch',
  'confidence': 0.9},
 {'caller': 'mymod.View',
  'callee': 'mymod.union_m',
  'kind': 'patch',
  'confidence': 0.9}]

---

CodeGraph.resolve_attr_calls

def resolve_attr_calls(
    
)->CodeGraph:

Post-sync: resolve pending attr_call hints into edges using populated graph_nodes.

---

static_edges

def static_edges(
    sources:dict=None, filenames:list=None, root:str=None
)->tuple:

Static call edges via pyan3. Provide either filenames+root or sources dict.

edges, meta = static_edges({"mod": "def foo(): pass\ndef bar(): foo()"})
meta['mod.foo']

{'node': 'mod.foo', 'flavor': 'function', 'file': 'mod'}

assert all(k in edges[0] for k in ('caller','callee','kind','confidence'))
assert all(e['kind'] == 'static' for e in edges)

static_edges({"bad": "def (: !!!"})

pyan3 static analysis failed: invalid syntax (bad, line 1)

([], {})

from textwrap import dedent

_src1 = dedent('''
class Database:
    def query(self): pass
''')
_src2 = dedent('''
from mod1 import Database
def use_db():
    db = Database()
    db.query()
''')
_edges, _ = static_edges({"mod1": _src1, "mod2": _src2})
assert not any(e['callee'].startswith('None.') for e in _edges), \
    f"None namespace callees found: {[e['callee'] for e in _edges if e['callee'].startswith('None.')]}"
print('cross-module callee fix ok:', [e['callee'] for e in _edges])

cross-module callee fix ok: ['mod1.Database.query', 'mod1.Database']

static_edges(filenames=['/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'], root=Path('/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages'))

([{'caller': 'litesearch.core._dtype_suffix',
   'callee': 'litesearch.core._dtype_suffixes',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.query',
   'callee': 'fastcore.all.Generator',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.query',
   'callee': 'fastcore.all.patch',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.query',
   'callee': 'fastlite.Database',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.query',
   'callee': 'apswutils.utils.cursor_row2dict',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.query',
   'callee': 'fastcore.all.Union',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.query',
   'callee': 'fastcore.all.Iterable',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.query',
   'callee': 'fastcore.all.Optional',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.get_store',
   'callee': 'fastcore.all.patch',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.get_store',
   'callee': 'fastlite.Database',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.vec_search',
   'callee': 'apswutils.db.Table',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.vec_search',
   'callee': 'litesearch.core._dtype_suffix',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.vec_search',
   'callee': 'fastcore.all.patch',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.vec_search',
   'callee': 'fastcore.all.ifnone',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.rrf_merge',
   'callee': 'fastcore.all.merge',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.database',
   'callee': 'fastlite.Database',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.database',
   'callee': 'fastcore.all.Path',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.database',
   'callee': 'usearch.sqlite_path',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.search',
   'callee': 'fastlite.Database',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.search',
   'callee': 'fastcore.all.merge',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.search',
   'callee': 'litesearch.core.rrf_merge',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.search',
   'callee': 'fastcore.all.patch',
   'kind': 'static',
   'confidence': 1.0},
  {'caller': 'litesearch.core.search',
   'callee': 'fastcore.all.ifnone',
   'kind': 'static',
   'confidence': 1.0}],
 {'litesearch.core.str': {'node': 'litesearch.core.str',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core._dtype_suffix.str': {'node': 'litesearch.core._dtype_suffix.str',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.query.str': {'node': 'litesearch.core.query.str',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.get_store.str': {'node': 'litesearch.core.get_store.str',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.vec_search.str': {'node': 'litesearch.core.vec_search.str',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.vec_search.listcomp.0.str': {'node': 'litesearch.core.vec_search.listcomp.0.str',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.rrf_merge.str': {'node': 'litesearch.core.rrf_merge.str',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.rrf_merge.lambda.0.str': {'node': 'litesearch.core.rrf_merge.lambda.0.str',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.database.str': {'node': 'litesearch.core.database.str',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.search.str': {'node': 'litesearch.core.search.str',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.__all__': {'node': 'litesearch.core.__all__',
   'flavor': 'name',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'fastcore.all.Path': {'node': 'fastcore.all.Path',
   'flavor': 'importeditem',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'fastcore.all.Generator': {'node': 'fastcore.all.Generator',
   'flavor': 'importeditem',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'fastcore.all.patch': {'node': 'fastcore.all.patch',
   'flavor': 'importeditem',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'fastcore.all.Optional': {'node': 'fastcore.all.Optional',
   'flavor': 'importeditem',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'fastcore.all.Union': {'node': 'fastcore.all.Union',
   'flavor': 'importeditem',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'fastcore.all.Iterable': {'node': 'fastcore.all.Iterable',
   'flavor': 'importeditem',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'fastcore.all.merge': {'node': 'fastcore.all.merge',
   'flavor': 'importeditem',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'fastcore.all.ifnone': {'node': 'fastcore.all.ifnone',
   'flavor': 'importeditem',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'fastlite.Database': {'node': 'fastlite.Database',
   'flavor': 'importeditem',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'apswutils.db.Table': {'node': 'apswutils.db.Table',
   'flavor': 'importeditem',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'apswutils.utils.cursor_row2dict': {'node': 'apswutils.utils.cursor_row2dict',
   'flavor': 'importeditem',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'numpy.int8': {'node': 'numpy.int8',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'numpy.float16': {'node': 'numpy.float16',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'numpy.float64': {'node': 'numpy.float64',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'numpy.float32': {'node': 'numpy.float32',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core._dtype_suffixes': {'node': 'litesearch.core._dtype_suffixes',
   'flavor': 'name',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core._dtype_suffix': {'node': 'litesearch.core._dtype_suffix',
   'flavor': 'function',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core._dtype_suffix.^^^argument^^^': {'node': 'litesearch.core._dtype_suffix.^^^argument^^^',
   'flavor': 'unspecified',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.query.^^^argument^^^': {'node': 'litesearch.core.query.^^^argument^^^',
   'flavor': 'unspecified',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.get_store.^^^argument^^^': {'node': 'litesearch.core.get_store.^^^argument^^^',
   'flavor': 'unspecified',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.vec_search.^^^argument^^^': {'node': 'litesearch.core.vec_search.^^^argument^^^',
   'flavor': 'unspecified',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.rrf_merge.^^^argument^^^': {'node': 'litesearch.core.rrf_merge.^^^argument^^^',
   'flavor': 'unspecified',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.rrf_merge.lambda.0.^^^argument^^^': {'node': 'litesearch.core.rrf_merge.lambda.0.^^^argument^^^',
   'flavor': 'unspecified',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.database.^^^argument^^^': {'node': 'litesearch.core.database.^^^argument^^^',
   'flavor': 'unspecified',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.search.^^^argument^^^': {'node': 'litesearch.core.search.^^^argument^^^',
   'flavor': 'unspecified',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.query': {'node': 'litesearch.core.query',
   'flavor': 'function',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.NoneType': {'node': 'litesearch.core.NoneType',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.query.NoneType': {'node': 'litesearch.core.query.NoneType',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.vec_search.NoneType': {'node': 'litesearch.core.vec_search.NoneType',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.search.NoneType': {'node': 'litesearch.core.search.NoneType',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.query.^^^argument^^^.execute': {'node': 'litesearch.core.query.^^^argument^^^.execute',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.get_store': {'node': 'litesearch.core.get_store',
   'flavor': 'function',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.bool': {'node': 'litesearch.core.bool',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.get_store.bool': {'node': 'litesearch.core.get_store.bool',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.rrf_merge.bool': {'node': 'litesearch.core.rrf_merge.bool',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.database.bool': {'node': 'litesearch.core.database.bool',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.search.bool': {'node': 'litesearch.core.search.bool',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.rrf_merge.float': {'node': 'litesearch.core.rrf_merge.float',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.int': {'node': 'litesearch.core.int',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.vec_search.int': {'node': 'litesearch.core.vec_search.int',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.search.int': {'node': 'litesearch.core.search.int',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.get_store.^^^argument^^^.t': {'node': 'litesearch.core.get_store.^^^argument^^^.t',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.search.^^^argument^^^.t': {'node': 'litesearch.core.search.^^^argument^^^.t',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.vec_search': {'node': 'litesearch.core.vec_search',
   'flavor': 'function',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.NoneType.__iter__': {'node': 'litesearch.core.NoneType.__iter__',
   'flavor': 'method',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  '*.enumerate.__iter__': {'node': '*.enumerate.__iter__',
   'flavor': 'method',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.NoneType.__next__': {'node': 'litesearch.core.NoneType.__next__',
   'flavor': 'method',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  '*.enumerate.__next__': {'node': '*.enumerate.__next__',
   'flavor': 'method',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.vec_search.listcomp.0': {'node': 'litesearch.core.vec_search.listcomp.0',
   'flavor': 'scope',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'str.join': {'node': 'str.join',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.vec_search.^^^argument^^^.name': {'node': 'litesearch.core.vec_search.^^^argument^^^.name',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.vec_search.^^^argument^^^.db': {'node': 'litesearch.core.vec_search.^^^argument^^^.db',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.rrf_merge': {'node': 'litesearch.core.rrf_merge',
   'flavor': 'function',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.rrf_merge.lambda.0': {'node': 'litesearch.core.rrf_merge.lambda.0',
   'flavor': 'scope',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.database': {'node': 'litesearch.core.database',
   'flavor': 'function',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'fastlite.Database.enable_wal': {'node': 'fastlite.Database.enable_wal',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'usearch.sqlite_path': {'node': 'usearch.sqlite_path',
   'flavor': 'importeditem',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'fastlite.Database.conn': {'node': 'fastlite.Database.conn',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.search': {'node': 'litesearch.core.search',
   'flavor': 'function',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.search.^^^argument^^^.strip': {'node': 'litesearch.core.search.^^^argument^^^.strip',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.search.^^^argument^^^.quote_fts': {'node': 'litesearch.core.search.^^^argument^^^.quote_fts',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'},
  'litesearch.core.search.^^^argument^^^.q': {'node': 'litesearch.core.search.^^^argument^^^.q',
   'flavor': 'attribute',
   'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py'}})

---

CodeGraph.from_sources

def from_sources(
    sources:dict, chunk_size:int=50
):

Build from {module_name: source_str}.

cg.db['graph_edges'].insert_all([
    {'caller':'A','callee':'B','kind':'static','confidence':1.0},
    {'caller':'B','callee':'C','kind':'static','confidence':1.0},
    {'caller':'A','callee':'C','kind':'static','confidence':1.0},
])
pr = _pagerank(cg.db)
print(pr)
assert 'C' in pr, "C should be in pagerank"
assert pr['C'] > pr['A'], f"C (most linked-to) should rank higher than A: {pr}"
print("pagerank ok:", pr)

{'A': 0.05000000000000001, 'B': 0.07125000000000001, 'C': 0.13181250000000003}
pagerank ok: {'A': 0.05000000000000001, 'B': 0.07125000000000001, 'C': 0.13181250000000003}

CodeGraph(':memory:').from_sources({'mod': 'def foo(): pass\ndef bar(): foo()\n'}).db.t.graph_nodes()

Processing sources: 1chunk [00:00, 601.94chunk/s]

[{'node': 'mod.foo',
  'flavor': 'function',
  'file': 'mod',
  'pagerank': 0.06938,
  'in_degree': 1,
  'out_degree': 0},
 {'node': 'mod.foo.^^^argument^^^',
  'flavor': 'unspecified',
  'file': 'mod',
  'pagerank': 0.0375,
  'in_degree': 0,
  'out_degree': 0},
 {'node': 'mod.bar.^^^argument^^^',
  'flavor': 'unspecified',
  'file': 'mod',
  'pagerank': 0.0375,
  'in_degree': 0,
  'out_degree': 0},
 {'node': 'mod.bar',
  'flavor': 'function',
  'file': 'mod',
  'pagerank': 0.0375,
  'in_degree': 0,
  'out_degree': 1}]

_src = '''
import functools
from fastcore.all import bind

fasthtml_app = partial(create_app, debug=True)
wrapped     = functools.partial(handler, extra=1)
bound_fn    = bind(base_fn, 42)
plain       = lambda x: x + 1
not_a_fn    = {'key': 'value'}
'''
_nod = _lambda_nodes(sources={'mymod': _src})
assert 'mymod.fasthtml_app' in _nod, f'partial() not indexed: {list(_nod)}'
assert 'mymod.wrapped' in _nod, 'functools.partial() not indexed'
assert 'mymod.bound_fn' in _nod, 'bind() not indexed'
assert 'mymod.plain' in _nod, 'lambda still indexed'
assert 'mymod.not_a_fn' not in _nod, 'dict wrongly indexed'
assert all(v['flavor'] == 'function' for v in _nod.values())
print('_lambda_nodes partial/bind: all assertions pass')

_lambda_nodes partial/bind: all assertions pass

---

CodeGraph.from_pkg

def from_pkg(
    pkg:str, path:pathlib.Path | str='.', recursive:bool=True, maxdepth:int=None, symlinks:bool=True,
    file_glob:str=None, file_re:str=None, folder_re:str=None, skip_file_glob:str=None, skip_file_re:str=None,
    skip_folder_re:str=None, func:callable=walk_join, ret_folders:bool=False, sort:bool=True, types:str | list=None,
    exts:str | list=None
)->CodeGraph:

Build from an installed package. Uses pyan3 on the package source files.

---

CodeGraph.from_dir

def from_dir(
    dir:str | pathlib.Path, path:pathlib.Path | str='.', recursive:bool=True, maxdepth:int=None, symlinks:bool=True,
    file_glob:str=None, file_re:str=None, folder_re:str=None, skip_file_glob:str=None, skip_file_re:str=None,
    skip_folder_re:str=None, func:callable=walk_join, ret_folders:bool=False, sort:bool=True, types:str | list=None,
    exts:str | list=None
)->CodeGraph:

Build from .py files under path — pyan reads files directly.

---

CodeGraph.process_files

def process_files(
    files, # list of .py file paths to analyse
    root:NoneType=None, # package root; auto-detected via __init__.py walk if None
    sz:int=50, # max files per pyan3 pass; reduce for very large packages
)->CodeGraph:

Build call-graph edges for the given source files and recompute centrality.

_g = CodeGraph(':memory:')
_g.gn.insert({'node':'litesearch.core.search','flavor':'function','file':'','pagerank':0.,'in_degree':0,'out_degree':0})
_g.pa.insert({'caller':'mymod.env_context','method':'search','pkgs':'litesearch'})
_g.resolve_attr_calls()
_edges = list(_g.ge(where="kind='attr_call'"))
test_eq(len(_edges), 1)
test_eq(_edges[0]['caller'], 'mymod.env_context')
test_eq(_edges[0]['callee'], 'litesearch.core.search')
test_eq(_edges[0]['confidence'], 0.7)
# idempotent
_g.resolve_attr_calls()
test_eq(len(list(_g.ge(where="kind='attr_call'"))), 1)
_g.ge(where="kind='attr_call'")

[{'id': 1,
  'caller': 'mymod.env_context',
  'callee': 'litesearch.core.search',
  'kind': 'attr_call',
  'confidence': 0.7}]

_g2 = CodeGraph(':memory:')
_g2.gn.insert_all([{'node':'p1.mod.search','flavor':'function','file':'','pagerank':0.,'in_degree':0,'out_degree':0},
                   {'node':'p2.mod.search','flavor':'function','file':'','pagerank':0.,'in_degree':0,'out_degree':0}])
_g2.pa.insert({'caller':'mymod.f','method':'search','pkgs':'p1,p2'})
_g2.resolve_attr_calls()
test_eq(len(_g2.ge()), 1)

_g3 = CodeGraph(':memory:')
_g3.gn.insert({'node':'other.core.search','flavor':'function','file':'','pagerank':0.,'in_degree':0,'out_degree':0})
_g3.pa.insert({'caller':'mymod.f','method':'search','pkgs':'litesearch'})
_g3.resolve_attr_calls()
test_eq(len(list(_g3.ge())), 0)

cg=CodeGraph(':memory:')

cg.from_pkg('litesearch').from_pkg('fastlite').from_pkg('apswutils').from_pkg('fastcore')

Processing files: 1chunk [00:00,  9.38chunk/s]
Processing files: 1chunk [00:00, 16.90chunk/s]
Processing files: 1chunk [00:00,  4.88chunk/s]
Processing files: 1chunk [00:01,  1.08s/chunk]

<__main__.CodeGraph>

# eval: false
cg.from_pkg('protobuf')

Processing files: 2chunk [00:02,  1.30s/chunk]

<__main__.CodeGraph>

cg.gn(select='*', where='node like "%get_store%"')

[{'node': 'litesearch.core.get_store.str',
  'flavor': 'attribute',
  'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py',
  'pagerank': 0.00022,
  'in_degree': 0,
  'out_degree': 0},
 {'node': 'litesearch.core.get_store.^^^argument^^^',
  'flavor': 'unspecified',
  'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py',
  'pagerank': 0.00022,
  'in_degree': 0,
  'out_degree': 0},
 {'node': 'litesearch.core.get_store.bool',
  'flavor': 'attribute',
  'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py',
  'pagerank': 0.00022,
  'in_degree': 0,
  'out_degree': 0},
 {'node': 'litesearch.core.get_store',
  'flavor': 'function',
  'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py',
  'pagerank': 0.00028,
  'in_degree': 1,
  'out_degree': 3},
 {'node': 'litesearch.core.get_store.^^^argument^^^.t',
  'flavor': 'attribute',
  'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py',
  'pagerank': 0.00022,
  'in_degree': 0,
  'out_degree': 0}]

cg.from_dir('../kosha')

<__main__.CodeGraph>

cg.db.t.graph_nodes.count

cg.gn('node like "%emb_doc%"')

---

CodeGraph.node_info

def node_info(
    node:str, # fully-qualified node name
    with_kind:bool=False, # include edge kind in callers/callees tuples
)->dict:

Full node record: graph_nodes row merged with callers, callees, and co-dispatch peers.

---

CodeGraph.edge_kinds

def edge_kinds(
    
)->dict:

Count of each edge kind in the graph (static, decorator, patch, registration, etc.).

---

CodeGraph.short_paths

def short_paths(
    nodes, # iterable of fully-qualified node names
    k:int=10, # consider only the first k nodes
)->L:

Shortest paths between all pairs in top-k nodes — one BFS per source node.

---

CodeGraph.short_path

def short_path(
    src:str, # fully-qualified source node name
    tgt:str, # fully-qualified target node name
)->L:

Shortest path from src to tgt via BFS. Returns L of node names or empty L.

---

CodeGraph.ranked

def ranked(
    k:int=10, # number of top nodes to return
    module:str=None, # optional module prefix filter (e.g. 'fastcore.basics')
)->L:

Top-k nodes by pagerank, optionally filtered to a module prefix.

---

CodeGraph.co_dispatched

def co_dispatched(
    node:str, # fully-qualified node name
)->L:

Nodes registered in the same dispatch group as node (route list, handler table, etc.).

---

CodeGraph.neighbors

def neighbors(
    n:str, # fully-qualified node name
    depth:int=1, # number of hops to expand
)->L:

callers + callees up to depth hops.

---

CodeGraph.callees

def callees(
    n:str, # fully-qualified node name
    with_kind:bool=False, # if True, return (callee, edge_kind) tuples instead of plain names
)->L:

All nodes called by n, optionally with edge kind.

---

CodeGraph.callers

def callers(
    n:str, # fully-qualified node name (e.g. 'mymod.foo')
    with_kind:bool=False, # if True, return (caller, edge_kind) tuples instead of plain names
)->L:

All nodes that call n, optionally with edge kind.

cg.db.t.graph_edges(where='caller like "%apswutils.db.Table.insert_all%" and callee like "%insert_chunk%"')

[{'id': 4450,
  'caller': 'apswutils.db.Table.insert_all',
  'callee': 'apswutils.db.Table.insert_chunk',
  'kind': 'static',
  'confidence': 1.0}]

to_date_co = cg.co_dispatched('fastcore.basics.to_date')
assert all('fastcore' in n for n in to_date_co), \
    f"fastcore.basics.to_date co_dispatched leaked into httpx: {to_date_co}"
assert len(to_date_co) > 0, "fastcore.basics.to_date should have co_dispatched peers"
print("co_dispatched cross-package isolation ok:", to_date_co)

co_dispatched cross-package isolation ok: ['fastcore.basics.to_int', 'fastcore.basics.to_float', 'fastcore.basics.to_bool']

cg.node_info('fastcore.basics.to_date')

{'node': 'fastcore.basics.to_date',
 'flavor': 'function',
 'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fastcore/basics.py',
 'pagerank': 3e-05,
 'in_degree': 0,
 'out_degree': 2,
 'callers': [],
 'callees': ['fastcore.basics._typeerr', 'fastcore.basics.str2date'],
 'co_dispatched': ['fastcore.basics.to_int', 'fastcore.basics.to_float', 'fastcore.basics.to_bool']}

n = cg.db.t.graph_nodes(where='node like "%core.env_context"')[0]['node']
cg.node_info(n)

{'node': 'kosha.core.env_context',
 'flavor': 'function',
 'file': '../kosha/core.py',
 'pagerank': 0.006,
 'in_degree': 5,
 'out_degree': 9,
 'callers': ['kosha.cli.env_context', 'kosha.cli.lambda.2', 'kosha.graph.context', 'kosha.cli', 'kosha.core.Kosha'],
 'callees': ['fastcore.all.L', 'kosha.core.parseq', 'kosha.core.filt2wh', 'kosha.core.pkgs2consider', 'fastcore.all.patch', 'kosha.core.emb_query', 'kosha.core.Kosha', 'fastcore.all.true', 'litesearch.core.search'],
 'co_dispatched': []}

cg.short_path('kosha.graph.context', 'litesearch.core.search')

['kosha.graph.context', 'kosha.core.env_context', 'litesearch.core.search']

path = cg.short_path('apswutils.db.Table.upsert', 'apswutils.db.Table.insert_chunk')
assert path[0] == 'apswutils.db.Table.upsert', f"wrong start: {path}"
assert path[-1] == 'apswutils.db.Table.insert_chunk', f"wrong end: {path}"
assert len(path) >= 2, f"path too short: {path}"
assert cg.short_path('apswutils.db.Table.upsert', 'apswutils.db.Table.upsert') == L(['apswutils.db.Table.upsert']), "src==tgt should return [src]"
assert cg.short_path('apswutils.db.Table.upsert', 'no.such.node') == L(), "missing tgt should return empty"
print("short_path ok:", path)

short_path ok: ['apswutils.db.Table.upsert', 'apswutils.db.Table.upsert_all', 'apswutils.db.Table.insert_all', 'apswutils.db.Table.insert_chunk']

cg.ranked(5, module='litesearch')

[{'node': 'litesearch.core.vec_search', 'pagerank': 0.075}, {'node': 'litesearch.core.search', 'pagerank': 0.00832}, {'node': 'litesearch.core.search.^^^argument^^^.q', 'pagerank': 0.00628}, {'node': 'litesearch.data.kw', 'pagerank': 0.00105}, {'node': 'litesearch.utils.FastEncode.__init__', 'pagerank': 0.00071}]

cg.callers('litesearch.data.kw')

['litesearch.core.get_store', 'litesearch.core.database', 'litesearch.data.pkg2chunks', 'litesearch.data.pre', 'litesearch.utils.FastEncode.encode', 'litesearch.utils.FastEncode.encode.lambda.0', 'litesearch.utils.FastEncode.encode_document', 'litesearch.utils.FastEncode.encode_query', 'litesearch.utils.FastEncodeImage.embed', 'litesearch.utils.FastEncodeImage.embed.lambda.0', 'litesearch.utils.FastEncodeMultimodal.encode_text', 'litesearch.utils.FastEncodeMultimodal.encode_image', 'litesearch.utils.encode_pdf_texts']

---

CodeGraph.file2nodes

def file2nodes(
    path:str
)->set:

Return set of node names whose file column matches path.

g_drop = CodeGraph(':memory:')
g_drop.db.t.graph_nodes.insert_all([
    {'node':'mod.foo','flavor':'function','file':'/proj/mod.py'},
    {'node':'mod.bar','flavor':'function','file':'/proj/mod.py'},
])
g_drop.db.t.graph_edges.insert({'caller':'mod.foo','callee':'mod.bar','kind':'static','confidence':1.0})
assert g_drop.file2nodes('/proj/mod.py') == {'mod.foo','mod.bar'}
g_drop._drop_file('/proj/mod.py')
assert g_drop.file2nodes('/proj/mod.py') == set()
assert not list(g_drop.db.execute("SELECT * FROM graph_edges"))
print("_drop_file + _nodes_for_file ok")

_drop_file + _nodes_for_file ok

---

CodeGraph.sync

def sync(
    dir:NoneType=None, pkgs:NoneType=None, force:bool=False
)->CodeGraph:

Incremental sync: from_dir per dir, from_pkg per pkg; manage file_index. Centrality recomputed once at the end.

---

CodeGraph.sync_pkgs

def sync_pkgs(
    pkgs:list, in_parallel:bool=False
)->CodeGraph:

Incremental sync for packages: drop missing files, update changed files, add new files.

---

CodeGraph.sync_dir

def sync_dir(
    dir:str | pathlib.Path, force:bool=False
)->CodeGraph:

Incremental sync for a directory: drop missing files, update changed files, add new files.

import tempfile, time

with tempfile.TemporaryDirectory() as tmp:
    pa, pb = Path(f'{tmp}/a.py'), Path(f'{tmp}/b.py')
    pa.write_text('def foo(): pass\n'); pb.write_text('def bar(): pass\n')

    g = CodeGraph(':memory:')
    g.sync(dir=tmp)
    fi = {r['path']: r['last_analyzed_at'] for r in g.db.t.file_index.rows}
    assert str(pa) in fi and str(pb) in fi, fi

    # Modify b.py — mtime changes
    time.sleep(0.01)
    pb.write_text('def bar(): return 42\n')
    pb.touch()  # ensure mtime update
    g.sync(dir=tmp)
    fi2 = {r['path']: r['last_analyzed_at'] for r in g.db.t.file_index.rows}
    assert fi2[str(pa)] == fi[str(pa)], f"a.py should not be reprocessed: {fi2}"
    assert fi2[str(pb)] != fi[str(pb)], f"b.py should be reprocessed: {fi2}"

    # Delete a.py — should be removed from file_index
    pa.unlink()
    g.sync(dir=tmp)
    fi3 = {r['path'] for r in g.db.t.file_index.rows}
    assert str(pa) not in fi3, f"a.py should be removed: {fi3}"
    print("sync incremental ok, remaining files:", fi3)

    # add packages to codegraph
    g.sync(pkgs=['httpx', 'fastcore'])
    assert len(g.gn(where='node like "httpx.%"')) > 0, "httpx nodes should be in graph"
    assert len(g.gn(where='node like "fastcore.%"')) > 0, "fastcore nodes should be in graph"

sync incremental ok, remaining files: {'/var/folders/kg/9vdw4mdd1fs58svgh4k1qhr09x7dqh/T/tmpbwef768n/b.py'}

loading code graph for packages: 100%|██████████| 2/2 [00:02<00:00,  1.17s/pkg]

---

Kosha.short_paths

def short_paths(
    
):

Shortest paths helper: shortest paths between all pairs in top-k nodes.

---

Kosha.short_path

def short_path(
    
):

Shortest path helper: shortest path between two nodes.

---

Kosha.neighbors

def neighbors(
    
):

Neighbors helper: callers + callees up to depth hops.

---

Kosha.ni

def ni(
    
):

Node info helper: graph_nodes row + callers, callees, co-dispatched nodes.

---

Kosha.ge

def ge(
    
):

graph_edges table for direct queries.

---

Kosha.gn

def gn(
    
):

graph_nodes table for direct queries.

---

Kosha.graphdb

def graphdb(
    
):

Underlying graph.db database for direct queries.

---

Kosha.graph

def graph(
    
)->CodeGraph:

code graph for this kosha instance (memoized — opens graph.db once per Kosha instance).

---

Kosha.public_api

def public_api(
    pkg:str, # package or module to index (e.g. 'kosha', 'litesearch.data')
    meta_cols:str='mod_name,docstring', # metadata keys to extract
    limit:int=200, # max names to return
)->L:

Return public API: all public_api=1 entries + @patch-derived methods from the call graph from env and repo stores.

k=Kosha()
api = k.public_api('kosha')
names = {r['mod_name'] for r in api}
assert any('sync' in n for n in names), f"patched 'sync' missing from public_api: {names}"

api_httpx = k.public_api('httpx')
assert len(api_httpx) > 0, 'httpx __all__-based entries should still be returned'
print('public_api patch-aware ok | kosha:', len(api), '| httpx:', len(api_httpx))
print('kosha sample:', sorted(names)[:8])

public_api patch-aware ok | kosha: 112 | httpx: 200
kosha sample: ['kosha.core._is_pkg_ingested', 'kosha.core.awatch_repo', 'kosha.core.env_context', 'kosha.core.nuke', 'kosha.core.pkg_context', 'kosha.core.pkgs2consider', 'kosha.core.pkgs_in_env', 'kosha.core.process_env']

k.public_api('litesearch.data')[:3]

[{'mod_name': 'litesearch.data.pdf_texts', 'docstring': None}, {'mod_name': 'litesearch.data.pdf_links', 'docstring': None}, {'mod_name': 'litesearch.data.pdf_images', 'docstring': None}]

k.public_api('fastlite', meta_cols='mod_name')

[{'mod_name': 'fastlite.core.t'}, {'mod_name': 'fastlite.core.c'}, {'mod_name': 'fastlite.core.c'}, {'mod_name': 'fastlite.core.__str__'}, {'mod_name': 'fastlite.core.__str__'}, {'mod_name': 'fastlite.core.q'}, {'mod_name': 'fastlite.core.link_dcs'}, {'mod_name': 'fastlite.core.__call__'}, {'mod_name': 'fastlite.core.selectone'}, {'mod_name': 'fastlite.core.set_classes'}, {'mod_name': 'fastlite.core.get_tables'}, {'mod_name': 'fastlite.core.v'}, {'mod_name': 'fastlite.core.create'}, {'mod_name': 'fastlite.core.import_file'}, {'mod_name': 'fastlite.kw.xtra'}, {'mod_name': 'fastlite.kw.get_last'}, {'mod_name': 'fastlite.kw.ids_and_rows_where'}, {'mod_name': 'fastlite.kw.get'}, {'mod_name': 'fastlite.kw.__getitem__'}, {'mod_name': 'fastlite.kw.create'}, {'mod_name': 'fastlite.kw.transform'}, {'mod_name': 'fastlite.kw.transform_sql'}, {'mod_name': 'fastlite.kw.update'}, {'mod_name': 'fastlite.kw.insert_all'}, {'mod_name': 'fastlite.kw.insert'}, {'mod_name': 'fastlite.kw.upsert'}, {'mod_name': 'fastlite.kw.lookup'}, {'mod_name': 'fastlite.core.t'}, {'mod_name': 'fastlite.core.c'}, {'mod_name': 'fastlite.core.c'}, {'mod_name': 'fastlite.core.__str__'}, {'mod_name': 'fastlite.core.__str__'}, {'mod_name': 'fastlite.core.q'}, {'mod_name': 'fastlite.core.all_dcs'}, {'mod_name': 'fastlite.core.create_mod'}, {'mod_name': 'fastlite.core.link_dcs'}, {'mod_name': 'fastlite.core.__call__'}, {'mod_name': 'fastlite.core.selectone'}, {'mod_name': 'fastlite.core.set_classes'}, {'mod_name': 'fastlite.core.get_tables'}, {'mod_name': 'fastlite.core.v'}, {'mod_name': 'fastlite.core.get_typ'}, {'mod_name': 'fastlite.core.create'}, {'mod_name': 'fastlite.core.import_file'}, {'mod_name': 'fastlite.core.diagram'}, {'mod_name': 'fastlite.kw.MissingPrimaryKey'}, {'mod_name': 'fastlite.kw.database'}, {'mod_name': 'fastlite.kw.xtra'}, {'mod_name': 'fastlite.kw.get_last'}, {'mod_name': 'fastlite.kw.ids_and_rows_where'}, {'mod_name': 'fastlite.kw.get'}, {'mod_name': 'fastlite.kw.__getitem__'}, {'mod_name': 'fastlite.kw.create'}, {'mod_name': 'fastlite.kw.transform'}, {'mod_name': 'fastlite.kw.transform_sql'}, {'mod_name': 'fastlite.kw.update'}, {'mod_name': 'fastlite.kw.insert_all'}, {'mod_name': 'fastlite.kw.insert'}, {'mod_name': 'fastlite.kw.upsert'}, {'mod_name': 'fastlite.kw.lookup'}]

k.ni('fastlite.kw.database')

{'node': 'fastlite.kw.database',
 'flavor': 'function',
 'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fastlite/kw.py',
 'pagerank': 1e-05,
 'in_degree': 0,
 'out_degree': 12,
 'callers': [],
 'callees': ['apsw.Connection', 'apswutils.db.Database', 'typing.Any', 'apswutils.db.Database', 'typing.Any', 'apsw.Connection', 'typing.Any', 'apswutils.db.Database', 'apsw.Connection', 'apsw.Connection', 'apswutils.db.Database', 'typing.Any'],
 'co_dispatched': []}

---

Kosha.status

def status(
    pyproject:bool=True, depth:int=1
)->dict:

Index freshness: file/pkg/node counts and stale file count.

---

Kosha.context

def context(
    q:str, # query with optional key:value filters
    limit:int=50, repo:bool=True, env:bool=True, graph:bool=True,
    compact:bool=False, # return slim dicts (mod_name, signature, docstring, lineno) instead of full chunks
    columns:str='content,metadata', sys_wide:bool=True, kw:VAR_KEYWORD
)->L: # forwarded to env_context / repo_context

Fan-out semantic search: parse filters, run repo + env searches, merge with chained RRF.

---

Kosha.sync

def sync(
    pkgs:NoneType=None, # list of package names to sync (e.g. ['httpx', 'fastcore']); if None, sync all env packages
    dir:NoneType=None, # directory to sync; if None, sync all of root
    verbose:bool=True, # print progress messages
    in_parallel:bool=False, # run repo, env, and graph sync in parallel; also fans env packages out across threads
    force:bool=False, # ignore file mtimes and reprocess all files
    force_graph:bool=False, # whether to force graph recomputation even if no files changed; ignored if force=True
    pyproject:bool=True, # if True, auto-detect env packages from pyproject.toml; if False, use pkgs argument as-is
    depth:int=1, # depth for pyproject env package detection; ignored if pyproject=False
    embed:bool=True, # whether to embed
)->Kosha:

Sync code store, env store, and code graph. Runs in a daemon thread by default.

k = Kosha()

k.graph.sync(pkgs=env_pkg_versions(),force=True)

loading code graph for packages:  47%|████▋     | 9/19 [02:02<00:57,  5.77s/pkg]/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/pyskills/edit.py:1: SyntaxWarning: invalid escape sequence '\s'
  """Text editing tools for modifying files and ipynb notebook cells. Each operation — insert, replace, delete lines, and string(s) replacement — has both a `file_*` and `cell_*` variant. All return unified diffs showing what changed, or `"none: No changes."`, or `"error: ..."`.
/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/pyskills/edit.py:1: SyntaxWarning: invalid escape sequence '\s'
  """Text editing tools for modifying files and ipynb notebook cells. Each operation — insert, replace, delete lines, and string(s) replacement — has both a `file_*` and `cell_*` variant. All return unified diffs showing what changed, or `"none: No changes."`, or `"error: ..."`.
/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/pyskills/edit.py:1: SyntaxWarning: invalid escape sequence '\s'
  """Text editing tools for modifying files and ipynb notebook cells. Each operation — insert, replace, delete lines, and string(s) replacement — has both a `file_*` and `cell_*` variant. All return unified diffs showing what changed, or `"none: No changes."`, or `"error: ..."`.
/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/pyskills/edit.py:1: SyntaxWarning: invalid escape sequence '\s'
  """Text editing tools for modifying files and ipynb notebook cells. Each operation — insert, replace, delete lines, and string(s) replacement — has both a `file_*` and `cell_*` variant. All return unified diffs showing what changed, or `"none: No changes."`, or `"error: ..."`.
/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/pyskills/edit.py:1: SyntaxWarning: invalid escape sequence '\s'
  """Text editing tools for modifying files and ipynb notebook cells. Each operation — insert, replace, delete lines, and string(s) replacement — has both a `file_*` and `cell_*` variant. All return unified diffs showing what changed, or `"none: No changes."`, or `"error: ..."`.
/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/pyskills/edit.py:1: SyntaxWarning: invalid escape sequence '\s'
  """Text editing tools for modifying files and ipynb notebook cells. Each operation — insert, replace, delete lines, and string(s) replacement — has both a `file_*` and `cell_*` variant. All return unified diffs showing what changed, or `"none: No changes."`, or `"error: ..."`.
<unknown>:1: SyntaxWarning: invalid escape sequence '\s'
loading code graph for packages: 100%|██████████| 19/19 [03:11<00:00, 10.06s/pkg]

<__main__.CodeGraph>

k.graph._centrality()

<__main__.CodeGraph>

env_pkg_versions()

{'fastcore': '1.12.47',
 'pandas': '3.0.2',
 'watchfiles': '1.1.1',
 'litesearch': '0.0.29',
 'Jinja2': '3.1.6',
 'fastlite': '0.2.4',
 'pillow': '12.2.0',
 'tokenizers': '0.23.1',
 'model2vec': '0.8.1',
 'pyskills': '0.0.7',
 'pyan3': '2.6.0',
 'usearch': '2.25.2',
 'notebook': '7.5.6',
 'chonkie': '1.6.5',
 'yake': '0.7.3',
 'pdf_oxide': '0.3.45',
 'codesigs': '0.0.2',
 'onnxruntime': '1.26.0',
 'anyio': '4.13.0'}

k.gn(select='*', where='node like "%get_store"')

[{'node': 'litesearch.core.get_store',
  'flavor': 'function',
  'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py',
  'pagerank': 0.00028,
  'in_degree': 1,
  'out_degree': 27}]

k.status()

{'files': 3,
 'packages': 182,
 'graph_nodes': 85874,
 'stale_files': 0,
 'stale_pkgs': {}}

k.sync()

syncing files [Path('/Users/71293/code/personal/orgs/kosha/kosha/skill.py')] .....

parse files from /Users/71293/code/personal/orgs/kosha: 100%|██████████| 1/1 [00:00<00:00, 1225.33it/s]

No packages to load.

[None, None, <__main__.CodeGraph object>]

first(k.context('how do I merge dicts package:fastcore path:basics', limit=1, compact=True))

{'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fastcore/basics.py',
 'lineno': 655,
 'mod_name': 'fastcore.basics.merge',
 'docstring': 'Merge all dictionaries in `ds`',
 'sig': 'def merge(*ds):'}

---

Kosha.api_call_paths

def api_call_paths(
    from_pkg:str, # package whose public API is the call source
    to_pkg:str, # package whose public API is the call target
    k:int=15, # top-k API nodes per package to consider
)->dict:

Shortest call-graph paths from from_pkg public API nodes to to_pkg public API nodes.

---

Kosha.top_nodes

def top_nodes(
    pkg:str, k:int=5
)->L:

Top-k public API nodes for pkg ranked by pagerank in the code graph.

---

Kosha.dep_stack

def dep_stack(
    seeds:list=None, depth:int=1, no_sys_pkg:bool=True, allow:list=None
)->list:

BFS over pkg_deps, ordered by coupling strength.

---

is_sys_pkg

def is_sys_pkg(
    pkg:str
)->bool:

Return True if pkg is a system-wide package (e.g. in site-packages), False if likely user code.

rows = list(k.envdb.t.pkg_deps(where='from_pkg="fastlite"'))
assert rows, 'pkg_deps should have rows after update_pkg'
assert any(r['to_pkg'] in ('fastcore','fastlite') for r in rows), f'expected fastcore/fastlite: {rows[:3]}'
print('pkg_deps via update_pkg ok:', rows[:3])

pkg_deps via update_pkg ok: [{'from_pkg': 'fastlite', 'to_pkg': 'apsw', 'n_modules': 1}, {'from_pkg': 'fastlite', 'to_pkg': 'apswutils', 'n_modules': 2}, {'from_pkg': 'fastlite', 'to_pkg': 'dataclasses', 'n_modules': 2}]

layers = k.dep_stack(['fastlite'])
print(layers)
print('-------------------------------------------------')
stack = L(layers).flatten()
assert 'fastcore' in stack, f'fastcore should be in dep_stack: {stack}'
print('_dep_stack ok:', stack)

[['fastlite'], {'fastcore', 'apsw', 'apswutils'}]
-------------------------------------------------
_dep_stack ok: ['fastlite', 'fastcore', 'apsw', 'apswutils']

k.dep_stack(['fastlite'], allow=['fastcore'])

[['fastlite'], {'fastcore'}]

k.api_call_paths('fastlite', 'apswutils')

{}

k.context('payments page monsterui fasthtml', repo=False)[0]

{ '_dist': 0.6549316644668579,
  '_rrf_score': 0.016666666666666666,
  '_src_id': 'env:141770',
  'callees': [],
  'callers': [],
  'co_dispatched': [],
  'content': 'def FastHTML(*args, pico=False, **kwargs):\n'
             '    "Create a FastHTML app and adds `bg-background '
             'text-foreground` to bodykw for frankenui themes"\n'
             "    if 'bodykw' not in kwargs: kwargs['bodykw'] = {}\n"
             "    if 'class' not in kwargs['bodykw']: "
             "kwargs['bodykw']['class'] = ''\n"
             "    kwargs['bodykw']['class'] = "
             "stringify((kwargs['bodykw']['class'],'bg-background "
             "text-foreground'))\n"
             "    bodykw = kwargs.pop('bodykw',{})\n"
             '    return fh.FastHTML(*args, pico=pico, **bodykw, **kwargs)',
  'metadata': { 'docstring': 'Create a FastHTML app and adds `bg-background '
                             'text-foreground` to bodykw for frankenui themes',
                'end_lineno': 33,
                'lang': '.py',
                'lineno': 27,
                'mod_name': 'monsterui.core.FastHTML',
                'name': 'FastHTML',
                'package': 'monsterui',
                'path': '/Users/71293/code/personal/orgs/vedicreader/.venv/lib/python3.13/site-packages/monsterui/core.py',
                'public_api': True,
                'type': 'FunctionDef',
                'uploaded_at': 1774495368.930001,
                'version': '1.0.45'},
  'rowid': 141770}

k.context('payments page packages:monsterui,python-fasthtml', repo=False)[:10]

[{'rowid': 49813, 'content': 'def HTMX(path="/", host=\'localhost\', app=None, port=8000, height="auto", link=False, iframe=True, client=None):\n    "An iframe which displays the HTMX application in a notebook."\n    if isinstance(height, int): height = f"{height}px"\n    scr = """{\n        let frame = this;\n        window.addEventListener(\'message\', function(e) {\n            if (e.source !== frame.contentWindow) return; // Only proceed if the message is from this iframe\n            if (e.data.height) frame.style.height = (e.data.height+1) + \'px\';\n        }, false);\n    }""" if height == "auto" else ""\n    proto = \'http\' if host==\'localhost\' else \'https\'\n    fullpath = f"{proto}://{host}:{port}{path}" if host else path\n    src = f\'src="{fullpath}"\'\n    if link: display(HTML(f\'<a href="{fullpath}" target="_blank">Open in new tab</a>\'))\n    if isinstance(path, (FT,tuple,Safe)):\n        assert app, \'Need an app to render a component\'\n        route = f\'/{unqid()}\'\n        res = path\n        app.get(route)(lambda: res)\n        cli = client if client is not None else TestClient(app)\n        page = cli.get(route).text\n        src = f\'srcdoc="{escape(page)}"\'\n    if iframe:\n        return HTML(f\'<iframe {src} style="width: 100%; height: {height}; border: none;" onload="{scr}" \' + """allow="accelerometer; autoplay; camera; clipboard-read; clipboard-write; display-capture; encrypted-media; fullscreen; gamepad; geolocation; gyroscope; hid; identity-credentials-get; idle-detection; magnetometer; microphone; midi; payment; picture-in-picture; publickey-credentials-get; screen-wake-lock; serial; usb; web-share; xr-spatial-tracking"></iframe> """)', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/jupyter.py', 'uploaded_at': 1775816437.484823, 'name': 'HTMX', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 121, 'end_lineno': 144, 'package': 'python-fasthtml', 'version': '0.13.3', 'mod_name': 'fasthtml.jupyter.HTMX'}, 'rank': -11.48379379859935, '_rrf_score': 0.016666666666666666, '_src_id': 'env:49813', 'node': 'fasthtml.jupyter.HTMX', 'flavor': 'function', 'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/jupyter.py', 'pagerank': 1e-05, 'in_degree': 1, 'out_degree': 10, 'callers': ['fasthtml.jupyter.ws_client'], 'callees': ['fasthtml.xtend.with_sid.get', 'IPython.display.HTML', 'fasthtml.xtend.robots_txt.get', 'fasthtml.xtend.sitemap_xml.get', 'fasthtml.core.static_route.get', 'fasthtml.core.static_route_exts.get', 'html.escape', 'starlette.testclient.TestClient', 'IPython.display.display', 'fasthtml.core.unqid'], 'co_dispatched': []}, {'rowid': 79857, 'content': 'def account_management(sess):\n    user_email = sess[\'auth\']\n    user_payments = payments("email=?", (user_email,))\n    # Create table rows for each payment\n    payment_rows = []\n    for payment in user_payments:\n        action_button = ""\n        if payment.payment_status == \'paid\':\n            action_button = Button("Request Refund", hx_post=f"/refund?checkout_sid={payment.checkout_session_id}",hx_target="#refund-status")\n        elif payment.payment_status == \'refunded\': action_button = "Refunded"\n        \n        # Add row to table\n        payment_rows.append(\n            Tr(*map(Td, (payment.created_at, payment.amount, payment.payment_status, action_button))))\n    \n    # Create payment history table\n    payment_history = Table(\n        Thead(Tr(*map(Th, ("Date", "Amount", "Status", "Action")))),\n        Tbody(*payment_rows))\n    \n    return Titled(\n        "Account Management",\n        Div(H2(f"Account: {user_email}"),\n            H3("Payment History"),\n            payment_history,\n            Div(id="refund-status"),  # Target for refund status messages\n            A("Back to Home", href="/")))', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'uploaded_at': 1777248395.7674882, 'name': 'account_management', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 183, 'end_lineno': 209, 'mod_name': 'fasthtml.stripe_otp.account_management', 'package': 'python-fasthtml', 'version': '0.13.4', 'docstring': {}, 'public_api': True}, '_dist': 0.5673940181732178, '_rrf_score': 0.016666666666666666, '_src_id': 'env:79857', 'node': 'fasthtml.stripe_otp.account_management', 'flavor': 'function', 'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'pagerank': 1e-05, 'in_degree': 0, 'out_degree': 4, 'callers': [], 'callees': ['fasthtml.xtend.A', 'fasthtml.fastapp.fast_app', 'fasthtml.svg.PathFT.A', 'fasthtml.xtend.Titled'], 'co_dispatched': []}, {'rowid': 79788, 'content': 'def HTMX(path="/", host=\'localhost\', app=None, port=8000, height="auto", link=False, iframe=True, client=None):\n    "An iframe which displays the HTMX application in a notebook."\n    if isinstance(height, int): height = f"{height}px"\n    scr = """{\n        let frame = this;\n        window.addEventListener(\'message\', function(e) {\n            if (e.source !== frame.contentWindow) return; // Only proceed if the message is from this iframe\n            if (e.data.height) frame.style.height = (e.data.height+1) + \'px\';\n        }, false);\n    }""" if height == "auto" else ""\n    proto = \'http\' if host==\'localhost\' else \'https\'\n    fullpath = f"{proto}://{host}:{port}{path}" if host else path\n    src = f\'src="{fullpath}"\'\n    if link: display(HTML(f\'<a href="{fullpath}" target="_blank">Open in new tab</a>\'))\n    if isinstance(path, (FT,tuple,Safe)):\n        assert app, \'Need an app to render a component\'\n        route = f\'/{unqid()}\'\n        res = path\n        app.get(route)(lambda: res)\n        cli = client if client is not None else TestClient(app)\n        page = cli.get(route).text\n        src = f\'srcdoc="{escape(page)}"\'\n    if iframe:\n        return HTML(f\'<iframe {src} style="width: 100%; height: {height}; border: none;" onload="{scr}" \' + """allow="accelerometer; autoplay; camera; clipboard-read; clipboard-write; display-capture; encrypted-media; fullscreen; gamepad; geolocation; gyroscope; hid; identity-credentials-get; idle-detection; magnetometer; microphone; midi; payment; picture-in-picture; publickey-credentials-get; screen-wake-lock; serial; usb; web-share; xr-spatial-tracking"></iframe> """)', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/jupyter.py', 'uploaded_at': 1777248395.759724, 'name': 'HTMX', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 121, 'end_lineno': 144, 'mod_name': 'fasthtml.jupyter.HTMX', 'package': 'python-fasthtml', 'version': '0.13.4', 'docstring': 'An iframe which displays the HTMX application in a notebook.', 'public_api': True}, 'rank': -11.163207804625452, '_rrf_score': 0.01639344262295082, '_src_id': 'env:79788', 'node': 'fasthtml.jupyter.HTMX', 'flavor': 'function', 'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/jupyter.py', 'pagerank': 1e-05, 'in_degree': 1, 'out_degree': 10, 'callers': ['fasthtml.jupyter.ws_client'], 'callees': ['fasthtml.xtend.with_sid.get', 'IPython.display.HTML', 'fasthtml.xtend.robots_txt.get', 'fasthtml.xtend.sitemap_xml.get', 'fasthtml.core.static_route.get', 'fasthtml.core.static_route_exts.get', 'html.escape', 'starlette.testclient.TestClient', 'IPython.display.display', 'fasthtml.core.unqid'], 'co_dispatched': []}, {'rowid': 49862, 'content': 'def account_management(sess):\n    user_email = sess[\'auth\']\n    user_payments = payments("email=?", (user_email,))\n    # Create table rows for each payment\n    payment_rows = []\n    for payment in user_payments:\n        action_button = ""\n        if payment.payment_status == \'paid\':\n            action_button = Button("Request Refund", hx_post=f"/refund?checkout_sid={payment.checkout_session_id}",hx_target="#refund-status")\n        elif payment.payment_status == \'refunded\': action_button = "Refunded"\n        \n        # Add row to table\n        payment_rows.append(\n            Tr(*map(Td, (payment.created_at, payment.amount, payment.payment_status, action_button))))\n    \n    # Create payment history table\n    payment_history = Table(\n        Thead(Tr(*map(Th, ("Date", "Amount", "Status", "Action")))),\n        Tbody(*payment_rows))\n    \n    return Titled(\n        "Account Management",\n        Div(H2(f"Account: {user_email}"),\n            H3("Payment History"),\n            payment_history,\n            Div(id="refund-status"),  # Target for refund status messages\n            A("Back to Home", href="/")))', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'uploaded_at': 1775816437.4866984, 'name': 'account_management', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 183, 'end_lineno': 209, 'package': 'python-fasthtml', 'version': '0.13.3', 'mod_name': 'fasthtml.stripe_otp.account_management'}, '_dist': 0.5864739418029785, '_rrf_score': 0.01639344262295082, '_src_id': 'env:49862', 'node': 'fasthtml.stripe_otp.account_management', 'flavor': 'function', 'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'pagerank': 1e-05, 'in_degree': 0, 'out_degree': 4, 'callers': [], 'callees': ['fasthtml.xtend.A', 'fasthtml.fastapp.fast_app', 'fasthtml.svg.PathFT.A', 'fasthtml.xtend.Titled'], 'co_dispatched': []}, {'rowid': 79855, 'content': 'def cancel():\n    return Titled(\n        "Cancelled",\n        Div(H2("Payment Cancelled"),\n            P("Your payment was cancelled."),\n            A("Back to Home", href="/")))', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'uploaded_at': 1777248395.7674882, 'name': 'cancel', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 149, 'end_lineno': 154, 'mod_name': 'fasthtml.stripe_otp.cancel', 'package': 'python-fasthtml', 'version': '0.13.4', 'docstring': {}, 'public_api': True}, '_dist': 0.612480878829956, '_rrf_score': 0.016129032258064516, '_src_id': 'env:79855', 'node': 'fasthtml.stripe_otp.cancel', 'flavor': 'function', 'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'pagerank': 1e-05, 'in_degree': 0, 'out_degree': 4, 'callers': [], 'callees': ['fasthtml.svg.PathFT.A', 'fasthtml.fastapp.fast_app', 'fasthtml.xtend.Titled', 'fasthtml.xtend.A'], 'co_dispatched': []}, {'rowid': 49860, 'content': 'def cancel():\n    return Titled(\n        "Cancelled",\n        Div(H2("Payment Cancelled"),\n            P("Your payment was cancelled."),\n            A("Back to Home", href="/")))', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'uploaded_at': 1775816437.4866984, 'name': 'cancel', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 149, 'end_lineno': 154, 'package': 'python-fasthtml', 'version': '0.13.3', 'mod_name': 'fasthtml.stripe_otp.cancel'}, '_dist': 0.6130797863006592, '_rrf_score': 0.015873015873015872, '_src_id': 'env:49860', 'node': 'fasthtml.stripe_otp.cancel', 'flavor': 'function', 'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'pagerank': 1e-05, 'in_degree': 0, 'out_degree': 4, 'callers': [], 'callees': ['fasthtml.svg.PathFT.A', 'fasthtml.fastapp.fast_app', 'fasthtml.xtend.Titled', 'fasthtml.xtend.A'], 'co_dispatched': []}, {'rowid': 79854, 'content': 'def success(sess, checkout_sid:str):    \n    # Get payment record from database (saved in the webhook)\n    payment = payments[checkout_sid]\n\n    if not payment or payment.payment_status != \'paid\': \n        return Titled("Error", P("Payment not found"))\n\n    return Titled(\n        "Success",\n        Div(H2("Payment Successful!"),\n            P(f"Thank you for your purchase, {sess[\'auth\']}"),\n            P(f"Amount Paid: ${payment.amount / 100:.2f}"),\n            P(f"Status: {payment.payment_status}"),\n            P(f"Transaction ID: {payment.checkout_session_id}"),\n            A("Back to Home", href="/")))', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'uploaded_at': 1777248395.7674882, 'name': 'success', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 131, 'end_lineno': 145, 'mod_name': 'fasthtml.stripe_otp.success', 'package': 'python-fasthtml', 'version': '0.13.4', 'docstring': {}, 'public_api': True}, '_dist': 0.6225478053092957, '_rrf_score': 0.015625, '_src_id': 'env:79854', 'node': 'fasthtml.stripe_otp.success', 'flavor': 'function', 'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'pagerank': 1e-05, 'in_degree': 0, 'out_degree': 4, 'callers': [], 'callees': ['fasthtml.fastapp.fast_app', 'fasthtml.svg.PathFT.A', 'fasthtml.xtend.Titled', 'fasthtml.xtend.A'], 'co_dispatched': []}, {'rowid': 49859, 'content': 'def success(sess, checkout_sid:str):    \n    # Get payment record from database (saved in the webhook)\n    payment = payments[checkout_sid]\n\n    if not payment or payment.payment_status != \'paid\': \n        return Titled("Error", P("Payment not found"))\n\n    return Titled(\n        "Success",\n        Div(H2("Payment Successful!"),\n            P(f"Thank you for your purchase, {sess[\'auth\']}"),\n            P(f"Amount Paid: ${payment.amount / 100:.2f}"),\n            P(f"Status: {payment.payment_status}"),\n            P(f"Transaction ID: {payment.checkout_session_id}"),\n            A("Back to Home", href="/")))', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'uploaded_at': 1775816437.4866984, 'name': 'success', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 131, 'end_lineno': 145, 'package': 'python-fasthtml', 'version': '0.13.3', 'mod_name': 'fasthtml.stripe_otp.success'}, '_dist': 0.6284384727478027, '_rrf_score': 0.015384615384615385, '_src_id': 'env:49859', 'node': 'fasthtml.stripe_otp.success', 'flavor': 'function', 'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'pagerank': 1e-05, 'in_degree': 0, 'out_degree': 4, 'callers': [], 'callees': ['fasthtml.fastapp.fast_app', 'fasthtml.svg.PathFT.A', 'fasthtml.xtend.Titled', 'fasthtml.xtend.A'], 'co_dispatched': []}, {'rowid': 49855, 'content': 'def home(sess):\n    auth = sess[\'auth\']\n    return Titled(\n        "Buy Now", \n        Div(H2("Demo Product - $19.99"),\n            P(f"Welcome, {auth}"),\n            Button("Buy Now", hx_post="/create-checkout-session", hx_swap="none"),\n            A("View Account", href="/account")))', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'uploaded_at': 1775816437.4866984, 'name': 'home', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 57, 'end_lineno': 64, 'package': 'python-fasthtml', 'version': '0.13.3', 'mod_name': 'fasthtml.stripe_otp.home'}, '_dist': 0.6632375121116638, '_rrf_score': 0.015151515151515152, '_src_id': 'env:49855', 'node': 'fasthtml.stripe_otp.home', 'flavor': 'function', 'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'pagerank': 1e-05, 'in_degree': 0, 'out_degree': 4, 'callers': [], 'callees': ['fasthtml.svg.PathFT.A', 'fasthtml.fastapp.fast_app', 'fasthtml.xtend.Titled', 'fasthtml.xtend.A'], 'co_dispatched': []}, {'rowid': 79848, 'content': 'def home(sess):\n    auth = sess[\'auth\']\n    return Titled(\n        "Buy Now", \n        Div(H2("Demo Product - $19.99"),\n            P(f"Welcome, {auth}"),\n            Button("Buy Now", hx_post="/create-checkout-session", hx_swap="none"),\n            A("View Account", href="/account")))', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'uploaded_at': 1777248395.7674882, 'name': 'home', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 57, 'end_lineno': 64, 'mod_name': 'fasthtml.stripe_otp.home', 'package': 'python-fasthtml', 'version': '0.13.4', 'docstring': {}, 'public_api': True}, '_dist': 0.6656088829040527, '_rrf_score': 0.014925373134328358, '_src_id': 'env:79848', 'node': 'fasthtml.stripe_otp.home', 'flavor': 'function', 'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py', 'pagerank': 1e-05, 'in_degree': 0, 'out_degree': 4, 'callers': [], 'callees': ['fasthtml.svg.PathFT.A', 'fasthtml.fastapp.fast_app', 'fasthtml.xtend.Titled', 'fasthtml.xtend.A'], 'co_dispatched': []}]

---

Kosha.where_to_add

def where_to_add(
    description:str, limit:int=5
)->L:

Likely insertion points for new code: file + line from top context results + co_dispatched peers.

pts = k.where_to_add('add a new CLI command', limit=3)
assert isinstance(pts, L)
if pts:
    assert all(k in pts[0] for k in ('node','path','insert_after','co_dispatched','pagerank'))
    print('where_to_add ok:', pts[0]['path'], 'line', pts[0]['insert_after'])
else: print('where_to_add ok: no results (index may be empty)')

where_to_add ok: /Users/71293/code/personal/orgs/kosha/kosha/cli.py line 251