graph

A call graph of your code and its packages, for seeing what a change will touch.

Edges come from two passes over the source. The static pass uses ast-grep to find every place a function, class or method is named inside another and binds that name to its definition. The dynamic pass uses Python’s ast module to find dependencies that are not written as calls, such as decorators and handler tables.

CodeGraph holds the graph in one SQLite database. Pass a file path, or ':memory:' for a throwaway graph. It has six tables.

Kosha.graph opens the CodeGraph for your repository. You rarely build one yourself. The examples below use an empty in-memory graph.

CodeGraph

def CodeGraph(
    path:str | pathlib.Path=':memory:', busy_timeout:int=None, # ms to wait for a locked database
):

A call graph of Python code, stored in SQLite, with methods to build and query it.

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: {'co_dispatch', 'graph_edges', 'pending_attr_calls', 'graph_meta', 'graph_nodes', 'file_index'}
# no failed DDL logged on construction (old dbs still migrate)
import logging as _lg
_recs=[]
class _Cap(_lg.Handler):
    def emit(self,r): _recs.append(r.getMessage())
_l=_lg.getLogger('apsw'); _l.addHandler(_Cap())
_g = CodeGraph(':memory:')
assert not [m for m in _recs if 'duplicate column' in m], _recs

# legacy db missing the newer columns gets migrated
from tempfile import TemporaryDirectory
with TemporaryDirectory() as _d:
    _p = f'{_d}/legacy.db'
    _db = database(_p)
    _db.t.graph_nodes.create(node=str, flavor=str, file=str, pagerank=float, in_degree=int, out_degree=int, pk='node')
    _db.t.file_index.create(path=str, root=str, last_analyzed_at=float, pk='path')
    _db.conn.close()
    _lg2 = CodeGraph(_p)
    assert 'method' in _lg2.gn.columns_dict
    assert 'source_hash' in _lg2.fi.columns_dict
    _lg2.db.conn.close()
_l.removeHandler(_l.handlers[-1])
print('migration ok')
migration ok

Dynamic edges

Some dependencies are never written as calls. A web framework calls hello because @app.route('/hello') registered it. The fastcore @patch decorator adds a plain function to a class as a method. dyn_edges reads the source of one module and returns these dependencies as edges. It finds six kinds:

  • decorator: @app.route(...) above def hello gives an edge from app to hello. The decorator must be an attribute of a name, such as app.route. Confidence is 0.95 when app comes from a known web framework: fasthtml, flask, fastapi, starlette, django, aiohttp, sanic, tornado, litestar or blacksheep. Otherwise confidence is 0.75 when the method name is a registration verb, and 0.5 for any other name. The registration verbs are add, register, include, on, connect, mount, sub, subscribe, listen, use, route, handle, hook, attach, bind, plug, get, post, put, delete, patch, head, options, before, after and middleware. A verb can have a suffix, as in add_route. Case does not matter.
  • registration: a call such as app.register(hello) gives an edge from app to hello. The method name must be a registration verb, and hello must be a function defined in the module. Confidence follows the same rule as decorator.
  • connect_body: a function named connect, setup, register, init_app, create_app, mount, configure, bootstrap, install or activate gives edges when its body passes a plain name to a method of its first parameter. Each edge runs from the function to the passed name. Confidence is 0.9.
  • patch: @patch def f(self:Table) gives an edge from Table to f. An annotation such as Table|View gives one edge for each class. Confidence is 0.9.
  • delegates: @delegates(target) on wrapper gives an edge from wrapper to target. Confidence is 0.85.
  • co_dispatch: a module-level list, tuple or dict of the module’s functions, such as HANDLERS = [a, b, c], makes those functions peers. For a dict, the values are the functions. Confidence is 0.8. The peers of a handler show the pattern to follow when you add a new handler. CodeGraph stores peers as groups in the co_dispatch table, not as edges.

The examples below show each kind.

dyn_edges

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

Edges from decorators, registration calls, setup functions, @patch, @delegates and handler tables in src, with nodes named under module.

A call like self.db.search(q) names a method, search, on an object whose type the source does not state. Kosha can resolve the call only when it knows which search methods exist. While it reads a module, Kosha records one hint for each such call inside an indexed function. A hint holds the calling function, the method name and the packages the module imports, in import order. resolve_attr_calls turns the hints into edges later. A module that imports nothing gives hints that never resolve.

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)
# a function missing from `meta` gives no hint
test_eq(_attr_call_hints(sources={'m': _src}, meta={}), [])
# `self.method()` and calls on other objects give no hint
_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'})

In this Flask snippet, @app.route produces a decorator edge from mymod.app to mymod.hello. The call other_app.register(hello) produces a registration edge from mymod.other_app to mymod.hello. A function with at least one parameter and one of these names also produces connect_body edges: connect, setup, register, init_app, create_app, mount, configure, bootstrap, install, activate. These edges run from the function to each plain name that the function passes to a method of its first parameter. Here setup produces a connect_body edge from mymod.setup to mymod.hello. A registration edge needs the argument to be a function defined in the same module. Source that does not parse produces no edges. The two cells after this one show the delegates and patch kinds.

_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
'''
_es = dyn_edges(_DELEGATES_SRC, 'mymod')
test_eq({e['callee'] for e in _es}, {'mymod.target'})
_es
[{'caller': 'mymod.wrapper_pos',
  'callee': 'mymod.target',
  'kind': 'delegates',
  'confidence': 0.85},
 {'caller': 'mymod.wrapper_kw',
  'callee': 'mymod.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}]

resolve_attr_calls turns the pending hints into attr_call edges with confidence 0.7. For each hint it finds the graph nodes whose last name is the method and whose package the calling module imports. It picks the node from the package that the module imports last. It adds no edge when the caller already has an edge to that node. from_sources, from_dir, from_pkg and sync call it at the end of a build. The tests further down call it on hints inserted by hand.

CodeGraph.resolve_attr_calls

def resolve_attr_calls()->CodeGraph:

Turn pending self.x.method() hints into attr_call edges, now that the graph knows which methods exist.

Static edges

static_edges reads source with ast-grep. It returns a list of edges and a dict of node metadata keyed by node name. Every edge has kind static and confidence 1.0. The metadata holds the flavor (function, class or method), the file and the short method name. Pass either sources, a dict of {module name: source text}, or filenames with the root directory that module names are relative to. Without root, static_edges uses the parent of the first file’s folder. Passing both sources and filenames, or neither, raises AssertionError.

An edge means that the caller uses the callee by name. A call helper() counts. Passing helper as a value also counts. Kosha looks for the definition of a name in three places, in this order:

  1. A definition in an enclosing scope: the function, its class or the module.
  2. The module’s imports. The edge points to the imported name, even when the corpus does not define it.
  3. The only definition with that name in the whole corpus.

Kosha makes no edge when none of the three applies. Three further rules limit the edges. Kosha reads an attribute only when it is reached through self, and it resolves self.m in step 1 only. An attribute on any other object, such as x.get, makes no edge, because x.get() could call the get of any object. Step 3 never applies to a name that is a Python builtin, such as list. A name that appears only in an import statement makes no edge.

The first example graphs two lines of source. A file that does not parse gives no edges and no nodes.

static_edges

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

Call edges and node metadata for sources ({module: source}) or for filenames under root, read with ast-grep.

edges, meta = static_edges({"mod": "def foo(): pass\ndef bar(): foo()"})
meta['mod.foo']
{'node': 'mod.foo', 'flavor': 'function', 'file': '', 'method': 'foo'}
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 (: !!!"})
([], {})
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']
# resolution rules: enclosing scope, then this module's imports, then a unique global
_edges, _meta = static_edges({
    'app':  'from lib import helper\ndef run(): helper()\ndef also(): run()',
    'lib':  'def helper(): pass',
    'noise':'class C:\n\tdef get(self): pass'})
_es = {(e['caller'], e['callee']) for e in _edges}
test_eq(('app.run','lib.helper') in _es, True)     # through the import map
test_eq(('app.also','app.run') in _es, True)       # through the module scope
print(sorted(_es))
[('app.also', 'app.run'), ('app.run', 'lib.helper')]
# an attribute call must not bind by short name alone, and a builtin must not bind at all
_edges, _ = static_edges({
    'user': 'def f(x):\n\ty = list()\n\treturn x.get()',
    'lib':  'class Store:\n\tdef get(self): pass\ndef list(): pass'})
_cs = {e['callee'] for e in _edges}
assert 'lib.Store.get' not in _cs, f'x.get() bound to an unrelated method: {_cs}'
assert 'lib.list' not in _cs, f'the builtin list() bound to a corpus def: {_cs}'
print('conservative resolution ok:', _cs)
conservative resolution ok: set()
# an import statement binds a name. It is not a use of the name and makes no edge.
_edges, _ = static_edges({'a': 'def helper(): pass', 'b': 'from a import helper'})
test_eq(_edges, [])

# `self.m()` resolves inside the class, and nowhere else
_edges, _ = static_edges({'m': 'class C:\n\tdef m(self): pass\n\tdef n(self): self.m()',
                          'other': 'class D:\n\tdef m(self): pass'})
test_eq({(e['caller'],e['callee']) for e in _edges}, {('m.C.n','m.C.m')})

Building a graph

Every node has three numbers. in_degree is the number of edges into the node, which is its number of callers. out_degree is the number of edges out of the node, which is its number of callees. pagerank is the score from the PageRank algorithm, which Google first used to rank web pages. A node has a high score when many nodes depend on it, or when heavily used nodes depend on it. Kosha uses the score as a measure of reach. A change to a node with a high PageRank affects more code.

Kosha refreshes degrees after every update. PageRank is slow on a large graph. Kosha recomputes it when no PageRank has run yet, or when the number of edges has changed by more than 10% since the last run.

from_sources builds a graph from a dict of {module name: source text}. Use it to try the graph on a snippet. It also records each module-level name bound to a lambda, or to a partial, bind or curry call, as a function node.

CodeGraph.from_sources

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

Build the graph from a dict of {module name: source text}.

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.1975795937322862, 'B': 0.2815511087403979, 'C': 0.5208692975273158}
pagerank ok: {'A': 0.1975795937322862, 'B': 0.2815511087403979, 'C': 0.5208692975273158}
# PageRank always scores the complete graph.
_g = CodeGraph(':memory:')
_g.ge.insert_all([{'caller':f'n{i}','callee':'hub','kind':'static','confidence':1.0} for i in range(50)] +
                 [{'caller':'hub','callee':'leaf','kind':'static','confidence':1.0}])
_g._centrality({'leaf'})            # one changed node
_rows = {r['node']: r for r in _g.gn()}
test_eq(_rows['hub']['in_degree'], 50)
assert _rows['hub']['pagerank'] > _rows['n0']['pagerank'], _rows
test_eq(len(_rows), 52)             # every node scored, not just the one passed in

# Degrees update on every call. PageRank updates only past its threshold, or when forced.
_before = _rows['hub']['pagerank']
_g.ge.insert({'caller':'n0','callee':'leaf','kind':'static','confidence':1.0})
_g._centrality()
test_eq({r['node']:r['pagerank'] for r in _g.gn()}['hub'], _before)
test_eq({r['node']:r['in_degree'] for r in _g.gn()}['leaf'], 2)   # degrees refresh regardless
_g._centrality(force=True)
assert {r['node']:r['pagerank'] for r in _g.gn()}['hub'] != _before

# PageRank values sum to 1 when a node has no outgoing edge.
_pg = CodeGraph(':memory:')
_pg.ge.insert({'caller':'a','callee':'b','kind':'static','confidence':1.0})
_pr = _pagerank(_pg.db)
assert np.isclose(sum(_pr.values()), 1.0), _pr
assert _pr['b'] > _pr['a'], _pr
# A node that no edge names also gets a new PageRank value. The stored value is replaced.
_g = CodeGraph(':memory:')
_g.ge.insert_all([{'caller':'a','callee':'b','kind':'static','confidence':1.0}])
_g.gn.insert_all([{'node':'lonely','flavor':'function','file':'x.py','pagerank':0.5}], upsert=True, pk='node')
_g._centrality(force=True)
_rows = {r['node']: r for r in _g.gn()}
assert _rows['lonely']['pagerank'] < _rows['b']['pagerank'], _rows
assert abs(sum(r['pagerank'] for r in _rows.values()) - 1) < 0.01, _rows
# On a graph of 200,000 edges, no PageRank value rounds to 0.
_g3 = CodeGraph(':memory:')
_g3.ge.insert_all([{'caller':f'n{i}','callee':f'n{i+1}','kind':'static','confidence':1.0} for i in range(200_000)])
_g3._centrality(force=True)
_pr3 = [r['pagerank'] for r in _g3.gn(select='pagerank')]
test_eq(len([v for v in _pr3 if v == 0]), 0)
assert abs(sum(_pr3) - 1) < 0.01, sum(_pr3)
# Bridging pairs form one co-dispatch group.
test_eq(_uf_groups([('a','b'),('c','d'),('b','c')]), [{'a','b','c','d'}])
_g = CodeGraph(':memory:')
_g._add_dyn({'m': 'def a(): pass\ndef b(): pass\ndef c(): pass\nX = [a, b]\nY = (b, c)\n'})
_grp = defaultdict(set)
for r in _g.cd(): _grp[r['group_id']].add(r['node'])
test_eq(list(_grp.values()), [{'m.a','m.b','m.c'}])

# New batches merge with stored groups.
_g = CodeGraph(':memory:')
_g._add_dyn({'m': 'def a(): pass\ndef b(): pass\nX = [a, b]\n'})
_g._add_dyn({'m': 'def b(): pass\ndef c(): pass\nY = [b, c]\n'})
_grp = defaultdict(set)
for r in _g.cd(): _grp[r['group_id']].add(r['node'])
test_eq(list(_grp.values()), [{'m.a','m.b','m.c'}])
CodeGraph(':memory:').from_sources({'mod': 'def foo(): pass\ndef bar(): foo()\n'}).db.t.graph_nodes()
[{'node': 'mod.foo',
  'flavor': 'function',
  'file': '',
  'method': 'foo',
  'pagerank': 0.64912,
  'in_degree': 1,
  'out_degree': 0},
 {'node': 'mod.bar',
  'flavor': 'function',
  'file': '',
  'method': 'bar',
  'pagerank': 0.35088,
  '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

To graph real code, use from_dir for a directory of .py files or from_pkg for an installed package. from_pkg finds the package through Python’s import system. Both pass their files to process_files, which you can also call with your own list. process_files splits the files into batches of sz files. It runs the batches in worker processes when there is more than one batch, and in the calling process when there is one.

Module names come from file paths. from_dir names modules relative to the directory’s parent. from_dir('../kosha') gives kosha.core, kosha.graph and so on. process_files without a root uses each file’s import root, the nearest ancestor directory without an __init__.py. from_dir and from_pkg also accept globtastic filters such as skip_folder_re. Each method returns the graph. Calls chain.

CodeGraph.from_pkg

def from_pkg(
    pkg:str, path:list=None, *, recursive:bool=True, maxdepth:int=None, symlinks:bool=True, file_glob:str=None,
    file_re:str=None, path_glob:str=None, path_re:str=None, folder_re:str=None, skip_file_glob:str=None,
    skip_file_re:str=None, skip_folder_re:str=None, func:<built-in function callable>=walk_join,
    ret_folders:bool=False, sort:bool=True, types:str | list=None, exts:str | list=None
)->CodeGraph:

Build from the source of an installed package.

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, path_glob:str=None, path_re:str=None,
    folder_re:str=None, skip_file_glob:str=None, skip_file_re:str=None, skip_folder_re:str=None,
    func:<built-in function callable>=walk_join, ret_folders:bool=False, sort:bool=True, types:str | list=None,
    exts:str | list=None
)->CodeGraph:

Build from the .py files under dir, naming modules from its parent directory.

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 extraction pass; reduce for very large packages
)->__main__.CodeGraph:

Add the call graph of the given .py files, then update degrees and PageRank.

import multiprocessing as mp, sys, tempfile
# One batch parses in the calling process. nbdev-docs on Linux runs these cells under spawn, where a pool cannot pickle a function defined in a notebook.
# fastcore forces fork on macOS. The cell sets `sys.platform` to 'linux' to reproduce the spawn case here.
_sm, _pf = mp.get_start_method(), sys.platform; mp.set_start_method('spawn', force=True); sys.platform = 'linux'
try:
    assert CodeGraph(':memory:').from_sources({'m': 'def a(): b()\ndef b(): pass\n'}).gn(where="node='m.a'")
    _sd = Path(tempfile.mkdtemp()); (_sd/'m.py').write_text('def a(): pass\n')
    assert CodeGraph(':memory:').process_files([_sd/'m.py'], root=_sd).gn(where="node='m.a'")
finally: mp.set_start_method(_sm, force=True); sys.platform = _pf

The next three cells test resolve_attr_calls on hints inserted by hand: one hint with one matching node, a hint that could match two packages, and a hint whose package has no matching node.

_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)
# a second call adds no edge
_g.resolve_attr_calls()
test_eq(len(list(_g.ge(where="kind='attr_call'"))), 1)
_g.ge(where="kind='attr_call'")
[{'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)

The rest of the notebook queries a graph built from four installed packages (litesearch, fastlite, apswutils and fastcore) and from the kosha source itself.

cg=CodeGraph(':memory:')
cg.from_pkg('litesearch').from_pkg('fastlite').from_pkg('apswutils').from_pkg('fastcore')
<__main__.CodeGraph>
cg=cg.from_pkg('protobuf')
cg.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',
  'method': 'get_store',
  'pagerank': 0.00022,
  'in_degree': 1,
  'out_degree': 4}]
cg.from_dir('../kosha')
<__main__.CodeGraph>
cg.db.t.graph_nodes.count
14165
cg.gn('node like "%emb_doc%"')
[]

Querying the graph

All queries take full node names. callers(n) and callees(n) return the nodes one edge away, and with_kind=True returns (name, kind) pairs instead. neighbors(n, depth) returns every node within depth edges in either direction.

co_dispatched(node) returns the other members of the node’s co-dispatch group: the functions it shares a route list or handler table with. ranked(k, module) returns the k nodes with the highest PageRank, optionally only those under a module prefix.

short_path(src, tgt) follows edges from caller to callee and returns the shortest chain from src to tgt, or an empty list when there is none. short_paths(nodes, k) does this for every pair among the first k nodes and returns the paths longest first. edge_kinds() counts the edges of each kind.

Call node_info(node) before you change a function. It returns the node’s row from graph_nodes together with its callers, callees and co-dispatch peers. node_infos returns the same records for many nodes with one query per table. Kosha.context uses node_infos to annotate search hits.

CodeGraph.node_info

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

The graph_nodes row for node plus its 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
)->fastcore.foundation.L:

Shortest paths between pairs among the first k of nodes, longest first.

CodeGraph.short_path

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

Shortest chain of edges from src to tgt, or an empty L if there is none.

CodeGraph.ranked

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

The k nodes with the highest PageRank, optionally only those whose name starts with module and a dot.

CodeGraph.co_dispatched

def co_dispatched(
    node:str, # fully-qualified node name
)->fastcore.foundation.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
)->fastcore.foundation.L:

Every node within depth edges of n, in either direction.

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
)->fastcore.foundation.L:

Nodes that n has an edge to; with_kind=True gives (callee, kind) pairs.

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
)->fastcore.foundation.L:

Nodes with an edge into n; with_kind=True gives (caller, kind) pairs.

CodeGraph.node_infos

def node_infos(
    nodes, # iterable of fully-qualified node names
    with_kind:bool=False, # include edge kind in callers/callees tuples
)->dict:

node_info for many nodes at once, as {node: record}, with one query per table.

from fastcore.test import test_eq
# node_infos (batched) must match node_info (per-node) exactly
_tg = CodeGraph(':memory:').from_sources({'m': 'def a(): b()\ndef b(): c()\ndef c(): pass\n'})
_ns = L(_tg.gn(select='node')).attrgot('node')
_bat = _tg.node_infos(_ns)
for n in _ns:
    _si = _tg.node_info(n)
    test_eq(set(_bat[n]['callers']), set(_si['callers']))
    test_eq(set(_bat[n]['callees']), set(_si['callees']))
    test_eq(set(_bat[n]['co_dispatched']), set(_si['co_dispatched']))
test_eq(_tg.node_infos([]), {})
print('node_infos == node_info OK')
█

 |----------------------------------------| 0.00% [0/1 00:00<?]
 |████████████████████████████████████████| 100.00% [1/1 00:00<00:00]
                                                                     
node_infos == node_info OK
cg.db.t.graph_edges(where='caller like "%apswutils.db.Table.insert_all%" and callee like "%insert_chunk%"')
[{'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_float', 'fastcore.basics.to_bool', 'fastcore.basics.to_int']
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',
 'method': 'to_date',
 'pagerank': 0.00019,
 'in_degree': 1,
 'out_degree': 2,
 'callers': ['fastcore.basics'],
 'callees': ['fastcore.basics._typeerr', 'fastcore.basics.str2date'],
 'co_dispatched': ['fastcore.basics.to_bool', 'fastcore.basics.to_float', 'fastcore.basics.to_int']}
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',
 'method': 'env_context',
 'pagerank': 0.00047,
 'in_degree': 1,
 'out_degree': 9,
 'callers': ['kosha.core.Kosha'],
 'callees': ['fastcore.all.L', 'fastcore.all.true', 'json.loads', 'kosha.core.Kosha', 'kosha.core.fanout', 'kosha.core.filt2wh', 'kosha.core.parseq', 'litesearch.api.search', 'litesearch.core.rerank_hits'],
 'co_dispatched': []}
cg.short_path('kosha.graph.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.data.kw', 'pagerank': 0.00314}, {'node': 'litesearch.core.db_lock', 'pagerank': 0.00158}, {'node': 'litesearch.api.Index', 'pagerank': 0.0013}, {'node': 'litesearch.core.write_txn', 'pagerank': 0.00104}, {'node': 'litesearch.core.process_content', 'pagerank': 0.00095}]
cg.callers('litesearch.data.kw')
['litesearch.api.add', 'litesearch.api.add_code', 'litesearch.api.search', 'litesearch.api.toc', 'litesearch.api.sections', 'litesearch.api.read', 'litesearch.api.context', 'litesearch.core.embed_chunk', 'litesearch.core.process_content', 'litesearch.core.get_store', 'litesearch.core.get_index', 'litesearch.core._sync_index', 'litesearch.core.sync', 'litesearch.core.database', 'litesearch.data.pdf_markdown', 'litesearch.data.ocr_parse', 'litesearch.data.oxide_parse', 'litesearch.data.oxide_parse._page', 'litesearch.data.SafeFastChunker.chunk', 'litesearch.data._parse_files', 'litesearch.data.pkg2chunks', 'litesearch.data.pre', 'litesearch.graph.get_graph', 'litesearch.graph.text_entities', 'litesearch.graph._pool', 'litesearch.graph._usearch_clusters', 'litesearch.graph.graph_search', 'litesearch.tree.get_tree', 'litesearch.tree.add_file', 'litesearch.tree._add_parsed', 'litesearch.tree.add_dir', 'litesearch.tree.doc_search', 'litesearch.tree.sections', 'litesearch.tree.context', 'litesearch.utils.FastEncode._encode_batch', 'litesearch.utils.FastEncode.encode', 'litesearch.utils.FastEncode.encode_document', 'litesearch.utils.FastEncode.encode_query', 'litesearch.utils.AutoLateChunkFastEncode.encode_auto', 'litesearch.utils.doc_encoder._', 'litesearch.utils.query_encoder._', 'litesearch.utils.FastEncodeImage._encode_batch', 'litesearch.utils.FastEncodeImage.embed', 'litesearch.utils.FastEncodeMultimodal.encode_text', 'litesearch.utils.FastEncodeMultimodal.encode_image', 'litesearch.utils.encode_pdf_texts']

Keeping the graph current

Kosha deletes the nodes and edges of a changed file before it writes new ones. file2nodes lists the nodes defined in a file. The private _drop_file deletes those nodes and their edges.

CodeGraph.file2nodes

def file2nodes(
    path:str
)->set:

Names of the nodes defined in the file at 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

The graph on a Kosha

Kosha.graph opens .kosha/graph.db in the repository, once per Kosha object. graphdb, gn and ge expose the database and its node and edge tables for direct queries. ni, neighbors, short_path and short_paths are the CodeGraph methods of the same names. ni is node_info. On a Kosha called k, k.ni('fastcore.basics.merge') returns the node row with its callers, callees and co-dispatch peers.

Kosha.short_paths

def short_paths():

The graph’s short_paths among a list of nodes.

Kosha.short_path

def short_path():

The graph’s short_path between two nodes.

Kosha.neighbors

def neighbors():

The graph’s neighbors: every node within depth edges.

Kosha.ni

def ni():

The graph’s node_info: a node’s row plus its callers, callees and co-dispatch peers.

Kosha.ge

def ge():

The graph_edges table, for direct queries.

Kosha.gn

def gn():

The graph_nodes table, for direct queries.

Kosha.graphdb

def graphdb():

The graph database, for direct queries.

Kosha.graph

def graph()->__main__.CodeGraph:

The call graph in .kosha/graph.db, opened once per Kosha object.

Public API

public_api(pkg) lists the public names of a package, or of a module within it. It reads both the package index and the repo index. A name is public when it is in the module’s __all__. In a module without __all__, a name is public when it has no leading underscore. The list also holds functions that fastcore’s @patch adds to a class. Kosha finds them through the graph’s patch edges.

The list has this order: the package itself, then top-level names before nested names such as methods, then alphabetical order. meta_cols is a comma-separated string of metadata fields. Each entry is a dict with one key per field. limit caps the number of entries.

Kosha.public_api

def public_api(
    pkg:str, # package or module name, such as `fastcore` or `fastcore.basics`
    meta_cols:str='mod_name,docstring', # comma-separated metadata fields to return for each name
    limit:int=200
)->fastcore.foundation.L:

Public names of a package or module, with top-level names before nested methods.

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: 119 | httpx: 158
kosha sample: ['kosha.cli.api_paths', 'kosha.cli.context', 'kosha.cli.daemon', 'kosha.cli.dep_stack', 'kosha.cli.env_context', 'kosha.cli.install', 'kosha.cli.main', 'kosha.cli.ni']
k.public_api('litesearch.data')
[{'mod_name': 'litesearch.data.Profile', 'docstring': 'How one *kind* of document should be read: parser, structure, chunker.'}, {'mod_name': 'litesearch.data.SafeFastChunker', 'docstring': '`FastChunker` whose cuts land on character boundaries.'}, {'mod_name': 'litesearch.data.add_wc', 'docstring': 'Add wild card * to each word in the query.`kw` uses UAX#29 segmentation, which keeps'}, {'mod_name': 'litesearch.data.chunk_markdown', 'docstring': 'Split markdown into paragraph chunks on blank lines'}, {'mod_name': 'litesearch.data.chunk_spans', 'docstring': 'Split text into chunks, returning (start_char, end_char, text) spans into the original text.'}, {'mod_name': 'litesearch.data.chunk_texts', 'docstring': 'Chunk texts using Fast Chunker'}, {'mod_name': 'litesearch.data.clean', 'docstring': 'Clean the query by removing * and returning None for empty queries.'}, {'mod_name': 'litesearch.data.clean_md', 'docstring': None}, {'mod_name': 'litesearch.data.dir2chunks', 'docstring': 'Return code chunks from a directory with extra metadata.'}, {'mod_name': 'litesearch.data.dir2files', 'docstring': 'Return list of python files in a package excluding tests and setup files.'}, {'mod_name': 'litesearch.data.file_parse', 'docstring': 'Parse a code file or code string and return code chunks with metadata.'}, {'mod_name': 'litesearch.data.fix_layout', 'docstring': 'Swap a page whose markdown cut values loose from their labels for its plain-text layer.'}, {'mod_name': 'litesearch.data.images_to_pdf', 'docstring': 'Convert a list of images into a multi-page image-only PDF (one image per page).'}, {'mod_name': 'litesearch.data.img2png', 'docstring': 'Normalise image input (PIL Image / bytes / path) to PNG bytes as RGB.'}, {'mod_name': 'litesearch.data.installed_packages', 'docstring': 'Return list of installed packages. If nms is provided, return only those packages.'}, {'mod_name': 'litesearch.data.ipynb_parse', 'docstring': 'Parse ipynb files'}, {'mod_name': 'litesearch.data.kw', 'docstring': 'Query keywords by UAX#29 word segmentation (apsw), minus stopwords.'}, {'mod_name': 'litesearch.data.mk_wider', 'docstring': 'Widen the query by joining words with OR operator.'}, {'mod_name': 'litesearch.data.mv_skill_md', 'docstring': 'Copy bundled SKILL.md to skill directories (.agents/, .claude/, .Codex/).'}, {'mod_name': 'litesearch.data.needs_ocr', 'docstring': 'True when pdf_oxide flagged the page as scanned (no extractable text layer).'}, {'mod_name': 'litesearch.data.non_py_sigs', 'docstring': 'Signatures for paths that are not python code files using codesigs'}, {'mod_name': 'litesearch.data.ocr_parse', 'docstring': None}, {'mod_name': 'litesearch.data.orphan_vals', 'docstring': 'Count value-only lines'}, {'mod_name': 'litesearch.data.oxide_parse', 'docstring': 'Parse a pdf using pdf-oxide to markdown with images extracted to a subdir.'}, {'mod_name': 'litesearch.data.pdf_parse', 'docstring': 'PdfDocument to markdown; if OCR is required, uses LiteParse for better formatting'}, {'mod_name': 'litesearch.data.pkg2chunks', 'docstring': 'Return code chunks from a package with extra metadata.'}, {'mod_name': 'litesearch.data.pkg2files', 'docstring': 'Return list of python files in a package excluding tests and setup files.'}, {'mod_name': 'litesearch.data.png_det', 'docstring': 'Return (width, height, concatenated IDAT bytes) from a PNG.'}, {'mod_name': 'litesearch.data.pre', 'docstring': 'Preprocess the query for fts search.'}, {'mod_name': 'litesearch.data.profile_for', 'docstring': 'The profile that claims this document, or None for the default pipeline.'}, {'mod_name': 'litesearch.data.pyparse', 'docstring': 'Parse a code string or python file and return code chunks as list of dicts with content and metadata.\n\nOnly module-level definitions are ever returned, so `tree.body` *is* the candidate list: walking\nthe whole tree to tag every node with a `parent` and then filtering on "parent is the Module"\ncomputes an answer the body already gives. Dropping the walk and hoisting the line split take\nthe same 400 files from 23.4s to 3.3s, chunk-for-chunk identical.'}, {'mod_name': 'litesearch.data.register_profile', 'docstring': 'Add a profile to the registry, replacing any of the same name.'}, {'mod_name': 'litesearch.data.repo_root', 'docstring': 'Find the root of the current git repository, or None if not in a repo.'}, {'mod_name': 'litesearch.data.spec', 'docstring': 'Return the importlib ModuleSpec for a package, or None if not found.'}, {'mod_name': 'litesearch.data.Profile.claims', 'docstring': None}, {'mod_name': 'litesearch.data.SafeFastChunker.chunk', 'docstring': None}]
k.public_api('fastlite', meta_cols='mod_name')
[{'mod_name': 'fastlite.core.all_dcs'}, {'mod_name': 'fastlite.core.create_mod'}, {'mod_name': 'fastlite.core.diagram'}, {'mod_name': 'fastlite.core.get_typ'}, {'mod_name': 'fastlite.kw.MissingPrimaryKey'}, {'mod_name': 'fastlite.kw.database'}]
k.ni('fastlite.kw.database')
{'callers': [], 'callees': [], 'co_dispatched': []}

Freshness

status reports how far the index has drifted from the code on disk. It returns a dict with these keys:

  • files: the number of repo files indexed.
  • packages: the number of indexed package versions.
  • graph_nodes: the number of nodes in the graph.
  • stale_files: the number of indexed repo files modified since they were indexed.
  • new_files: the number of Python files in the repo that are not indexed.
  • stale_pkgs: a dict of {name: version} for each installed package not indexed at its installed version.

The packages considered are those listed in pyproject.toml and their dependencies, depth levels down. A sync with no package list indexes the packages in stale_pkgs.

Kosha.status

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

How fresh the index is: counts of files, packages and graph nodes, plus stale files, new files and stale packages.

status reads the project’s environment too: stale_pkgs lists a dependency installed only in the project’s .venv, and drops it once update_pkg has indexed that copy.

_root = Path(tempfile.mkdtemp())/'proj'; (_root/'.git').mkdir(parents=True)
(_root/'pyproject.toml').write_text('[project]\nname = "proj"\nversion = "0"\ndependencies = ["kkfake"]\n')
_sp = _root/'.venv/lib/python3.13/site-packages'; (_sp/'kkfake').mkdir(parents=True)
(_sp/'kkfake'/'__init__.py').write_text('def hello(): return 1\n')
(_sp/'kkfake-1.0.dist-info').mkdir()
(_sp/'kkfake-1.0.dist-info'/'METADATA').write_text('Metadata-Version: 2.1\nName: kkfake\nVersion: 1.0\nAuthor-email: a@b.co\n')
_kp, _cwd = Kosha(dir=_root, xdg_dir=_root/'xdg'), os.getcwd()
os.chdir(_root)
try:
    _before = _kp.status()['stale_pkgs']
    _kp.update_pkg('kkfake', embed=False, verbose=False)
    _after = _kp.status()['stale_pkgs']
finally: os.chdir(_cwd)
test_eq(list(_before), ['kkfake'])
test_eq(_after, {})
_before

status and sync take the dependencies from the pyproject.toml in Kosha.root, not from the repository around the current folder. A server whose working folder is elsewhere, or inside another repository, still indexes the packages of the Kosha(dir=...) it was given. The next cell runs from repository a against a Kosha on b, which has no .git of its own.

def _fake_repo(base, nm, git=False):
    r = base/nm; (r/'.git').mkdir(parents=True) if git else r.mkdir()
    (r/'pyproject.toml').write_text(f'[project]\nname = "{nm}"\nversion = "0"\ndependencies = ["kkfake{nm}"]\n')
    sp = r/'.venv/lib/python3.13/site-packages'; (sp/f'kkfake{nm}').mkdir(parents=True)
    (sp/f'kkfake{nm}'/'__init__.py').write_text('def hello(): return 1\n')
    (sp/f'kkfake{nm}-1.0.dist-info').mkdir()
    (sp/f'kkfake{nm}-1.0.dist-info'/'METADATA').write_text(f'Metadata-Version: 2.1\nName: kkfake{nm}\nVersion: 1.0\nAuthor-email: a@b.co\n')
    return r
_tmp = Path(tempfile.mkdtemp())
_ra, _rb = _fake_repo(_tmp, 'a', git=True), _fake_repo(_tmp, 'b')
_kb, _cwd = Kosha(dir=_rb, xdg_dir=_tmp/'xdg'), os.getcwd()
os.chdir(_ra)
try:
    _stale = _kb.status()['stale_pkgs']
    test_eq(list(_stale), ['kkfakeb'])
    test_eq(list(env_pkg_versions(path=_kb.env_path, root=_kb.root)), ['kkfakeb'])
finally: os.chdir(_cwd)
_stale

Ranking

A search returns candidates from two lists, the repo index and the package index. Kosha merges the lists with reciprocal rank fusion (RRF). Each candidate scores 1/(k + rank) for every list that holds it. That order does not use the structure of code. rank_results adjusts the scores with rules adapted from the semble code-search library. In the table, top is the highest score among the candidates.

Rule Condition Effect on the score
1 is_symbol_query is true and the chunk defines the symbol, as in def Client, class Client or CREATE TABLE Client Add 3.0 x top. Add 4.5 x top when the file name also matches the symbol.
2a The query is natural language and the file or folder name matches at least 10% of its words Add top x the share of words matched
2b The query is natural language and names a CamelCase symbol that the chunk defines Add 1.5 x top, or 2.25 x top when the file name also matches
3 The chunk comes from a package that the query names Multiply by 1.5
4 A file has several high-scoring chunks Add up to 0.2 x top to the best chunk of the file, in proportion to the sum of the file’s scores
5a The path is a test file or test folder, a compat, _compat or legacy folder, or an examples folder Multiply by 0.3 for each of these that matches
5b The file is __init__.py or package-info.java Multiply by 0.5
5c The file ends in .d.ts Multiply by 0.7
6 The file already supplies a chunk to the result Each further chunk from the file counts half as much as the one before it

is_symbol_query is true for a dotted, ::, -> or backslash-qualified name, a name with a leading underscore, a name with a capital letter or underscore after its first character, and a name that starts with a capital letter. Examples are Client, Sinatra::Base and _private. penalise_paths=False turns off rules 5a to 5c. Rule 6 stops one file from filling the results.

rank_results

def rank_results(
    results, query, soft_pkgs:NoneType=None, top_k:NoneType=None, penalise_paths:bool=True
)->fastcore.foundation.L:

Reorder search hits for code: symbol and file-name boosts, package boosts, and penalties for tests and repeated files.

is_symbol_query

def is_symbol_query(
    query:str
)->bool:

True if the query looks like a bare symbol or namespace-qualified identifier.

from fastcore.test import test_eq, test_fail

# is_symbol_query: symbols vs NL
for s in ['Client','Sinatra::Base','_private','StateManager','foo.bar']: assert is_symbol_query(s), s
for s in ['session','merge dicts','extract bytes from image']: assert not is_symbol_query(s), s

# path penalties: tests/__init__/examples/.d.ts penalised, plain source not
assert _file_penalty('kosha/core.py') == 1.0
assert _file_penalty('tests/test_core.py') < 1.0
assert _file_penalty('pkg/__init__.py') == _MODERATE_PENALTY
assert _file_penalty('examples/demo.py') == _STRONG_PENALTY
assert _file_penalty('types/foo.d.ts') == _MILD_PENALTY

def _r(content='x', path='a.py', pkg=None, rrf=0.01, **kw):
    md = {'path': path, 'package': pkg}
    return dict(content=content, metadata=md, path=path, _rrf_score=rrf, **kw)

# package boost: fastcore ranked above equal-base non-fastcore, but non-fastcore still present
res = [_r('a', 'other.py', pkg='numpy', rrf=0.01), _r('b', 'fc.py', pkg='fastcore', rrf=0.01)]
ranked = rank_results(res, 'extract bytes', soft_pkgs={'fastcore'})
test_eq(len(ranked), 2)
test_eq(_rk_pkg(ranked[0]), 'fastcore')

# empty / no-crash
test_eq(rank_results([], 'q'), L())
test_eq(len(rank_results([_r()], '')), 1)
print('ranking tests passed')
ranking tests passed

Incremental sync

Re-reading a large environment takes minutes. Sync reads only the files whose content has changed. file_index stores a BLAKE2 hash of every analysed file. sync_dir compares the hashes, drops the nodes of deleted files and re-extracts changed files.

In the default mode='fast', sync_dir replaces the nodes and outgoing edges of a changed file. It keeps the edges from other files into that file. It removes edges into nodes that no longer exist. mode='full' re-extracts through process_files, which uses worker processes.

sync_pkgs runs sync_dir for the directory of each installed package. CodeGraph.sync runs sync_dir and sync_pkgs. If anything changed, it then updates degrees, and it updates PageRank when the edge count changed by more than 10%. metrics=False skips that step. Call recompute_metrics() once after several syncs. force=True re-reads every file.

This cell redefines _drop_file, _index_files, _stale and resolve_attr_calls from earlier in the notebook. The exported module contains the later versions.

CodeGraph.resolve_attr_calls

def resolve_attr_calls()->CodeGraph:

Turn pending self.x.method() hints into attr_call edges, now that the graph knows which methods exist.

CodeGraph.recompute_metrics

def recompute_metrics()->CodeGraph:

Recompute PageRank and degree metrics once after graph updates.

CodeGraph.sync

def sync(
    dir:NoneType=None, pkgs:NoneType=None, force:bool=False, mode:str='fast', metrics:bool=True, path:list=None
)->CodeGraph:

Update the graph for a directory and a list of packages. If anything changed and metrics is true, update degrees, and update PageRank when the edge count changed by more than 10%. force clears the pending method-call hints and reprocesses every file.

CodeGraph.sync_pkgs

def sync_pkgs(
    pkgs:list[str], force:bool=False, mode:str='fast', path:list=None
)->CodeGraph:

Update the graph for installed packages, skipping files whose content hash is unchanged.

CodeGraph.sync_dir

def sync_dir(
    dir:str | pathlib.Path, force:bool=False, mode:str='fast', root:NoneType=None
)->CodeGraph:

Update the graph for the .py files under dir. Skip files whose content hash is unchanged, and drop the rows of files that no longer exist. force reprocesses every file. mode="full" re-extracts changed files with process_files. root sets the module root.

from tempfile import TemporaryDirectory
from kosha.core import imp_root
with TemporaryDirectory() as _d:
    _sub = Path(_d)/'pkg'; _sub.mkdir()
    (_sub/'core.py').write_text('def helper(x): return x+1\ndef caller(x): return helper(helper(x))\n')
    (_sub/'other.py').write_text('from core import caller\ndef outer(): return caller(2)\n')
    _g2 = CodeGraph(':memory:').sync(dir=_d)
    _ids = {r['node'] for r in _g2.gn()}
    test_eq(imp_root(_sub/'core.py'), _sub)
    assert 'core.helper' in _ids and 'pkg.core.helper' not in _ids, sorted(_ids)
    test_eq(list(_g2.callers('core.helper')), ['core.caller'])

Syncing and searching from Kosha

Kosha.sync brings the repo index, the package index and the call graph up to date. It runs update_repo, update_pkgs and the graph sync in parallel threads. in_parallel=False runs them in turn. With no pkgs, it indexes the packages that status() reports as stale. repo, env and graph switch each part off. embed=False stores new chunks without vectors. Only keyword search finds them until process_repo or process_env adds the vectors. Kosha.sync returns the results of the three parts, not the Kosha.

Kosha.context is the default search. It runs repo_context and env_context in parallel and merges their hits with RRF. It reorders the hits with rank_results and keeps the top limit. With graph=True, the default, each hit also holds its graph record: pagerank, in_degree, out_degree, callers, callees and co_dispatched.

There are two ways to name a package in a query, and they differ. A package:fastcore filter skips the repo search and returns only code from fastcore. A package name written as a plain word, as in fastcore merge dicts, only boosts the chunks of that package. Chunks of other packages still appear.

compact=True returns slim dicts with mod_name, docstring, lineno, path and sig. sig is the first line of the chunk. rerank=True reorders the final hits with a flashrank cross-encoder. That needs the flashrank model on disk. The saved output below shows the step skipped on a machine without the model.

Kosha.context

def context(
    q:str, # query with optional key:value filters
    limit:int=50, repo:bool=True, env:bool=True, graph:bool=True, boost:bool=True,
    compact:bool=False, # return slim dicts (mod_name, docstring, lineno, path, sig) instead of full chunks
    columns:str='content,metadata', sys_wide:bool=True,
    rerank:bool=False, # reorder the boosted top-k with a flashrank cross-encoder
    rerank_model:str=None, # flashrank model name; None uses the fast default
    **kw
)->fastcore.foundation.L: # forwarded to env_context / repo_context

Search the repo and the packages together, rerank the merged hits for code, and attach call-graph details.

Kosha.sync

def sync(
    pkgs:NoneType=None, # list of package names to sync (e.g. ['httpx', 'fastcore']); if None, sync every stale env package
    dir:NoneType=None, # directory to sync; if None, sync all of root
    verbose:bool=True, # print progress messages
    in_parallel:bool=True, # run the repo, package and graph updates in parallel threads; False runs them in turn
    graph_mode:str='fast', # 'full' re-extracts changed files in worker processes
    graph_metrics:bool=True, # refresh PageRank after the graph update; False leaves it for `k.graph.recompute_metrics()`
    force:bool=False, # reprocess every file and package, even unchanged ones, graph included
    repo:bool=True, # index the repo
    env:bool=True, # index installed packages; False indexes the repo only, even if `pkgs` is given
    graph:bool=True, # update the call graph
    sync_graph:NoneType=None, # deprecated: the old name for `graph`
    pyproject:bool=True, # if True, auto-detect env packages from the pyproject.toml in `Kosha.root` (not `dir`); if False, use pkgs argument as-is
    depth:int=1, # depth for pyproject env package detection; ignored if pyproject=False
    embed:bool=True, # compute vectors for new chunks; False stores them for keyword search only
    pkg_parallel:bool=False, # ingest env packages concurrently; requires Kosha(busy_timeout=...)
    chunk:int=5000, # rows per write transaction when pkg_parallel=True
)->Kosha:

Bring the repo index, the package index and the call graph up to date.

# `status()['stale_pkgs']` is a {name: version} dict. `listify` and `L` wrap a dict as one element instead of iterating its keys.
# `sync` must take the keys before it normalises the package names.
test_eq(listify({'fastcore': '1.8.0'}), [{'fastcore': '1.8.0'}])
test_eq(L(list({'fastcore': '1.8.0', 'httpx[cli]': '0.27'})).map(_pkg_name).unique(), ['fastcore', 'httpx'])

# extras are stripped, including multiple comma-separated extras and a version pin
for s in ['rishi[litert,llama,all]', ' rishi[litert,llama,all] ', 'rishi[all]>=0.2', 'rishi']: test_eq(_pkg_name(s), 'rishi')
test_eq(L(list({'rishi[litert,llama,all]': '0.1', 'rishi': '0.1'})).map(_pkg_name).unique(), ['rishi'])

print('sync pkg-normalisation tests passed')
sync pkg-normalisation tests passed
# `verbose=False` keeps stdout clean. A caller that prints JSON to stdout runs sync in the background.
import io, contextlib, tempfile
with tempfile.TemporaryDirectory() as d:
    d = Path(d); (d/'m.py').write_text('def f(): return 1\n')
    quiet = Kosha(d, xdg_dir=d/'xdg')
    with contextlib.redirect_stdout(io.StringIO()) as out: quiet.sync(verbose=False, env=False, graph=False, embed=False)
    test_eq(out.getvalue(), '')

Fetching 6 files:   0%|          | 0/6 [00:00<?, ?it/s]
Fetching 6 files: 100%|##########| 6/6 [00:00<00:00, 19.44it/s]
Fetching 6 files: 100%|##########| 6/6 [00:00<00:00, 19.40it/s]

parse files from /tmp/tmplakkpc7x:   0%|          | 0/1 [00:00<?, ?it/s]
parse files from /tmp/tmplakkpc7x: 100%|##########| 1/1 [00:00<00:00, 3968.12it/s]
k = Kosha()
# rerank reorders the same top-k rather than changing which chunks are retrieved.
base = k.context('atomic file write', limit=10, graph=False)
try: rr = k.context('atomic file write', limit=10, graph=False, rerank=True)
except Exception as e: print('rerank skipped:', type(e).__name__)
else:
    assert len(rr) <= 10
    mods = lambda r: {x['metadata']['mod_name'] for x in r}
    assert mods(rr) <= mods(base), 'rerank must not introduce chunks the pipeline did not retrieve'
    print(L(zip(base.attrgot('metadata').attrgot('mod_name'), rr.attrgot('metadata').attrgot('mod_name'))))
rerank skipped: NoSuchFile
k.gn(select='*', where='node like "%get_store"')
[{'node': 'core.get_store',
  'flavor': 'function',
  'file': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/litesearch/core.py',
  'pagerank': 0.00018,
  'in_degree': 1,
  'out_degree': 0,
  'method': 'get_store'}]
k.status()
{'files': 5,
 'packages': 498,
 'graph_nodes': 10469,
 'stale_files': 0,
 'stale_pkgs': {},
 'new_files': 1}
k.sync()
Syncing dir=/Users/71293/code/personal/orgs/kosha, repo=True, env=True, graph=True, force=False
syncing files [Path('/Users/71293/code/personal/orgs/kosha/kosha/skill.py')] .....

parse files from /Users/71293/code/personal/orgs/kosha:   0%|                                                                           | 0/1 [00:00<?, ?it/s]
parse files from /Users/71293/code/personal/orgs/kosha: 100%|█████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00, 1157.69it/s]
{'changed': 0, 'same': 0, 'removed': 0}
synced repo
[None, None, <__main__.CodeGraph object>]
'fossick' in env_pkg_versions()
False
first(k.context('how do I merge dicts package:fastcore path:basics', limit=1, graph=False, compact=False))
{ '_rrf_score': 0.016666666666666666,
  '_score': 0.030000000000000002,
  '_source': 'env',
  '_src_id': 'env:38620',
  'content': 'def merge(*ds):\n'
             '    "Merge all dictionaries in `ds`"\n'
             '    return {k:v for d in ds if d is not None for k,v in '
             'd.items()}',
  'metadata': { 'docstring': 'Merge all dictionaries in `ds`',
                'end_lineno': 715,
                'lang': '.py',
                'lineno': 713,
                'mod_name': 'fastcore.basics.merge',
                'name': 'merge',
                'package': 'fastcore',
                'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fastcore/basics.py',
                'public_api': True,
                'type': 'FunctionDef',
                'uploaded_at': 1785661233.7787635,
                'version': '2.1.17'},
  'rank': -10.799262836903566,
  'rowid': 38620}

Package dependencies

When Kosha indexes a package, it counts which other packages the modules of that package import. It stores the counts in the pkg_deps table with the columns from_pkg, to_pkg and n_modules. dep_stack(seeds, depth) walks that table breadth first. Layer 0 is the seeds. Layer 1 is the packages that the seeds import, and so on, up to depth layers. With no_sys_pkg=True, the result keeps only names that are installed distributions (is_sys_pkg). allow restricts the result to a list of names. Use dep_stack to see what a package depends on before you read its code.

top_nodes(pkg, k) returns the k public names of a package with the highest PageRank, which are its most used entry points. api_call_paths(from_pkg, to_pkg, k) takes the top k nodes of each package. It returns {target: path} for the shortest call paths from the entry points of the first package to the entry points of the second. An empty dict, as in the fastlite example, means that no top node of the first package reaches a top node of the second.

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
)->fastcore.foundation.L:

The k public API nodes of pkg with the highest 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:

Layers of package imports from the pkg_deps table, breadth first from seeds. Layer 0 is seeds. depth is the maximum number of layers after it. no_sys_pkg keeps only installed packages. A non-empty allow keeps only the packages it lists.

is_sys_pkg

def is_sys_pkg(
    pkg:str, path:list=None
)->bool:

True if pkg is the name of an installed distribution in path, or in the running interpreter when path is empty.

k.update_pkg('fastlite', embed=False, verbose=False)   # a fresh env.db (as in CI) has not indexed it yet
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'], {'apswutils', 'apsw', 'fastcore'}]
-------------------------------------------------
_dep_stack ok: ['fastlite', 'apswutils', 'apsw', 'fastcore']
k.dep_stack(['fastlite'], allow=['fastcore'])
[['fastlite'], {'fastcore'}]
k.api_call_paths('fastlite', 'apswutils')
{}
k.context('payments page monsterui fasthtml', repo=False)[0]
{ '_rrf_score': 0.016129032258064516,
  '_score': 0.02278063164009319,
  '_source': 'env',
  '_src_id': 'env:35278',
  'callees': [],
  'callers': [],
  'co_dispatched': [],
  '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': { 'docstring': None,
                'end_lineno': 147,
                'lang': '.py',
                'lineno': 133,
                'mod_name': 'fasthtml.stripe_otp.success',
                'name': 'success',
                'package': 'python-fasthtml',
                'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py',
                'public_api': True,
                'type': 'FunctionDef',
                'uploaded_at': 1785635980.530908,
                'version': '0.14.11'},
  'rank': -27.32216094526391,
  'rowid': 35278}
k.context('payments page package:python-fasthtml')[0]
{ '_rrf_score': 0.016666666666666666,
  '_score': 0.02871265928424982,
  '_source': 'env',
  '_src_id': 'env:35281',
  'callees': [],
  'callers': [],
  'co_dispatched': [],
  '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': { 'docstring': None,
                'end_lineno': 211,
                'lang': '.py',
                'lineno': 185,
                'mod_name': 'fasthtml.stripe_otp.account_management',
                'name': 'account_management',
                'package': 'python-fasthtml',
                'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/stripe_otp.py',
                'public_api': True,
                'type': 'FunctionDef',
                'uploaded_at': 1785635980.530908,
                'version': '0.14.11'},
  'rank': -16.67155688011088,
  'rowid': 35281}
k.context('fetch a page using broswer session package:fossick', repo=False)
[{'rowid': 43620, 'content': 'def fetch_page(url:str, sel:str|None=None, heavy:bool=False, stealthy:bool=False, session:bool=False,\n               auto:bool=False, max_chars:int=8000) -> dict:\n    "Fetch a URL as markdown. Always pass sel (CSS selector) to skip nav/ads. heavy=JS rendering; stealthy=anti-bot fetcher; session=reuse the logged-in debug Chrome\'s cookies; auto=escalate plain->heavy->stealthy->session (winning tier in result)."\n    p = _fetch(url, sel=sel, heavy=heavy, stealthy=stealthy, session=session, auto=auto)\n    return {\'url\': p[\'url\'], \'status\': p[\'status\'], \'tier\': getattr(p, \'tier\', None),\n            \'markdown\': _trunc(to_md(p), max_chars)}', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/mcp.py', 'uploaded_at': 1784774842.3057618, 'name': 'fetch_page', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 69, 'end_lineno': 74, 'mod_name': 'fossick.mcp.fetch_page', 'package': 'fossick', 'version': '0.0.15', 'docstring': "Fetch a URL as markdown. Always pass sel (CSS selector) to skip nav/ads. heavy=JS rendering; stealthy=anti-bot fetcher; session=reuse the logged-in debug Chrome's cookies; auto=escalate plain->heavy->stealthy->session (winning tier in result).", 'public_api': True}, 'rank': -23.781455641736635, '_rrf_score': 0.016129032258064516, '_source': 'env', '_src_id': 'env:43620', '_score': 0.0272847274634158, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43545, 'content': 'def fetch(\n    url:str,              # URL to fetch\n    sel:str=None,         # CSS selector to extract (None = full page)\n    heavy:bool=False,     # JS rendering via headless browser\n    stealthy:bool=False,  # Anti-bot stealth fetcher\n    session:bool=False,   # Route through the persistent debug Chrome (reuses its logged-in cookies)\n    auto:bool=False,      # Auto-escalate plain->heavy->stealthy->session on bot-block detection\n    as_json:bool=False,   # Output JSON instead of markdown\n):\n    "Fetch a URL and print as markdown."\n    page = _fetch(url, sel=sel, heavy=heavy, stealthy=stealthy, session=session, auto=auto)\n    if as_json: print(json.dumps({\'url\': page[\'url\'], \'status\': page[\'status\'], \'tier\': getattr(page,\'tier\',None), \'content\': to_md(page)}))\n    else: print(to_md(page))', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/cli.py', 'uploaded_at': 1784774842.3050234, 'name': 'fetch', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 28, 'end_lineno': 40, 'mod_name': 'fossick.cli.fetch', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'Fetch a URL and print as markdown.', 'public_api': True}, 'rank': -24.55014030857464, '_rrf_score': 0.016666666666666666, '_source': 'env', '_src_id': 'env:43545', '_score': 0.026929660304643974, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 201624, 'content': 'def get_page(url, method=\'GET\', payload=None, heavy=False, stealthy=False, session=False,\n             allow_private=None, **kw) -> Response:\n\t"""Fetch `url`. `session=True` (or a port int) drives the persistent debug Chrome via CDP, reusing its logged-in cookies.\n\n\tEvery fetch in fossick arrives here, which is why the private-address check lives here and not in\n\t`fetch`: `crawl`, `replay_xhr`, `download_files` and `http_get` would each have to remember."""\n\tassert method.upper() in (\'GET\', \'POST\'), \'Only GET and POST methods are supported\'\n\tcheck_url(url, allow_private)\n\tif session:\n\t\tfrom fossick.cdp import cdp_ws, cdp_cookies\n\t\tport = 9223 if session is True else int(session)\n\t\tkw.setdefault(\'cdp_url\', cdp_ws(port))\n\t\tif (ck := cdp_cookies(url, port=port)): kw.setdefault(\'cookies\', ck)\n\t\tkw.pop(\'verify\', None)\n\t\tif not stealthy: heavy = True\n\tif not (heavy or stealthy): return http_page(url, method, payload, **kw)\n\tStealthyFetcher, DynamicFetcher = _browser_fetchers()\n\tf = DynamicFetcher.async_fetch if heavy else StealthyFetcher.async_fetch\n\treturn syncy(f(url, **kw))', 'metadata': {'path': '/Users/71293/code/personal/orgs/ramabana/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1786594693.3198924, 'name': 'get_page', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 172, 'end_lineno': 190, 'mod_name': 'fossick.core.get_page', 'package': 'fossick', 'version': '0.1.11', 'docstring': 'Fetch `url`. `session=True` (or a port int) drives the persistent debug Chrome via CDP, reusing its logged-in cookies.\n\nEvery fetch in fossick arrives here, which is why the private-address check lives here and not in\n`fetch`: `crawl`, `replay_xhr`, `download_files` and `http_get` would each have to remember.', 'public_api': True}, 'rank': -23.059271349007616, '_rrf_score': 0.01639344262295082, '_source': 'env', '_src_id': 'env:201624', '_score': 0.026572917741426724, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43585, 'content': 'def fetch_all(urls:list,           # URLs to fetch\n              sel:str=None,         # CSS selector to extract per page (None = full page)\n              concurrency:int=8,    # Max parallel fetches\n              heavy:bool=False, # Full JS rendering via headless browser\n              stealthy:bool=False, # Anti-bot stealth fetcher (Cloudflare etc.)\n              **kw                  # Extra kwargs passed to fetch()\n             ) -> list:\n    "Fetch a list of URLs in parallel; returns Page dicts in the same order as urls"\n    return parallel(fetch, urls, sel=sel, heavy=heavy, stealthy=stealthy, threadpool=True, n_workers=concurrency, **kw)', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1784774842.3054273, 'name': 'fetch_all', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 235, 'end_lineno': 243, 'mod_name': 'fossick.core.fetch_all', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'Fetch a list of URLs in parallel; returns Page dicts in the same order as urls', 'public_api': True}, 'rank': -16.740357935560443, '_rrf_score': 0.015873015873015872, '_source': 'env', '_src_id': 'env:43585', '_score': 0.026478394833063013, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192798, 'content': 'async def shop_cart_page(page:Page, url:str=None) -> dict:\n    "Open the cart page (hint, on-page cart link, then `/cart`) and return the cart with its lines."\n    plat, cur = await page.shop_prep(), await page.eval(\'location.href\')\n    url = url or site_hint(cur).get(\'cart\') or (f\'{_origin(cur)}/cart\' if plat == \'shopify\' else None) \\\n              or await page.eval(\'__fk.cartLink()\') or f\'{_origin(cur)}/cart\'\n    await _goto(page, url)\n    return await page.shop_cart(lines=True)', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/shop.py', 'uploaded_at': 1785977784.1639626, 'name': 'shop_cart_page', 'lang': '.py', 'type': 'AsyncFunctionDef', 'lineno': 652, 'end_lineno': 658, 'mod_name': 'fossick.shop.shop_cart_page', 'package': 'fossick', 'version': '0.1.1', 'docstring': 'Open the cart page (hint, on-page cart link, then `/cart`) and return the cart with its lines.', 'public_api': False}, 'rank': -8.757235748920703, '_rrf_score': 0.0136986301369863, '_source': 'env', '_src_id': 'env:192798', '_score': 0.025547945205479452, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192750, 'content': 'def _fetcher():\n    """scrapling\'s plain-HTTP Fetcher, imported on first request. It pulls curl_cffi and\n    playwright\'s generated bindings — 0.21s of import for something only http_page and\n    replay_xhr touch."""\n    from scrapling.fetchers import Fetcher\n    return Fetcher', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1785977784.1587985, 'name': '_fetcher', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 93, 'end_lineno': 98, 'mod_name': 'fossick.core._fetcher', 'package': 'fossick', 'version': '0.1.1', 'docstring': "scrapling's plain-HTTP Fetcher, imported on first request. It pulls curl_cffi and\nplaywright's generated bindings — 0.21s of import for something only http_page and\nreplay_xhr touch.", 'public_api': False}, 'rank': -15.768862757024996, '_rrf_score': 0.015384615384615385, '_source': 'env', '_src_id': 'env:192750', '_score': 0.025230635957628672, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192771, 'content': 'def shop_open(url:str, port:int=9223) -> dict:\n    "Open a store in the persistent debug Chrome, or move the open session to `url` (a category or product page — passing the site root keeps it where it is), and report the platform, anything blocking (cookie banner, login, store/postcode), the current cart and what is on the page. Later shop_* tools act on this session."\n    s = _sh(url, port=port)\n    return {\'url\': s.url, \'platform\': s.platform(), \'blockers\': s.blockers(),\n            \'cart\': _jsonable(s.cart()), \'products\': _jsonable(s.products(25))}', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/mcp.py', 'uploaded_at': 1785977784.1604002, 'name': 'shop_open', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 248, 'end_lineno': 252, 'mod_name': 'fossick.mcp.shop_open', 'package': 'fossick', 'version': '0.1.1', 'docstring': 'Open a store in the persistent debug Chrome, or move the open session to `url` (a category or product page — passing the site root keeps it where it is), and report the platform, anything blocking (cookie banner, login, store/postcode), the current cart and what is on the page. Later shop_* tools act on this session.', 'public_api': True}, 'rank': -14.626207960943582, '_rrf_score': 0.012987012987012988, '_source': 'env', '_src_id': 'env:192771', '_score': 0.02088293276492673, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 193908, 'content': 'def _why(pg) -> str:\n    "Why a fetched page was unusable — surfaced in `research()[\'dropped\']` instead of vanishing."\n    if pg is None: return \'fetch failed\'\n    if (st := getattr(pg, \'status\', 0)) != 200: return f\'http {st}\'\n    if getattr(pg, \'tier\', \'\') == \'blocked\' or _blocked(pg): return \'bot wall / challenge page\'\n    return \'no readable content\'', 'metadata': {'path': '/Users/71293/code/personal/orgs/vishalakshi/.venv/lib/python3.13/site-packages/fossick/search.py', 'uploaded_at': 1786273543.1417477, 'name': '_why', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 271, 'end_lineno': 276, 'mod_name': 'fossick.search._why', 'package': 'fossick', 'version': '0.1.5', 'docstring': "Why a fetched page was unusable — surfaced in `research()['dropped']` instead of vanishing.", 'public_api': False}, 'rank': -15.374744350208976, '_rrf_score': 0.013157894736842105, '_source': 'env', '_src_id': 'env:193908', '_score': 0.02086407248433081, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 201673, 'content': 'def research(q:str,                 # search query\n             n:int=5,               # number of readable sources to gather\n             engine:str=\'search\',   # \'search\' (fused metasearch) or \'google\' (stealth browser ranking)\n             sel:str=None,          # CSS selector to narrow each page before markdown\n             chars:int=4000,        # max markdown chars kept per page\n             auto:bool=True,        # auto-escalate fetching (plain->heavy->stealthy->session) per page\n             region:str=\'auto\',     # ddgs region, or \'auto\' to read the country off the query\n             timelimit:str=None,    # d | w | m | y — only sources from the last day/week/month/year\n             curated:bool=True,     # drop mirror domains and rank by source authority before reading\n             intent:str=\'auto\',     # authority class (\'policy\', \'docs\', …), \'auto\' to read it off `q`\n             facets:int=4,          # searches to fan out over when `q` holds more than one question; 1 disables\n             rewrite=None,          # optional `f(q) -> list[str]` (a model) in place of the heuristic planner\n             focused:bool=True,     # keep query-relevant passages instead of the first `chars` chars\n             candidates:int=None,   # how many hits to draw on when backfilling (default 3n)\n             pages:int=1,           # result pages per backend (raise it when `candidates` exceeds one page)\n             concurrency:int=8,\n             **kw,                  # extra kwargs passed to fetch()\n            ) -> dict:\n    "Query -> read the top `n` *readable* results -> a cited markdown corpus. Returns {query, sources, digest, dropped, region, curation}."\n    if not (q and q.strip()):\n        return dict(query=q, sources=[], digest=\'\', dropped=[], region=_region(q, region),\n                    curation=None, plan=[])\n    cand = candidates or max(3*n, n+5)\n    region = _region(q, region)\n    qs = _plan(q, max_queries=facets, rewrite=rewrite)\n    def _hits(sq):\n        got = (google(sq, n=cand, region=region, timelimit=timelimit) if engine == \'google\'\n               else search(sq, n=cand, region=region, timelimit=timelimit, pages=pages))\n        return [dict(h, queries=[sq]) for h in got]\n    # Interleaved, not fused. RRF across facets would promote whatever ranks for all three, which is\n    # the generalist guide rather than the source that actually owns one of them.\n    hits = _hits(qs[0]) if len(qs) < 2 else interleave(\n        list(parallel(_hits, qs, n_workers=len(qs), threadpool=True)))\n    # Curate before fetching, not after: `research` reads the candidates in order, so the ordering\n    # *is* the choice of which pages get read. Filtering afterwards would have paid for them first.\n    hits, curation = _curate(q, hits, intent=intent) if curated else (hits, None)\n    todo, seen = [], set()\n    for h in hits:\n        u = h.get(\'href\') or h.get(\'url\')\n        if u and (key := norm_url(u)) not in seen:\n            seen.add(key); todo.append((h, u))\n    sources, dropped, i = [], [], 0\n    while len(sources) < n and i < len(todo):\n        batch = todo[i:i + (n - len(sources))]; i += len(batch)\n        for (h, u), pg in zip(batch, _fetch_batch([u for _, u in batch], concurrency, sel=sel, auto=auto, **kw)):\n            title = h.get(\'title\') or u\n            md = \'\'\n            if usable(pg):\n                try: md = to_md(pg, sel=sel)\n                except Exception: md = \'\'\n            if len(md.strip()) < _MIN_CHARS:\n                dropped.append(dict(title=title, href=u, reason=_why(pg))); continue\n            sources.append(dict(title=title, href=u, tier=getattr(pg, \'tier\', \'plain\'),\n                                date=page_date(pg), queries=h.get(\'queries\') or [q],\n                                md=focus(md, q, chars) if focused else md[:chars]))\n    sources = sources[:n]\n    # The date goes in the digest, not just the dict: the model reading this is the one that has to\n    # notice the price it is quoting was published in 2018.\n    def _head(s):\n        # Which facet a source answers is worth as much as its date: it tells the reader this page is\n        # the cost answer, not the regulation one.\n        bits = [f\'## {s["title"]}\', s[\'href\']]\n        if s[\'date\']: bits.append(f\'published: {s["date"]}\')\n        if (qq := [x for x in s[\'queries\'] if x != q]): bits.append(f\'answers: {"; ".join(qq)}\')\n        return \'\\n\'.join(bits)\n    digest = \'\\n\\n---\\n\\n\'.join(f\'{_head(s)}\\n\\n{s["md"]}\' for s in sources)\n    return dict(query=q, sources=sources, digest=digest, dropped=dropped, region=region,\n                curation=curation, plan=qs)', 'metadata': {'path': '/Users/71293/code/personal/orgs/ramabana/.venv/lib/python3.13/site-packages/fossick/search.py', 'uploaded_at': 1786594693.3271804, 'name': 'research', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 486, 'end_lineno': 553, 'mod_name': 'fossick.search.research', 'package': 'fossick', 'version': '0.1.11', 'docstring': 'Query -> read the top `n` *readable* results -> a cited markdown corpus. Returns {query, sources, digest, dropped, region, curation}.', 'public_api': True}, 'rank': -14.711653179413915, '_rrf_score': 0.012195121951219513, '_source': 'env', '_src_id': 'env:201673', '_score': 0.019960904593906165, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43531, 'content': 'async def selector(self:Page):\n    "Live page as a scrapling `Selector` for CSS/xpath querying"\n    return Selector(await self.html(), url=await self.eval(\'location.href\'))', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/cdp.py', 'uploaded_at': 1784774842.303492, 'name': 'selector', 'lang': '.py', 'type': 'AsyncFunctionDef', 'lineno': 139, 'end_lineno': 141, 'mod_name': 'fossick.cdp.selector', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'Live page as a scrapling `Selector` for CSS/xpath querying', 'public_api': False}, 'rank': -7.672706204781399, '_rrf_score': 0.01098901098901099, '_source': 'env', '_src_id': 'env:43531', '_score': 0.019774337658305358, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 193868, 'content': 'def _browser_fetchers():\n    """scrapling\'s browser fetchers, imported on demand.\n\n    Importing them can fail outright — scrapling generates a browser fingerprint at import time and\n    browserforge raises when it has no header set for the running platform. That surfaces as a bare\n    `ValueError` from a `from ... import ...` line, which every caller here used to treat as "no\n    results". Re-raise it as something that names the actual problem."""\n    try:\n        from scrapling.fetchers import StealthyFetcher, DynamicFetcher\n        return StealthyFetcher, DynamicFetcher\n    except Exception as e:\n        raise BrowserUnavailable(\n            f\'scrapling browser fetchers unavailable ({type(e).__name__}: {e}). \'\n            \'heavy/stealthy fetching, fetch(auto=True) escalation and google() all need them — \'\n            \'try `scrapling install`, or upgrade scrapling/browserforge.\') from e', 'metadata': {'path': '/Users/71293/code/personal/orgs/vishalakshi/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1786273543.1358001, 'name': '_browser_fetchers', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 101, 'end_lineno': 115, 'mod_name': 'fossick.core._browser_fetchers', 'package': 'fossick', 'version': '0.1.5', 'docstring': 'scrapling\'s browser fetchers, imported on demand.\n\nImporting them can fail outright — scrapling generates a browser fingerprint at import time and\nbrowserforge raises when it has no header set for the running platform. That surfaces as a bare\n`ValueError` from a `from ... import ...` line, which every caller here used to treat as "no\nresults". Re-raise it as something that names the actual problem.', 'public_api': False}, 'rank': -8.773927169544365, '_rrf_score': 0.012345679012345678, '_source': 'env', '_src_id': 'env:193868', '_score': 0.019102378170396866, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 201674, 'content': 'async def shop_products(page:Page, limit:int=40) -> list:\n    "Products on the current page as `[{i, title, price, url, add, qty, oos, vid}]`. `i` feeds `shop_add`."\n    await page.shop_prep()\n    return ifnone(await page.eval_json(\'__fk.products\', limit), [])', 'metadata': {'path': '/Users/71293/code/personal/orgs/ramabana/.venv/lib/python3.13/site-packages/fossick/shop.py', 'uploaded_at': 1786594693.333264, 'name': 'shop_products', 'lang': '.py', 'type': 'AsyncFunctionDef', 'lineno': 479, 'end_lineno': 482, 'mod_name': 'fossick.shop.shop_products', 'package': 'fossick', 'version': '0.1.11', 'docstring': 'Products on the current page as `[{i, title, price, url, add, qty, oos, vid}]`. `i` feeds `shop_add`.', 'public_api': False}, 'rank': -8.291953697203464, '_rrf_score': 0.011235955056179775, '_source': 'env', '_src_id': 'env:201674', '_score': 0.017163446507218957, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 193865, 'content': "async def calls(self:CDP, url=None, pattern='.*', tail=3):\n    'Outgoing requests matching `pattern`. Navigates if url given, else passive.'\n    page = await self.new_page()\n    rx, out = compile_pattern(pattern), {}\n    await page.network.enable()\n    async with self.on('Network.requestWillBeSent') as q:\n        async with self.on('Network.loadingFinished') as done:\n            async def drain():\n                while True:\n                    m = await q.get()\n                    if m.get('sessionId') not in (None, page.sid): continue\n                    r, rid = m['params']['request'], m['params']['requestId']\n                    if rx.search(r['url']): out[rid] = r|dict(rid=rid)\n            d = asyncio.create_task(drain())\n            if url: await page.goto(url)\n            else: await asyncio.sleep(tail)\n            await asyncio.sleep(tail)\n            d.cancel()\n            while not done.empty():\n                m = await done.get()\n                if (rid:=m['params']['requestId']) not in out: continue\n                try:\n                    body = await self.network.getResponseBody(requestId=rid, sid=page.sid)\n                    out[rid] = out[rid] | dict(response_body=body)\n                except Exception: pass  # redirect or evicted — no body available\n    return out", 'metadata': {'path': '/Users/71293/code/personal/orgs/vishalakshi/.venv/lib/python3.13/site-packages/fossick/cdp.py', 'uploaded_at': 1786273543.1295874, 'name': 'calls', 'lang': '.py', 'type': 'AsyncFunctionDef', 'lineno': 93, 'end_lineno': 118, 'mod_name': 'fossick.cdp.calls', 'package': 'fossick', 'version': '0.1.5', 'docstring': 'Outgoing requests matching `pattern`. Navigates if url given, else passive.', 'public_api': False}, 'rank': -12.014494549109804, '_rrf_score': 0.01, '_source': 'env', '_src_id': 'env:193865', '_score': 0.015275467391424873, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 202594, 'content': "def to_md(res_or_html,  # Page dict (from fetch/crawl) or raw HTML string\n          sel:str=None,  # CSS selector to extract before conversion; returns '' if no match\n          multi:bool=False,  # Return all selector matches joined\n          wrap_tag:str=None,  # Wrap each multi-result in <wrap_tag>...</wrap_tag>; only used when multi=True\n          ignore_links:bool=True,\n          ignore_images:bool=False,\n          readability:bool=True,  # False keeps the whole document, for local files and saved pages\n          ) -> str:\n    'Convert a Page dict or HTML string to clean markdown'\n    html = _html(res_or_html)\n    _md = lambda h: html2md(str(h), ignore_links, ignore_images, readability)\n    if sel:\n\t    els = Selector(html).css(sel)\n\t    mds = L(els).map(lambda m: _md(m.get()))\n\t    if multi: return '\\n'.join(mds.map(lambda m: _wrap_md(m, wrap_tag) if wrap_tag else m))\n\t    html = str(mds[0]) if mds else ''\n    return _md(html)", 'metadata': {'path': '/Users/71293/code/personal/orgs/cassowary/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1786785524.3651237, 'name': 'to_md', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 101, 'end_lineno': 117, 'mod_name': 'fossick.core.to_md', 'package': 'fossick', 'version': '0.1.12', 'docstring': 'Convert a Page dict or HTML string to clean markdown', 'public_api': True}, 'rank': -11.784226116377013, '_rrf_score': 0.008928571428571428, '_source': 'env', '_src_id': 'env:202594', '_score': 0.013832801291636805, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 193866, 'content': 'def search(\n    query:str,            # search query\n    n:int=10,             # number of results\n    region:str=\'us-en\',   # ddgs region (us-en, uk-en, ru-ru, ...)\n    google:bool=False,    # use direct Google via stealth browser instead of ddgs metasearch\n    backend:str=None,     # restrict to one ddgs backend (google, brave, duckduckgo, mojeek, yahoo, startpage)\n    pages:int=1,          # result pages to pull from each backend (a bigger candidate pool for large --n)\n    rerank:str=\'lexical\', # rerank the fused hits: lexical | flashrank | auto | none\n    as_json:bool=False,   # output JSON\n):\n    "Search the web (backends fused with RRF, or direct Google with --google); prints title, URL, and snippet."\n    results = _google(query, n=n) if google else _search(query, n=n, region=region, backend=backend,\n                                                        pages=pages, method=rerank)\n    if as_json: print(json.dumps(list(results), default=str)); return\n    for r in results:\n        print(f"## {r.get(\'title\',\'\')}")\n        print(f"URL: {r.get(\'href\',\'\')}")\n        if (body := r.get(\'body\') or r.get(\'content\')): print(body[:200])\n        print()', 'metadata': {'path': '/Users/71293/code/personal/orgs/vishalakshi/.venv/lib/python3.13/site-packages/fossick/cli.py', 'uploaded_at': 1786273543.134387, 'name': 'search', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 68, 'end_lineno': 86, 'mod_name': 'fossick.cli.search', 'package': 'fossick', 'version': '0.1.5', 'docstring': 'Search the web (backends fused with RRF, or direct Google with --google); prints title, URL, and snippet.', 'public_api': True}, 'rank': -7.098237777150365, '_rrf_score': 0.00847457627118644, '_source': 'env', '_src_id': 'env:193866', '_score': 0.012945311348665148, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 201619, 'content': 'def research(\n    query:str,            # search query\n    n:int=5,              # number of top results to read\n    google:bool=False,    # use direct Google ranking (stealth browser) instead of ddgs metasearch\n    region:str=\'auto\',    # ddgs \'country-lang\' region, or \'auto\' to read the country off the query\n    timelimit:str=None,   # d | w | m | y — only sources from the last day/week/month/year\n    sel:str=None,         # CSS selector to narrow each page before markdown\n    chars:int=4000,       # max markdown chars per source\n    pages:int=1,          # result pages to pull from each backend when searching\n    as_json:bool=False,   # output JSON\n):\n    "Search, then read the top results into one cited markdown corpus."\n    res = _research(query, n=n, engine=\'google\' if google else \'search\', region=region,\n                    timelimit=timelimit, sel=sel, chars=chars, pages=pages)\n    if as_json: print(json.dumps(res, default=str)); return\n    print(f"# Research: {res[\'query\']}\\n")\n    print(res[\'digest\'])\n    for d in res[\'dropped\']: print(f"\\n<!-- skipped {d[\'href\']} — {d[\'reason\']} -->")', 'metadata': {'path': '/Users/71293/code/personal/orgs/ramabana/.venv/lib/python3.13/site-packages/fossick/cli.py', 'uploaded_at': 1786594693.3177588, 'name': 'research', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 355, 'end_lineno': 372, 'mod_name': 'fossick.cli.research', 'package': 'fossick', 'version': '0.1.11', 'docstring': 'Search, then read the top results into one cited markdown corpus.', 'public_api': True}, 'rank': -7.487563104279774, '_rrf_score': 0.008130081300813009, '_source': 'env', '_src_id': 'env:201619', '_score': 0.012419079180020222, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192745, 'content': 'async def eval_json(page:Page, fn:str, *args):\n    "Call JavaScript function `fn` with JSON-safe `args`, returning its decoded result."\n    return await page.eval(f\'({fn})({",".join(json.dumps(o, default=str) for o in args)})\')', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/cdp.py', 'uploaded_at': 1785977784.1497684, 'name': 'eval_json', 'lang': '.py', 'type': 'AsyncFunctionDef', 'lineno': 256, 'end_lineno': 258, 'mod_name': 'fossick.cdp.eval_json', 'package': 'fossick', 'version': '0.1.1', 'docstring': 'Call JavaScript function `fn` with JSON-safe `args`, returning its decoded result.', 'public_api': False}, 'rank': -7.539287328883038, '_rrf_score': 0.007575757575757576, '_source': 'env', '_src_id': 'env:192745', '_score': 0.011964023077377912, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 201625, 'content': "def fetch(url:str,                # URL to fetch\n          sel:str=None,           # CSS selector to extract (None = full page)\n          method:str='GET',       # HTTP method; 'POST' sends payload as JSON body\n          payload:dict=None,      # POST body (JSON) or GET query params\n          heavy:bool=False,       # Full JS rendering via headless browser\n          stealthy:bool=False,    # Anti-bot stealth fetcher (Cloudflare etc.)\n          capture_xhr:bool=False, # Capture XHR calls made by the page (only works with heavy or stealthy)\n          session:bool=False,     # Route through the persistent debug Chrome (reuses its logged-in cookies); True or a port int\n          auto:bool=False,        # Auto-escalate plain->heavy->stealthy->session on bot-block detection\n          allow_private:bool=None,# Allow localhost/intranet targets (default: refuse; $FOSSICK_ALLOW_PRIVATE=1)\n          **kw,                   # Extra kwargs passed to scrapling (e.g. verify, headers)\n         ) -> Response:\n    'Fetch `url`, return a Response with `.html` (raw or `sel`-extracted), `.xhr`, `.status`.'\n    if auto: return _fetch_auto(url, sel=sel, method=method, payload=payload, capture_xhr=capture_xhr,\n                            allow_private=allow_private, **kw)\n    heavy = heavy or capture_xhr\n    if capture_xhr: kw.setdefault('capture_xhr', '.*')\n    page = get_page(url, method=method, payload=payload, heavy=heavy, stealthy=stealthy, session=session,\n                    allow_private=allow_private, **kw)\n    return _finish(page, sel, capture_xhr)", 'metadata': {'path': '/Users/71293/code/personal/orgs/ramabana/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1786594693.3198924, 'name': 'fetch', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 261, 'end_lineno': 280, 'mod_name': 'fossick.core.fetch', 'package': 'fossick', 'version': '0.1.11', 'docstring': 'Fetch `url`, return a Response with `.html` (raw or `sel`-extracted), `.xhr`, `.status`.', 'public_api': True}, 'rank': -22.73756535891292, '_rrf_score': 0.015625, '_source': 'env', '_src_id': 'env:201625', '_score': 0.01171875, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192754, 'content': 'def crawl(start_url:str,               # URL to start from\n          sel:str=None,                # CSS selector to extract per page\n          follow_sel:str=\'a[href]\',    # CSS selector for links to follow\n          same_domain:bool=True,       # Only follow links on same domain\n          max_pages:int=10,            # Max pages to visit\n          delay:float=0,               # Seconds to wait between requests (polite crawling)\n          heavy:bool=False, # Full JS rendering via headless browser\n          stealthy:bool=False, # Anti-bot stealth fetcher (Cloudflare etc.)\n          reuse:bool=True,  # For heavy/stealthy crawls, keep one browser open across pages\n          **kw,                        # Extra kwargs passed to scrapling (e.g. verify, timeout)\n         ) -> list:\n    \'Crawl from `start_url`, following `follow_sel` links, return list of Page dicts\'\n    bd, v, q, res = urlparse(start_url).netloc, set(), deque([start_url]), []\n    use_sess = reuse and (heavy or stealthy)\n    ctx = browser_session(stealthy=stealthy, **kw) if use_sess else nullcontext(None)\n    with ctx as sess:\n        while q and len(v) < max_pages:\n            url = q.popleft()\n            if url in v: continue\n            v.add(url)\n            if delay and res: time.sleep(delay)\n            try:\n                pg: Response = sess(url) if sess else fetch(url, heavy=heavy, stealthy=stealthy, **kw)\n                if pg.status != 200: continue\n                if sel: pg.html = str(el.get()) if (el:= pg.css(sel).first) else \'\'\n                res.append(pg)\n                for link in pg.css(follow_sel):\n                    href = link.attrib.get(\'href\', \'\')\n                    if not href or href.startswith((\'#\', \'javascript:\', \'mailto:\')): continue\n                    abs_url = pg.urljoin(href)\n                    if same_domain and urlparse(abs_url).netloc != bd: continue\n                    if abs_url not in v: q.append(abs_url)\n            except Exception as ex:\n                print(f"Error fetching {url}: {ex}. Continuing crawl.")\n                continue\n    return res', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1785977784.1587985, 'name': 'crawl', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 205, 'end_lineno': 240, 'mod_name': 'fossick.core.crawl', 'package': 'fossick', 'version': '0.1.1', 'docstring': 'Crawl from `start_url`, following `follow_sel` links, return list of Page dicts', 'public_api': True}, 'rank': -17.340028390063267, '_rrf_score': 0.015151515151515152, '_source': 'env', '_src_id': 'env:192754', '_score': 0.011363636363636364, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 194294, 'content': 'def paginate_api(url:str,                      # API endpoint URL\n                 payload:dict=None,            # Request body (POST) or params (GET)\n                 page_field:str=\'pageNumber\',  # Payload key to increment for each page\n                 size_field:str=\'pageSize\',    # Payload key for page size (detects last page)\n                 results_field:str=None,       # Response key with items list (auto-detect if None)\n                 method:str=\'POST\',            # HTTP method\n                 max_pages:int=10,\n                 page_size=24,                 # Page size to request (only used if not in payload)\n                 page_start:int=1,             # Starting page number (default 1)\n                 save:bool=False,              # If True, saves each page\'s items to disk\n                 save_file:str=\'{url}_page_{page}.json\', # Filepath pattern for saving (only used if save=True)\n                 force:bool=False,             # If True, forces re-fetching even if saved file exists\n                 **kw,                         # Extra kwargs passed to fetch() (verify, headers, etc.)\n                ) -> list:\n    "Paginate through a JSON API, collecting all results. Auto-detects the items list in response."\n    payload = dict(payload or {})\n    page_size = payload.get(size_field, page_size)\n    all_items = []\n    for page_num in range(page_start, max_pages + page_start):\n        sf = save_file.format(url=urlparse(url).netloc, page=page_num)\n        if not force and Path(sf).exists():\n            print(f"Page {page_num} already saved, skipping fetch")\n            with open(sf) as f: all_items.extend(json.load(f)); continue\n        if (pg:= fetch(url, method=method, payload={**payload, page_field: page_num}, **kw)).status != 200: break\n        try: data = pg.json()\n        except: break\n        if isinstance(data, list): items = data\n        elif results_field and (raw := data.get(results_field)) is not None:\n            items = list(raw.values()) if isinstance(raw, dict) else raw\n        else:\n            # a named field that is not at the top level is still a named field: look for it, and\n            # for records generally, anywhere in the payload before falling back to the shallow rule\n            items = json_records(data, min_len=1, key=results_field)\n            if not items:\n                lists = {k: v for k, v in data.items() if isinstance(v, list)}\n                items = max(lists.values(), key=len) if lists else []\n        if not items: break\n        all_items.extend(items)\n        if len(items) < page_size: break\n        if save: Path(sf).write_text(json.dumps(items, indent=2))\n    return all_items', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1786332098.5035472, 'name': 'paginate_api', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 701, 'end_lineno': 741, 'mod_name': 'fossick.core.paginate_api', 'package': 'fossick', 'version': '0.1.7', 'docstring': 'Paginate through a JSON API, collecting all results. Auto-detects the items list in response.', 'public_api': True}, 'rank': -15.202545482454132, '_rrf_score': 0.014925373134328358, '_source': 'env', '_src_id': 'env:194294', '_score': 0.011194029850746268, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43621, 'content': 'def fetch_pages(urls:list, sel:str|None=None, max_chars:int=4000) -> list:\n    "Fetch many known URLs in parallel; returns [{url, status, markdown}] in the same order as urls."\n    return [{\'url\': p[\'url\'], \'status\': p[\'status\'], \'markdown\': _trunc(to_md(p, sel=sel), max_chars)}\n            for p in _fetch_all(urls, sel=sel)]', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/mcp.py', 'uploaded_at': 1784774842.3057618, 'name': 'fetch_pages', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 77, 'end_lineno': 80, 'mod_name': 'fossick.mcp.fetch_pages', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'Fetch many known URLs in parallel; returns [{url, status, markdown}] in the same order as urls.', 'public_api': True}, 'rank': -16.19031608861205, '_rrf_score': 0.014285714285714285, '_source': 'env', '_src_id': 'env:43621', '_score': 0.010714285714285714, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43546, 'content': 'def crawl(\n    url:str,                    # start URL\n    sel:str=None,              # CSS selector to extract per page\n    follow_sel:str=\'a[href]\',  # CSS selector for links to follow\n    max_pages:int=10,          # max pages to visit\n    same_domain:bool=True,     # only follow same-domain links\n    heavy:bool=False,          # JS rendering via headless browser\n    stealthy:bool=False,       # anti-bot stealth fetcher\n    as_json:bool=False,        # output JSON\n):\n    "Crawl from a URL following links; prints each page as markdown."\n    pages = _crawl(url, sel=sel, follow_sel=follow_sel, max_pages=max_pages,\n                   same_domain=same_domain, heavy=heavy, stealthy=stealthy)\n    if as_json: print(json.dumps([{\'url\': p[\'url\'], \'status\': p[\'status\'], \'content\': to_md(p, sel=sel)} for p in pages], default=str))\n    else:\n        for p in pages:\n            print(f"# {p[\'url\']}")\n            print(to_md(p, sel=sel))\n            print()', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/cli.py', 'uploaded_at': 1784774842.3050234, 'name': 'crawl', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 44, 'end_lineno': 62, 'mod_name': 'fossick.cli.crawl', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'Crawl from a URL following links; prints each page as markdown.', 'public_api': True}, 'rank': -13.094051409706035, '_rrf_score': 0.013888888888888888, '_source': 'env', '_src_id': 'env:43546', '_score': 0.010416666666666666, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192786, 'content': 'async def shop_lines(page:Page) -> list:\n    "Cart line items read from the page: `[{i, text, price, qty, remove}]`. Run this on the cart page."\n    await page.shop_prep()\n    return await page.eval(\'__fk.lines()\') or []', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/shop.py', 'uploaded_at': 1785977784.1639626, 'name': 'shop_lines', 'lang': '.py', 'type': 'AsyncFunctionDef', 'lineno': 494, 'end_lineno': 497, 'mod_name': 'fossick.shop.shop_lines', 'package': 'fossick', 'version': '0.1.1', 'docstring': 'Cart line items read from the page: `[{i, text, price, qty, remove}]`. Run this on the cart page.', 'public_api': False}, 'rank': -8.553133797641046, '_rrf_score': 0.013513513513513514, '_source': 'env', '_src_id': 'env:192786', '_score': 0.010135135135135136, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 201628, 'content': 'def research(query:str, n:int=5, google:bool=False, region:str=\'auto\', timelimit:str|None=None,\n             sel:str|None=None, chars:int=4000) -> dict:\n    "Question -> cited answer: search, read the top n *readable* results in parallel (auto-escalating past bot walls, skipping bot-wall/empty pages and backfilling), return {query, sources, digest, dropped, region}. digest is cited markdown, one ## section per source, trimmed to the passages that answer the query and headed with the date the source says it was published. region is a ddgs \'country-lang\' code, or \'auto\' to read the country off the query — set it whenever the answer depends on jurisdiction or market. timelimit d|w|m|y keeps only recent sources."\n    return _jsonable(_research(query, n=n, engine=\'google\' if google else \'search\', region=region,\n                               timelimit=timelimit, sel=sel, chars=chars))', 'metadata': {'path': '/Users/71293/code/personal/orgs/ramabana/.venv/lib/python3.13/site-packages/fossick/mcp.py', 'uploaded_at': 1786594693.32086, 'name': 'research', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 66, 'end_lineno': 70, 'mod_name': 'fossick.mcp.research', 'package': 'fossick', 'version': '0.1.11', 'docstring': "Question -> cited answer: search, read the top n *readable* results in parallel (auto-escalating past bot walls, skipping bot-wall/empty pages and backfilling), return {query, sources, digest, dropped, region}. digest is cited markdown, one ## section per source, trimmed to the passages that answer the query and headed with the date the source says it was published. region is a ddgs 'country-lang' code, or 'auto' to read the country off the query — set it whenever the answer depends on jurisdiction or market. timelimit d|w|m|y keeps only recent sources.", 'public_api': True}, '_dist': 0.8210081458091736, '_rrf_score': 0.006578947368421052, '_source': 'env', '_src_id': 'env:201628', '_score': 0.010049649599621626, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192775, 'content': 'def shop_cart(page:bool=False) -> dict:\n    "The current cart. page=True opens the cart page first, which is what gives you the line indexes shop_line needs."\n    s = _sh()\n    return _jsonable(s.cart_page() if page else s.cart(lines=True))', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/mcp.py', 'uploaded_at': 1785977784.1604002, 'name': 'shop_cart', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 270, 'end_lineno': 273, 'mod_name': 'fossick.mcp.shop_cart', 'package': 'fossick', 'version': '0.1.1', 'docstring': 'The current cart. page=True opens the cart page first, which is what gives you the line indexes shop_line needs.', 'public_api': True}, 'rank': -8.40881532248973, '_rrf_score': 0.0125, '_source': 'env', '_src_id': 'env:192775', '_score': 0.009375000000000001, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 193909, 'content': 'def usable(pg, min_chars:int=_MIN_CHARS) -> bool:\n    "Did this fetch return a readable page, or an error / bot wall / empty JS shell?"\n    return pg is not None and getattr(pg, \'status\', 0) == 200 \\\n        and getattr(pg, \'tier\', \'\') != \'blocked\' and not _blocked(pg)', 'metadata': {'path': '/Users/71293/code/personal/orgs/vishalakshi/.venv/lib/python3.13/site-packages/fossick/search.py', 'uploaded_at': 1786273543.1417477, 'name': 'usable', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 278, 'end_lineno': 281, 'mod_name': 'fossick.search.usable', 'package': 'fossick', 'version': '0.1.5', 'docstring': 'Did this fetch return a readable page, or an error / bot wall / empty JS shell?', 'public_api': True}, 'rank': -14.069860121787508, '_rrf_score': 0.011764705882352941, '_source': 'env', '_src_id': 'env:193909', '_score': 0.008823529411764706, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 201667, 'content': 'def _google(q:str, n:int, lang:str, gl:str=None, tbs:str=None) -> list:\n    "Stealth-browser Google fetch+parse, memoized. Raises on empty so throttled/failed calls aren\'t cached."\n    # `gl` biases ranking towards a country; `cr` would *exclude* everything else, which throws away\n    # the authoritative .com page about Australian rules along with the noise.\n    url = f\'https://www.google.com/search?q={quote_plus(q)}&hl={lang}&num={n+2}\'\n    if gl: url += f\'&gl={gl}\'\n    if tbs: url += f\'&tbs={tbs}\'\n    pg = fetch(url, stealthy=True, network_idle=True)\n    res = _parse_google(pg.html_content, n) if pg.status == 200 else []\n    if not res: raise ValueError(\'no google results\')\n    return res', 'metadata': {'path': '/Users/71293/code/personal/orgs/ramabana/.venv/lib/python3.13/site-packages/fossick/search.py', 'uploaded_at': 1786594693.3271804, 'name': '_google', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 341, 'end_lineno': 351, 'mod_name': 'fossick.search._google', 'package': 'fossick', 'version': '0.1.11', 'docstring': "Stealth-browser Google fetch+parse, memoized. Raises on empty so throttled/failed calls aren't cached.", 'public_api': False}, 'rank': -11.945649847631184, '_rrf_score': 0.011111111111111112, '_source': 'env', '_src_id': 'env:201667', '_score': 0.008333333333333333, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43544, 'content': 'async def annotate(page:Page, save_dir=None, prefix=None, stop=None, tout=None):\n\t"Click elements interactively; returns (screenshot, [{n, role, name, selector}]) with badges baked in."\n\tstop, sid = stop or asyncio.Event(), page.sid\n\tprefix = prefix or await page.eval(\'document.title\')\n\tsave_dir = Path(save_dir) if save_dir else None\n\tif save_dir: save_dir.mkdir(parents=True, exist_ok=True)\n\tawait page.runtime.enable()\n\tawait page.accessibility.enable()\n\tawait page.runtime.addBinding(name=\'__axsel\')\n\tident = await page.page.addScriptToEvaluateOnNewDocument(source=ANNOTATE_JS)\n\tawait page.eval(ANNOTATE_JS)\n\tawait page.eval(ANNOTATE_BAR_JS)\n\tloop = asyncio.get_event_loop()\n\tdl = loop.time() + tout if tout else None\n\tselected, img = [], None\n\ttry:\n\t\tasync with page.cdp.on(\'Runtime.bindingCalled\') as q:\n\t\t\tprint(\'Click elements to annotate, ✓ Done when ready.\')\n\t\t\twhile not stop.is_set():\n\t\t\t\trem = (dl - loop.time()) if dl else None\n\t\t\t\tif rem is not None and rem <= 0: break\n\t\t\t\tget, halt = asyncio.ensure_future(q.get()), asyncio.ensure_future(stop.wait())\n\t\t\t\tfin, pend = await asyncio.wait({get, halt}, return_when=asyncio.FIRST_COMPLETED,timeout=rem)\n\t\t\t\tfor p in pend: p.cancel()\n\t\t\t\tif get not in fin: break\n\t\t\t\tm = get.result()\n\t\t\t\tif m.get(\'sessionId\') not in (None, sid) or m[\'params\'].get(\'name\') != \'__axsel\': continue\n\t\t\t\tdata = json.loads(m[\'params\'].get(\'payload\') or \'{}\')\n\t\t\t\tif data.get(\'done\'): break\n\t\t\t\tn, sel = data[\'n\'], data[\'sel\']\n\t\t\t\tloc = await page.DOM.getNodeForLocation(x=int(data[\'x\']), y=int(data[\'y\']),\n\t\t\t\t\tincludeUserAgentShadowDOM=False)\n\t\t\t\tnodes = await page.accessibility.getAXNodeAndAncestors(backendNodeId=loc[\'backendNodeId\'])\n\t\t\t\trole = next((nd[\'role\'][\'value\'] for nd in nodes\n\t\t\t\t             if nd.get(\'role\',{}).get(\'value\',\'\') not in _AX_SKIP), \'element\')\n\t\t\t\tname = next((nd[\'name\'][\'value\'] for nd in nodes\n\t\t\t\t             if nd.get(\'name\',{}).get(\'value\',\'\')), \'\')\n\t\t\t\tawait page.eval(f"window.__axBadge({json.dumps(n)},{json.dumps(role)},{json.dumps(sel)})")\n\t\t\t\tselected.append({\'n\': n, \'role\': role, \'name\': name, \'selector\': sel})\n\t\timg = await page.screenshot()\n\tfinally:\n\t\ttry:\n\t\t\tawait page.page.removeScriptToEvaluateOnNewDocument(identifier=ident)\n\t\t\tawait page.runtime.removeBinding(name=\'__axsel\')\n\t\t\tawait page.eval(ANNOTATE_CLEANUP_JS)\n\t\texcept Exception: pass\n\tif save_dir and img:\n\t\tf = save_dir / f\'{prefix}.png\'\n\t\tf.write_bytes(img.data)\n\t\tprint(\'saved\', f.name)\n\treturn img, selected', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/cdp.py', 'uploaded_at': 1784774842.303492, 'name': 'annotate', 'lang': '.py', 'type': 'AsyncFunctionDef', 'lineno': 422, 'end_lineno': 472, 'mod_name': 'fossick.cdp.annotate', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'Click elements interactively; returns (screenshot, [{n, role, name, selector}]) with badges baked in.', 'public_api': False}, 'rank': -11.481091605696575, '_rrf_score': 0.010752688172043012, '_source': 'env', '_src_id': 'env:43544', '_score': 0.008064516129032258, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 193879, 'content': "def url2nb(url, # URL to convert (PDF, arxiv, or HTML page)\n           nb_path=None, # optional path to save the notebook; defaults to <url-stem>.ipynb\n           force=False, # if True, forces re-download and conversion even if notebook exists\n           **kwargs # extra kwargs passed to fetch() (e.g. verify=False to skip SSL verification)\n)-> Path:\n\t'Convert any URL (PDF, arxiv, or HTML page) to a Jupyter notebook'\n\tif url.endswith('.pdf'): return pdf2nb(url, nb_path=nb_path, **kwargs)\n\tnb_path = Path(nb_path or f'{_nb_stem(url)}.ipynb')\n\tif nb_path.exists() and not force: return nb_path\n\tif 'arxiv.org' in url:\n\t\tres = read_arxiv(url, save_dir=str(nb_path.parent))\n\t\treturn pdf2nb(res['pdf_path'], nb_path=nb_path, force=force, **kwargs)\n\tmd = clean_md(to_md(fetch(url, **kwargs)))\n\tNotebook(new_nb(cells=_md2cells(md)), path=nb_path).save()\n\treturn nb_path", 'metadata': {'path': '/Users/71293/code/personal/orgs/vishalakshi/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1786273543.1358001, 'name': 'url2nb', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 521, 'end_lineno': 535, 'mod_name': 'fossick.core.url2nb', 'package': 'fossick', 'version': '0.1.5', 'docstring': 'Convert any URL (PDF, arxiv, or HTML page) to a Jupyter notebook', 'public_api': True}, 'rank': -13.0107697454001, '_rrf_score': 0.008849557522123894, '_source': 'env', '_src_id': 'env:193879', '_score': 0.00663716814159292, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192743, 'content': 'class JSError(RuntimeError): "JavaScript raised while evaluating an expression in the page."', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/cdp.py', 'uploaded_at': 1785977784.1497684, 'name': 'JSError', 'lang': '.py', 'type': 'ClassDef', 'lineno': 68, 'end_lineno': 68, 'mod_name': 'fossick.cdp.JSError', 'package': 'fossick', 'version': '0.1.1', 'docstring': 'JavaScript raised while evaluating an expression in the page.', 'public_api': True}, 'rank': -7.218769223662956, '_rrf_score': 0.007462686567164179, '_source': 'env', '_src_id': 'env:192743', '_score': 0.005597014925373134, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192753, 'content': 'def browser_session(stealthy:bool=False, headless:bool=True, **init_kw):\n    "Open ONE browser and yield a `fetch(url, sel=None, **)` func that reuses it across calls (skips per-URL relaunch)."\n    from scrapling.fetchers import AsyncDynamicSession, AsyncStealthySession\n    init_kw.pop(\'verify\', None)\n    Sess = AsyncStealthySession if stealthy else AsyncDynamicSession\n    s = Sess(headless=headless, **init_kw)\n    syncy(s.__aenter__())\n    try:\n        def _f(u, sel=None, **fkw): return _finish(syncy(s.fetch(u, **fkw)), sel)\n        yield _f\n    finally: syncy(s.__aexit__(None, None, None))', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1785977784.1587985, 'name': 'browser_session', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 191, 'end_lineno': 201, 'mod_name': 'fossick.core.browser_session', 'package': 'fossick', 'version': '0.1.1', 'docstring': 'Open ONE browser and yield a `fetch(url, sel=None, **)` func that reuses it across calls (skips per-URL relaunch).', 'public_api': True}, 'rank': -15.27923561978961, '_rrf_score': 0.014705882352941176, '_source': 'env', '_src_id': 'env:192753', '_score': 0.005514705882352941, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 201329, 'content': 'def _fetch_auto(url, sel=None, tiers=_AUTO_TIERS, **kw) -> Response:\n    """Escalate plain -> heavy -> stealthy -> stealthy+debug-Chrome until a non-blocked response; sets `.tier`.\n\n    Every tier\'s failure is recorded on `.errs`. When the browser tiers can\'t even load, escalation\n    silently degrades to "plain HTTP only" and the caller gets the bot wall back as if it were the\n    page — so say it once, loudly, rather than returning a challenge page that looks like content."""\n    page, errs = None, []\n    for opts in tiers:\n        tier = \'+\'.join(k for k, v in opts.items() if v) or \'plain\'\n        try: page = fetch(url, sel=sel, **opts, **kw)\n        except BrowserUnavailable as e:\n            errs.append((tier, str(e)))\n            _warn_once(\'browser\', str(e))\n            continue\n        except Exception as e:\n            errs.append((tier, f\'{type(e).__name__}: {e}\')); continue\n        if not _blocked(page):\n            setattr(page, \'tier\', tier)\n            setattr(page, \'errs\', errs)\n            return page\n        errs.append((tier, \'blocked\'))\n    if page is not None:\n        setattr(page, \'tier\', \'blocked\')\n        setattr(page, \'errs\', errs)\n    return page', 'metadata': {'path': '/Users/71293/code/personal/orgs/ramabana/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1786512967.331086, 'name': '_fetch_auto', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 187, 'end_lineno': 211, 'mod_name': 'fossick.core._fetch_auto', 'package': 'fossick', 'version': '0.1.10', 'docstring': 'Escalate plain -> heavy -> stealthy -> stealthy+debug-Chrome until a non-blocked response; sets `.tier`.\n\nEvery tier\'s failure is recorded on `.errs`. When the browser tiers can\'t even load, escalation\nsilently degrades to "plain HTTP only" and the caller gets the bot wall back as if it were the\npage — so say it once, loudly, rather than returning a challenge page that looks like content.', 'public_api': False}, 'rank': -15.64352995196204, '_rrf_score': 0.014492753623188406, '_source': 'env', '_src_id': 'env:201329', '_score': 0.005434782608695652, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43604, 'content': 'def find_xhr(url:str,             # URL to visit with browser\n             pattern:str=\'*\',     # Glob or regex pattern to filter captured XHR URLs\n             session:bool=False,  # Use the persistent debug Chrome (authenticated) instead of a throwaway browser\n             port:int=9223,       # debug Chrome port (only when session=True)\n             tail:int=3,          # seconds to keep listening after navigation (only when session=True)\n             **kw,                # Extra kwargs passed to fetch (verify, network_idle, etc.)\n            ) -> list:\n    "Visit `url`, return [{url, content_type, data, capture}] for each XHR/fetch call made. `capture` replays via `replay_xhr`."\n    rx = compile_pattern(pattern)\n    if session:\n        from fossick.cdp import cdp_connect, syncy\n        async def _run():\n            cdp = await cdp_connect(port=port)\n            return await cdp.calls(url=url, pattern=pattern, tail=tail)\n        out = []\n        for r in syncy(_run()).values():\n            rb = r.get(\'response_body\')\n            if isinstance(rb, dict):\n                rb = base64.b64decode(rb.get(\'body\',\'\')).decode(errors=\'replace\') if rb.get(\'base64Encoded\') else rb.get(\'body\',\'\')\n            try: d = json.loads(rb) if rb else None\n            except Exception: d = rb\n            ct = {k.lower(): v for k, v in (r.get(\'headers\') or {}).items()}.get(\'content-type\', \'\')\n            out.append(AttrDict(url=r[\'url\'], content_type=ct, data=d, capture=r))\n        return out\n    kw.setdefault(\'network_idle\', True)\n    xhr = fetch(url, capture_xhr=True, **kw).xhr\n    return [x for x in xhr if rx.search(x.url)]', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1784774842.3054273, 'name': 'find_xhr', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 470, 'end_lineno': 496, 'mod_name': 'fossick.core.find_xhr', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'Visit `url`, return [{url, content_type, data, capture}] for each XHR/fetch call made. `capture` replays via `replay_xhr`.', 'public_api': True}, 'rank': -14.03362281629975, '_rrf_score': 0.014084507042253521, '_source': 'env', '_src_id': 'env:43604', '_score': 0.00528169014084507, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 202593, 'content': "def html2md(s:str, ignore_links=True, ignore_images=False,\n            readability:bool=True,  # Extract the article (right for fetched pages); False converts the whole document\n           ):\n    'Convert `s` from HTML to markdown'\n    import html2text\n    from readability import Document\n    if not s.strip(): return ''\n    o = html2text.HTML2Text(bodywidth=5000)\n    o.ignore_links, o.mark_code, o.ignore_images = ignore_links, True, ignore_images\n    return o.handle(Document(s).summary(keep_all_images=True) if readability else _clean_html(s))", 'metadata': {'path': '/Users/71293/code/personal/orgs/cassowary/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1786785524.3651237, 'name': 'html2md', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 90, 'end_lineno': 99, 'mod_name': 'fossick.core.html2md', 'package': 'fossick', 'version': '0.1.12', 'docstring': 'Convert `s` from HTML to markdown', 'public_api': True}, 'rank': -10.433179343856812, '_rrf_score': 0.007042253521126761, '_source': 'env', '_src_id': 'env:202593', '_score': 0.00528169014084507, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43632, 'content': 'def find_hidden_apis(url:str, pattern:str=\'*\', session:bool=False, preview_chars:int=500) -> list:\n    "Visit a page with a browser and capture the JSON/XHR API calls it makes (glob/regex pattern filters URLs). session=True captures through the logged-in debug Chrome. Each hit has a capture_id for replay_capture."\n    hits = _find_xhr(url, pattern=pattern, session=session)\n    _stash([h.get(\'capture\') for h in hits])\n    return [{\'capture_id\': i, \'url\': h[\'url\'], \'content_type\': h.get(\'content_type\'),\n             \'data_preview\': _trunc(json.dumps(_jsonable(h.get(\'data\')), default=str), preview_chars)}\n            for i, h in enumerate(hits)]', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/mcp.py', 'uploaded_at': 1784774842.3057618, 'name': 'find_hidden_apis', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 146, 'end_lineno': 152, 'mod_name': 'fossick.mcp.find_hidden_apis', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'Visit a page with a browser and capture the JSON/XHR API calls it makes (glob/regex pattern filters URLs). session=True captures through the logged-in debug Chrome. Each hit has a capture_id for replay_capture.', 'public_api': True}, 'rank': -13.689984325923723, '_rrf_score': 0.013333333333333334, '_source': 'env', '_src_id': 'env:43632', '_score': 0.005, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43556, 'content': 'def paginate_api(\n    url:str,                      # API endpoint URL\n    payload:str=None,             # request body/params as a JSON string\n    page_field:str=\'pageNumber\',  # payload key to increment per page\n    results_field:str=None,       # response key with the items list (auto-detect if None)\n    method:str=\'POST\',            # HTTP method\n    max_pages:int=10,             # max pages to fetch\n):\n    "Paginate a JSON API and print all collected items as JSON."\n    body = json.loads(payload) if payload else None\n    items = _paginate_api(url, payload=body, page_field=page_field,\n                          results_field=results_field, method=method, max_pages=max_pages)\n    print(json.dumps(items, default=str))', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/cli.py', 'uploaded_at': 1784774842.3050234, 'name': 'paginate_api', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 218, 'end_lineno': 230, 'mod_name': 'fossick.cli.paginate_api', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'Paginate a JSON API and print all collected items as JSON.', 'public_api': True}, 'rank': -13.469040914474467, '_rrf_score': 0.012048192771084338, '_source': 'env', '_src_id': 'env:43556', '_score': 0.004518072289156627, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192787, 'content': 'async def shop_blockers(page:Page) -> list:\n    "What is standing in the way: `cookie-banner`, `location-required`, `login-required`, `captcha`, `modal-open`."\n    await page.shop_prep()\n    return await page.eval(\'__fk.blockers()\') or []', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/shop.py', 'uploaded_at': 1785977784.1639626, 'name': 'shop_blockers', 'lang': '.py', 'type': 'AsyncFunctionDef', 'lineno': 500, 'end_lineno': 503, 'mod_name': 'fossick.shop.shop_blockers', 'package': 'fossick', 'version': '0.1.1', 'docstring': 'What is standing in the way: `cookie-banner`, `location-required`, `login-required`, `captcha`, `modal-open`.', 'public_api': False}, 'rank': -7.92902183749531, '_rrf_score': 0.011904761904761904, '_source': 'env', '_src_id': 'env:192787', '_score': 0.004464285714285714, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192773, 'content': 'def shop_products(limit:int=25) -> list:\n    "Products on the page the session is currently on, numbered for shop_add."\n    return _jsonable(_sh().products(limit))', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/mcp.py', 'uploaded_at': 1785977784.1604002, 'name': 'shop_products', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 260, 'end_lineno': 262, 'mod_name': 'fossick.mcp.shop_products', 'package': 'fossick', 'version': '0.1.1', 'docstring': 'Products on the page the session is currently on, numbered for shop_add.', 'public_api': True}, 'rank': -14.052574016101026, '_rrf_score': 0.011494252873563218, '_source': 'env', '_src_id': 'env:192773', '_score': 0.004310344827586207, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43540, 'content': 'async def click_sel(page:Page, sel:str, wait=True):\n    "Click the element matching CSS `sel`."\n    bid = await page.node_for(sel)\n    if bid is None: raise ValueError(f\'No element matching {sel!r}\')\n    await (page.click_and_wait(bid) if wait else page.click(bid))', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/cdp.py', 'uploaded_at': 1784774842.303492, 'name': 'click_sel', 'lang': '.py', 'type': 'AsyncFunctionDef', 'lineno': 226, 'end_lineno': 230, 'mod_name': 'fossick.cdp.click_sel', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'Click the element matching CSS `sel`.', 'public_api': False}, 'rank': -8.152051839308514, '_rrf_score': 0.00980392156862745, '_source': 'env', '_src_id': 'env:43540', '_score': 0.003676470588235294, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 193912, 'content': 'def _fetch_batch(urls:list, concurrency:int=8, **kw) -> list:\n    "fetch() each url in parallel, returning None where it raised — one dead host mustn\'t kill the batch."\n    def _one(u):\n        try: return fetch(u, **kw)\n        except Exception: return None\n    return list(parallel(_one, urls, n_workers=min(concurrency, len(urls) or 1), threadpool=True))', 'metadata': {'path': '/Users/71293/code/personal/orgs/vishalakshi/.venv/lib/python3.13/site-packages/fossick/search.py', 'uploaded_at': 1786273543.1417477, 'name': '_fetch_batch', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 327, 'end_lineno': 332, 'mod_name': 'fossick.search._fetch_batch', 'package': 'fossick', 'version': '0.1.5', 'docstring': "fetch() each url in parallel, returning None where it raised — one dead host mustn't kill the batch.", 'public_api': False}, 'rank': -8.813468346457274, '_rrf_score': 0.009708737864077669, '_source': 'env', '_src_id': 'env:193912', '_score': 0.003640776699029126, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 201658, 'content': 'def _backend_lists(q:str, backends, region=\'us-en\', timeout=8, pages=1, **kw) -> list:\n    "One rank-ordered list per backend, pulling `pages` result pages from each. Every (backend, page) runs in parallel."\n    jobs = [(be, pg) for be in backends for pg in range(1, pages+1)]\n    if not jobs: return []\n    got = parallel(lambda j: (j, _engine_hits(j[0], q, region=region, page=j[1], timeout=timeout, **kw)),\n                   jobs, n_workers=len(jobs), threadpool=True)\n    by_be = {}\n    for (be, pg), hits in got: by_be.setdefault(be, {})[pg] = hits\n    out = []\n    for be in backends:\n        merged, seen = [], set()\n        for pg in sorted(by_be.get(be, {})):\n            for h in by_be[be][pg]:\n                if (key := norm_url(h.get(\'href\') or h.get(\'url\') or \'\')) in seen: continue  # page 2 repeats page 1\n                seen.add(key)\n                merged.append(dict(h, rank=len(merged)+1))    # rank runs on across pages, so RRF sees one list\n        if merged: out.append(merged)\n    return out', 'metadata': {'path': '/Users/71293/code/personal/orgs/ramabana/.venv/lib/python3.13/site-packages/fossick/search.py', 'uploaded_at': 1786594693.3271804, 'name': '_backend_lists', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 65, 'end_lineno': 82, 'mod_name': 'fossick.search._backend_lists', 'package': 'fossick', 'version': '0.1.11', 'docstring': 'One rank-ordered list per backend, pulling `pages` result pages from each. Every (backend, page) runs in parallel.', 'public_api': False}, 'rank': -8.444303779953932, '_rrf_score': 0.00909090909090909, '_source': 'env', '_src_id': 'env:201658', '_score': 0.003409090909090909, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192744, 'content': 'async def eval(self:CDP, expr:str, sid=None):\n    """Evaluate `expr` in the page and return its decoded value.\n\n    Upgrades fastcdp\'s `eval` in three ways every caller wants: promises are awaited, objects and\n    arrays come back as Python values instead of `None`, and a JS exception is raised as `JSError`\n    rather than silently read as an empty result."""\n    r = await self.runtime.evaluate(sid=sid, expression=expr, awaitPromise=True, returnByValue=True)\n    if \'exceptionDetails\' in r:\n        d = r[\'exceptionDetails\']\n        raise JSError(nested_idx(d, \'exception\', \'description\') or d.get(\'text\') or \'JS error\')\n    return r.get(\'value\')', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/cdp.py', 'uploaded_at': 1785977784.1497684, 'name': 'eval', 'lang': '.py', 'type': 'AsyncFunctionDef', 'lineno': 71, 'end_lineno': 81, 'mod_name': 'fossick.cdp.eval', 'package': 'fossick', 'version': '0.1.1', 'docstring': "Evaluate `expr` in the page and return its decoded value.\n\nUpgrades fastcdp's `eval` in three ways every caller wants: promises are awaited, objects and\narrays come back as Python values instead of `None`, and a JS exception is raised as `JSError`\nrather than silently read as an empty result.", 'public_api': False}, '_dist': 0.8112309575080872, '_rrf_score': 0.006756756756756757, '_source': 'env', '_src_id': 'env:192744', '_score': 0.002533783783783784, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43607, 'content': 'def download_files(url:str,\n                   pattern:str=\'*\',          # Glob or regex to match download hrefs\n                   save_dir:str=\'.\',         # Directory to save files\n                   follow:bool=False,        # Crawl linked pages looking for more files\n                   follow_sel:str=\'a[href]\', # Selector for page links to follow when crawling\n                   file_sel:str=\'a[href]\',   # Selector to scan for download links on each page\n                   same_domain:bool=True,\n                   max_pages:int=10,\n                   delay:float=0,\n                   force:bool=False,         # Re-download even if file already exists\n                   heavy:bool=False,\n                   stealthy:bool=False,\n                   **kw,                     # Extra kwargs passed to fetch/http_get (e.g. verify)\n                  ) -> list[Path]:\n    "Download all hrefs matching pattern from url; optionally crawl linked pages first"\n    save_dir = Path(save_dir)\n    save_dir.mkdir(parents=True, exist_ok=True)\n    rx = compile_pattern(pattern)\n    pages = (crawl(url, follow_sel=follow_sel, same_domain=same_domain, max_pages=max_pages, delay=delay,\n       heavy=heavy, stealthy=stealthy, **kw) if follow else [fetch(url, heavy=heavy, stealthy=stealthy, **kw)])\n    saved, seen = [], set()\n    for pg in pages:\n        for a in pg.css(file_sel):\n            href = a.attrib.get(\'href\', \'\')\n            if not rx.search(href): continue\n            abs_url = pg.urljoin(href)\n            if abs_url in seen: continue\n            seen.add(abs_url)\n            fname = save_dir / abs_url.rsplit(\'/\', 1)[-1].split(\'?\')[0]\n            if fname.exists() and not force: saved.append(fname); continue\n            if (r:=get_page(abs_url, verify=kw.get(\'verify\', False))).status == 200: fname.write_bytes(r.body)\n            saved.append(fname)\n    return saved', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1784774842.3054273, 'name': 'download_files', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 562, 'end_lineno': 594, 'mod_name': 'fossick.core.download_files', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'Download all hrefs matching pattern from url; optionally crawl linked pages first', 'public_api': True}, 'rank': -13.72049456068755, '_rrf_score': 0.01282051282051282, '_source': 'env', '_src_id': 'env:43607', '_score': 0.002403846153846154, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43635, 'content': "def _page():\n    'The page opened by the last `browse` call.'\n    if _browser['page'] is None: raise ValueError('no open page — call browse(url) first')\n    return _browser['page']", 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/mcp.py', 'uploaded_at': 1784774842.3057618, 'name': '_page', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 177, 'end_lineno': 180, 'mod_name': 'fossick.mcp._page', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'The page opened by the last `browse` call.', 'public_api': False}, 'rank': -8.642643429254889, '_rrf_score': 0.012658227848101266, '_source': 'env', '_src_id': 'env:43635', '_score': 0.0023734177215189874, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192788, 'content': 'async def shop_dismiss(page:Page) -> list:\n    "Click through cookie/consent banners and closable modals. Returns the labels it clicked."\n    await page.shop_prep()\n    return await page.eval(\'__fk.dismiss()\') or []', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/shop.py', 'uploaded_at': 1785977784.1639626, 'name': 'shop_dismiss', 'lang': '.py', 'type': 'AsyncFunctionDef', 'lineno': 506, 'end_lineno': 509, 'mod_name': 'fossick.shop.shop_dismiss', 'package': 'fossick', 'version': '0.1.1', 'docstring': 'Click through cookie/consent banners and closable modals. Returns the labels it clicked.', 'public_api': False}, 'rank': -7.958128943948474, '_rrf_score': 0.011627906976744186, '_source': 'env', '_src_id': 'env:192788', '_score': 0.002180232558139535, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192763, 'content': 'def replay_xhr(capture:dict,        # a request dict from cdp.calls() or find_xhr(session=True)[i].capture\n               data=None,           # override request body (str/dict); defaults to the captured body\n               port:int=9223,       # debug Chrome port to pull cookies from\n               use_cookies:bool=True, # attach the browser\'s cookies for this domain (keeps auth)\n               **kw,                # Extra kwargs passed to the underlying Fetcher (headers, params, verify)\n              ) -> Response:\n    "Replay a browser-captured request as a fast plain-HTTP call, authenticated via the debug Chrome\'s cookies."\n    url = capture[\'url\']\n    method = (capture.get(\'method\') or \'GET\').upper()\n    headers = {k: v for k, v in (capture.get(\'headers\') or {}).items()\n               if k.lower() not in (\'content-length\', \'host\', \'connection\', \'accept-encoding\', \'cookie\')}\n    if use_cookies:\n        from fossick.cdp import cdp_cookies\n        if (ck := cdp_cookies(url, port=port, as_dict=True)): kw.setdefault(\'cookies\', ck)\n    kw.setdefault(\'headers\', headers)\n    kw.setdefault(\'verify\', False)\n    body = data if data is not None else capture.get(\'postData\')\n    F = _fetcher()\n    if method == \'GET\': return F.get(url, **kw)\n    if isinstance(body, str):\n        try: body = json.loads(body)\n        except: return F.post(url, data=body, **kw)\n    return F.post(url, json=body, **kw)', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1785977784.1587985, 'name': 'replay_xhr', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 542, 'end_lineno': 564, 'mod_name': 'fossick.core.replay_xhr', 'package': 'fossick', 'version': '0.1.1', 'docstring': "Replay a browser-captured request as a fast plain-HTTP call, authenticated via the debug Chrome's cookies.", 'public_api': True}, 'rank': -9.750607432202157, '_rrf_score': 0.010869565217391304, '_source': 'env', '_src_id': 'env:192763', '_score': 0.0020380434782608695, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43530, 'content': 'async def html(self:Page):\n    "Live outerHTML of the page (post-JS), as a string"\n    return await self.eval(\'document.documentElement.outerHTML\')', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/cdp.py', 'uploaded_at': 1784774842.303492, 'name': 'html', 'lang': '.py', 'type': 'AsyncFunctionDef', 'lineno': 134, 'end_lineno': 136, 'mod_name': 'fossick.cdp.html', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'Live outerHTML of the page (post-JS), as a string', 'public_api': False}, 'rank': -7.709085861750738, '_rrf_score': 0.009615384615384616, '_source': 'env', '_src_id': 'env:43530', '_score': 0.0018028846153846155, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 201331, 'content': 'def get_pdf(url_or_path:str, # URL or local path to PDF\n            save_path=None, # optional path to save the PDF (directory or full path)\n            **scrapling_kw # extra kwargs passed to scrapling fetcher (e.g. verify, headers, timeout)\n            ) -> PdfDocument|None:\n    """Fetch PDF from URL or path and return as PdfDocument, or None if it is not a PDF.\n\n    Identity comes from the %PDF- magic bytes, not the URL suffix. arXiv serves its PDFs from\n    extensionless URLs (`arxiv.org/pdf/1706.03762v7`), so a suffix check rejected every paper\n    fossick is most used to read. NCBI PMC\'s download page is a proof-of-work interstitial;\n    its challenge is solved once here rather than mistaken for a failed PDF."""\n    pth = Path(url_or_path) if not url_or_path.startswith(\'http\') else None\n    if pth: src = pth.read_bytes() if pth.exists() else None\n    else: src = r.body if (r:=get_page(url_or_path, **scrapling_kw)).status == 200 else None\n    if not _is_pdf(src) and url_or_path.startswith(\'http\'): src = _pmc_pow_pdf(url_or_path, src or b\'\', scrapling_kw.get(\'timeout\', 60))\n    if not _is_pdf(src): return None\n    if save_path:\n        pth = p/url_or_path.rsplit(\'/\', 1)[-1].split(\'?\')[0] if (p:=Path(save_path)).is_dir() else p\n        if not pth.suffix: pth = pth.with_suffix(\'.pdf\')\n        pth.mk_bytes(src)\n    return PdfDocument.from_bytes(src)', 'metadata': {'path': '/Users/71293/code/personal/orgs/ramabana/.venv/lib/python3.13/site-packages/fossick/core.py', 'uploaded_at': 1786512967.331086, 'name': 'get_pdf', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 327, 'end_lineno': 346, 'mod_name': 'fossick.core.get_pdf', 'package': 'fossick', 'version': '0.1.10', 'docstring': "Fetch PDF from URL or path and return as PdfDocument, or None if it is not a PDF.\n\nIdentity comes from the %PDF- magic bytes, not the URL suffix. arXiv serves its PDFs from\nextensionless URLs (`arxiv.org/pdf/1706.03762v7`), so a suffix check rejected every paper\nfossick is most used to read. NCBI PMC's download page is a proof-of-work interstitial;\nits challenge is solved once here rather than mistaken for a failed PDF.", 'public_api': True}, 'rank': -11.963066473254559, '_rrf_score': 0.009259259259259259, '_source': 'env', '_src_id': 'env:201331', '_score': 0.001736111111111111, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 43550, 'content': 'def read_yt(\n    url:str,              # YouTube URL or video ID\n    force:bool=False,     # re-fetch even if cached\n    as_json:bool=False,   # output JSON\n):\n    "Fetch YouTube metadata and full transcript."\n    v = _read_yt(url, force=force)\n    if as_json: print(json.dumps(v, default=str))\n    else:\n        print(f"# {v[\'title\']}")\n        print(f"Channel: {v[\'channel\']}  Duration: {v.get(\'duration\')}s")\n        if v.get(\'source\'): print(f"\\n{v[\'source\']}")', 'metadata': {'path': '/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fossick/cli.py', 'uploaded_at': 1784774842.3050234, 'name': 'read_yt', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 117, 'end_lineno': 128, 'mod_name': 'fossick.cli.read_yt', 'package': 'fossick', 'version': '0.0.15', 'docstring': 'Fetch YouTube metadata and full transcript.', 'public_api': True}, 'rank': -7.848427184562115, '_rrf_score': 0.007692307692307693, '_source': 'env', '_src_id': 'env:43550', '_score': 0.0014423076923076924, 'callers': [], 'callees': [], 'co_dispatched': []}, {'rowid': 192770, 'content': 'def _sh(url=None, port=9223):\n    "The one shopping session, opened at `url` — or moved there — as needed."\n    from fossick.shop import shop as _open\n    if (s := _shop_session.get(\'s\')) is None:\n        if url is None: raise ValueError(\'no shop open — call shop_open(url) first\')\n        _shop_session[\'s\'] = s = _open(url, port=port)\n    elif url: s.open(url)\n    return s', 'metadata': {'path': '/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/fossick/mcp.py', 'uploaded_at': 1785977784.1604002, 'name': '_sh', 'lang': '.py', 'type': 'FunctionDef', 'lineno': 238, 'end_lineno': 245, 'mod_name': 'fossick.mcp._sh', 'package': 'fossick', 'version': '0.1.1', 'docstring': 'The one shopping session, opened at `url` — or moved there — as needed.', 'public_api': False}, 'rank': -7.822788149180066, '_rrf_score': 0.007518796992481203, '_source': 'env', '_src_id': 'env:192770', '_score': 0.0014097744360902255, 'callers': [], 'callees': [], 'co_dispatched': []}]
k.update_pkgs(['watchfiles'])
loading pkgs ['fossick'] ...

Updating packages:   0%|                                                                                                               | 0/1 [00:00<?, ?pkg/s]
Updating packages: 100%|███████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00,  8.53pkg/s]
Updating packages: 100%|███████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00,  8.50pkg/s]
updating pkg: fossick ...
package {'name': 'fossick', 'version': '0.1.14'} already loaded.

Where to add code

where_to_add(description) suggests where to put new code. It runs context on the description. For each hit in the repo index it returns a dict with node, path, insert_after, co_dispatched and pagerank. insert_after is the last line of the hit, or the last line of its last co-dispatch peer in the same file. Hits from installed packages are skipped because they are not in your repo. limit and other keyword arguments go to context.

Kosha.where_to_add

def where_to_add(
    description:str, limit:int=5, *, repo:bool=True, env:bool=True, graph:bool=True, boost:bool=True,
    compact:bool=False, columns:str='content,metadata', sys_wide:bool=True, rerank:bool=False, rerank_model:str=None
)->fastcore.foundation.L:

Places to add code for description: the file, the line to insert after, and the peers it would join.

pts = k.where_to_add('add a new way to search codebase', 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 274

The next cell tests the incremental sync. Nested functions keep their own calls. Unchanged files are skipped. Removed definitions disappear. A caller-supplied root keeps the package name in node names.

from tempfile import TemporaryDirectory

# Fast mode preserves lexical scope and does not attribute nested calls to the parent.
_src = '''
def target(): pass
class A:
    def same(self): target()
class B:
    def same(self): target()
def outer():
    def inner(): target()
    return inner
'''
_edges, _meta = static_edges({'m': _src})
_edge_pairs = {(e['caller'], e['callee']) for e in _edges}
assert {'m.A.same', 'm.B.same', 'm.outer', 'm.outer.inner'} <= set(_meta)
assert ('m.outer', 'm.target') not in _edge_pairs
assert {('m.A.same', 'm.target'), ('m.B.same', 'm.target'), ('m.outer.inner', 'm.target')} <= _edge_pairs

# Incremental fast sync records lambda ownership, skips unchanged source, and drops removed nodes.
with TemporaryDirectory() as _tmp:
    _path = Path(_tmp) / 'm.py'
    _path.write_text('x = lambda: 1\ndef keep(): return x()\ndef gone(): pass\n')
    _g = CodeGraph(':memory:').sync_dir(_tmp)
    _hash = _g.fi(where='path=?', where_args=[str(_path)])[0]['source_hash']
    _g.sync_dir(_tmp)
    assert _g.fi(where='path=?', where_args=[str(_path)])[0]['source_hash'] == _hash
    assert _g.gn(where='node=?', where_args=['m.x'])[0]['file'] == str(_path)
    _path.write_text('x = lambda: 2\ndef keep(): return x()\n')
    _g.sync_dir(_tmp)
    assert not _g.gn(where='node=?', where_args=['m.gone'])
    assert _g.gn(where='node=?', where_args=['m.x'])[0]['file'] == str(_path)

# Full mode replaces a changed file rather than retaining definitions removed from it.
with TemporaryDirectory() as _tmp:
    _path = Path(_tmp) / 'm.py'
    _path.write_text('def old(): pass\n')
    _g = CodeGraph(':memory:').sync_dir(_tmp, mode='full')
    _path.write_text('def new(): pass\n')
    _g.sync_dir(_tmp, mode='full')
    _nodes = set(L(_g.gn(select='node')).attrgot('node'))
    assert 'm.old' not in _nodes and 'm.new' in _nodes

# A caller-supplied root preserves the package segment in generated node names.
with TemporaryDirectory() as _tmp:
    _root, _pkg = Path(_tmp), Path(_tmp) / 'mypkg'
    _pkg.mkdir()
    (_pkg / 'core.py').write_text('def api(): pass\n')
    _g = CodeGraph(':memory:').sync_dir(_pkg, root=_root)
    assert _g.gn(where='node=?', where_args=['mypkg.core.api'])

# A file edited after it was parsed is recorded under the hash that was compared, so the next sync parses it again.
with TemporaryDirectory() as _tmp:
    _path = Path(_tmp) / 'm.py'
    _path.write_text('def a(): pass\n')
    _fp, _g = CodeGraph._fast_process_files, CodeGraph(':memory:')
    CodeGraph._fast_process_files = lambda self, *a, **k: (_fp(self, *a, **k), _path.write_text('def b(): pass\n'))[0]
    try: _g.sync_dir(_tmp)
    finally: CodeGraph._fast_process_files = _fp
    _g.sync_dir(_tmp)
    assert {r['node'] for r in _g.gn()} == {'m.b'}

# The busy timeout of a Kosha reaches its call graph, so concurrent writers wait for each other.
with TemporaryDirectory() as _tmp:
    _k = Kosha(dir=Path(_tmp), xdg_dir=Path(_tmp), busy_timeout=4321)
    assert _k.graph.db.q('pragma busy_timeout')[0]['timeout'] == 4321
    assert CodeGraph(Path(_tmp)/'g.db', busy_timeout=1234).db.q('pragma busy_timeout')[0]['timeout'] == 1234

# A node that loses its last edge goes back to degree zero.
_g = CodeGraph(':memory:')
_g.gn.insert_all([{'node': 'a'}, {'node': 'b'}])
_g.ge.insert({'caller': 'a', 'callee': 'b', 'kind': 'static', 'confidence': 1.0})
_g._centrality(); _g.ge.delete_where(); _g._centrality()
assert all(not r['in_degree'] and not r['out_degree'] for r in _g.gn())

print('fast incremental graph tests passed')
█

 |----------------------------------------| 0.00% [0/1 00:00<?]
 |████████████████████████████████████████| 100.00% [1/1 00:00<00:00]
                                                                     
█

 |----------------------------------------| 0.00% [0/1 00:00<?]
 |████████████████████████████████████████| 100.00% [1/1 00:00<00:00]
                                                                     
fast incremental graph tests passed