This is page 11 of 14. Use http://codebase.md/shashankss1205/codegraphcontext?page={x} to view the full context.
# Directory Structure
```
├── .github
│ └── workflows
│ ├── e2e-tests.yml
│ ├── post_discord_invite.yml
│ ├── test.yml
│ └── update-contributors.yml
├── .gitignore
├── CONTRIBUTING.md
├── contributors.md
├── docs
│ ├── docs
│ │ ├── architecture.md
│ │ ├── cli.md
│ │ ├── contributing_languages.md
│ │ ├── contributing.md
│ │ ├── cookbook.md
│ │ ├── core.md
│ │ ├── future_work.md
│ │ ├── images
│ │ │ ├── 1.png
│ │ │ ├── 11.png
│ │ │ ├── 12.png
│ │ │ ├── 13.png
│ │ │ ├── 14.png
│ │ │ ├── 16.png
│ │ │ ├── 19.png
│ │ │ ├── 2.png
│ │ │ ├── 20.png
│ │ │ ├── 21.png
│ │ │ ├── 22.png
│ │ │ ├── 23.png
│ │ │ ├── 24.png
│ │ │ ├── 26.png
│ │ │ ├── 28.png
│ │ │ ├── 29.png
│ │ │ ├── 3.png
│ │ │ ├── 30.png
│ │ │ ├── 31.png
│ │ │ ├── 32.png
│ │ │ ├── 33.png
│ │ │ ├── 34.png
│ │ │ ├── 35.png
│ │ │ ├── 36.png
│ │ │ ├── 38.png
│ │ │ ├── 39.png
│ │ │ ├── 4.png
│ │ │ ├── 40.png
│ │ │ ├── 41.png
│ │ │ ├── 42.png
│ │ │ ├── 43.png
│ │ │ ├── 44.png
│ │ │ ├── 5.png
│ │ │ ├── 6.png
│ │ │ ├── 7.png
│ │ │ ├── 8.png
│ │ │ ├── 9.png
│ │ │ ├── Indexing.gif
│ │ │ ├── tool_images
│ │ │ │ ├── 1.png
│ │ │ │ ├── 2.png
│ │ │ │ └── 3.png
│ │ │ └── Usecase.gif
│ │ ├── index.md
│ │ ├── installation.md
│ │ ├── license.md
│ │ ├── server.md
│ │ ├── tools.md
│ │ ├── troubleshooting.md
│ │ └── use_cases.md
│ ├── mkdocs.yml
│ └── site
│ ├── 404.html
│ ├── architecture
│ │ └── index.html
│ ├── assets
│ │ ├── images
│ │ │ └── favicon.png
│ │ ├── javascripts
│ │ │ ├── bundle.f55a23d4.min.js
│ │ │ ├── bundle.f55a23d4.min.js.map
│ │ │ ├── lunr
│ │ │ │ ├── min
│ │ │ │ │ ├── lunr.ar.min.js
│ │ │ │ │ ├── lunr.da.min.js
│ │ │ │ │ ├── lunr.de.min.js
│ │ │ │ │ ├── lunr.du.min.js
│ │ │ │ │ ├── lunr.el.min.js
│ │ │ │ │ ├── lunr.es.min.js
│ │ │ │ │ ├── lunr.fi.min.js
│ │ │ │ │ ├── lunr.fr.min.js
│ │ │ │ │ ├── lunr.he.min.js
│ │ │ │ │ ├── lunr.hi.min.js
│ │ │ │ │ ├── lunr.hu.min.js
│ │ │ │ │ ├── lunr.hy.min.js
│ │ │ │ │ ├── lunr.it.min.js
│ │ │ │ │ ├── lunr.ja.min.js
│ │ │ │ │ ├── lunr.jp.min.js
│ │ │ │ │ ├── lunr.kn.min.js
│ │ │ │ │ ├── lunr.ko.min.js
│ │ │ │ │ ├── lunr.multi.min.js
│ │ │ │ │ ├── lunr.nl.min.js
│ │ │ │ │ ├── lunr.no.min.js
│ │ │ │ │ ├── lunr.pt.min.js
│ │ │ │ │ ├── lunr.ro.min.js
│ │ │ │ │ ├── lunr.ru.min.js
│ │ │ │ │ ├── lunr.sa.min.js
│ │ │ │ │ ├── lunr.stemmer.support.min.js
│ │ │ │ │ ├── lunr.sv.min.js
│ │ │ │ │ ├── lunr.ta.min.js
│ │ │ │ │ ├── lunr.te.min.js
│ │ │ │ │ ├── lunr.th.min.js
│ │ │ │ │ ├── lunr.tr.min.js
│ │ │ │ │ ├── lunr.vi.min.js
│ │ │ │ │ └── lunr.zh.min.js
│ │ │ │ ├── tinyseg.js
│ │ │ │ └── wordcut.js
│ │ │ └── workers
│ │ │ ├── search.973d3a69.min.js
│ │ │ └── search.973d3a69.min.js.map
│ │ └── stylesheets
│ │ ├── main.2a3383ac.min.css
│ │ ├── main.2a3383ac.min.css.map
│ │ ├── palette.06af60db.min.css
│ │ └── palette.06af60db.min.css.map
│ ├── cli
│ │ └── index.html
│ ├── contributing
│ │ └── index.html
│ ├── contributing_languages
│ │ └── index.html
│ ├── cookbook
│ │ └── index.html
│ ├── core
│ │ └── index.html
│ ├── future_work
│ │ └── index.html
│ ├── images
│ │ ├── 1.png
│ │ ├── 11.png
│ │ ├── 12.png
│ │ ├── 13.png
│ │ ├── 14.png
│ │ ├── 16.png
│ │ ├── 19.png
│ │ ├── 2.png
│ │ ├── 20.png
│ │ ├── 21.png
│ │ ├── 22.png
│ │ ├── 23.png
│ │ ├── 24.png
│ │ ├── 26.png
│ │ ├── 28.png
│ │ ├── 29.png
│ │ ├── 3.png
│ │ ├── 30.png
│ │ ├── 31.png
│ │ ├── 32.png
│ │ ├── 33.png
│ │ ├── 34.png
│ │ ├── 35.png
│ │ ├── 36.png
│ │ ├── 38.png
│ │ ├── 39.png
│ │ ├── 4.png
│ │ ├── 40.png
│ │ ├── 41.png
│ │ ├── 42.png
│ │ ├── 43.png
│ │ ├── 44.png
│ │ ├── 5.png
│ │ ├── 6.png
│ │ ├── 7.png
│ │ ├── 8.png
│ │ ├── 9.png
│ │ ├── Indexing.gif
│ │ ├── tool_images
│ │ │ ├── 1.png
│ │ │ ├── 2.png
│ │ │ └── 3.png
│ │ └── Usecase.gif
│ ├── index.html
│ ├── installation
│ │ └── index.html
│ ├── license
│ │ └── index.html
│ ├── search
│ │ └── search_index.json
│ ├── server
│ │ └── index.html
│ ├── sitemap.xml
│ ├── sitemap.xml.gz
│ ├── tools
│ │ └── index.html
│ ├── troubleshooting
│ │ └── index.html
│ └── use_cases
│ └── index.html
├── images
│ ├── 1.png
│ ├── 11.png
│ ├── 12.png
│ ├── 13.png
│ ├── 14.png
│ ├── 16.png
│ ├── 19.png
│ ├── 2.png
│ ├── 20.png
│ ├── 21.png
│ ├── 22.png
│ ├── 23.png
│ ├── 24.png
│ ├── 26.png
│ ├── 28.png
│ ├── 29.png
│ ├── 3.png
│ ├── 30.png
│ ├── 31.png
│ ├── 32.png
│ ├── 33.png
│ ├── 34.png
│ ├── 35.png
│ ├── 36.png
│ ├── 38.png
│ ├── 39.png
│ ├── 4.png
│ ├── 40.png
│ ├── 41.png
│ ├── 42.png
│ ├── 43.png
│ ├── 44.png
│ ├── 5.png
│ ├── 6.png
│ ├── 7.png
│ ├── 8.png
│ ├── 9.png
│ ├── Indexing.gif
│ ├── tool_images
│ │ ├── 1.png
│ │ ├── 2.png
│ │ └── 3.png
│ └── Usecase.gif
├── LICENSE
├── MANIFEST.in
├── organizer
│ ├── CONTRIBUTING_LANGUAGES.md
│ ├── cookbook.md
│ ├── docs.md
│ ├── language_specific_nodes.md
│ ├── Tools_Exploration.md
│ └── troubleshoot.md
├── package-lock.json
├── pyproject.toml
├── README.md
├── scripts
│ ├── generate_lang_contributors.py
│ └── post_install_fix.sh
├── SECURITY.md
├── src
│ └── codegraphcontext
│ ├── __init__.py
│ ├── __main__.py
│ ├── cli
│ │ ├── __init__.py
│ │ ├── cli_helpers.py
│ │ ├── main.py
│ │ ├── setup_macos.py
│ │ └── setup_wizard.py
│ ├── core
│ │ ├── __init__.py
│ │ ├── database.py
│ │ ├── jobs.py
│ │ └── watcher.py
│ ├── prompts.py
│ ├── server.py
│ ├── tools
│ │ ├── __init__.py
│ │ ├── advanced_language_query_tool.py
│ │ ├── code_finder.py
│ │ ├── graph_builder.py
│ │ ├── languages
│ │ │ ├── c.py
│ │ │ ├── cpp.py
│ │ │ ├── go.py
│ │ │ ├── java.py
│ │ │ ├── javascript.py
│ │ │ ├── python.py
│ │ │ ├── ruby.py
│ │ │ ├── rust.py
│ │ │ └── typescript.py
│ │ ├── package_resolver.py
│ │ ├── query_tool_languages
│ │ │ ├── c_toolkit.py
│ │ │ ├── cpp_toolkit.py
│ │ │ ├── go_toolkit.py
│ │ │ ├── java_toolkit.py
│ │ │ ├── javascript_toolkit.py
│ │ │ ├── python_toolkit.py
│ │ │ ├── ruby_toolkit.py
│ │ │ ├── rust_toolkit.py
│ │ │ └── typescript_toolkit.py
│ │ └── system.py
│ └── utils
│ └── debug_log.py
├── tests
│ ├── __init__.py
│ ├── conftest.py
│ ├── sample_project
│ │ ├── advanced_calls.py
│ │ ├── advanced_classes.py
│ │ ├── advanced_classes2.py
│ │ ├── advanced_functions.py
│ │ ├── advanced_imports.py
│ │ ├── async_features.py
│ │ ├── callbacks_decorators.py
│ │ ├── circular1.py
│ │ ├── circular2.py
│ │ ├── class_instantiation.py
│ │ ├── cli_and_dunder.py
│ │ ├── complex_classes.py
│ │ ├── comprehensions_generators.py
│ │ ├── context_managers.py
│ │ ├── control_flow.py
│ │ ├── datatypes.py
│ │ ├── dynamic_dispatch.py
│ │ ├── dynamic_imports.py
│ │ ├── edge_cases
│ │ │ ├── comments_only.py
│ │ │ ├── docstring_only.py
│ │ │ ├── empty.py
│ │ │ ├── hardcoded_secrets.py
│ │ │ ├── long_functions.py
│ │ │ └── syntax_error.py
│ │ ├── function_chains.py
│ │ ├── generators.py
│ │ ├── import_reexports.py
│ │ ├── mapping_calls.py
│ │ ├── module_a.py
│ │ ├── module_b.py
│ │ ├── module_c
│ │ │ ├── __init__.py
│ │ │ ├── submodule1.py
│ │ │ └── submodule2.py
│ │ ├── namespace_pkg
│ │ │ └── ns_module.py
│ │ ├── pattern_matching.py
│ │ └── typing_examples.py
│ ├── sample_project_c
│ │ ├── cgc_sample
│ │ ├── include
│ │ │ ├── config.h
│ │ │ ├── math
│ │ │ │ └── vec.h
│ │ │ ├── module.h
│ │ │ ├── platform.h
│ │ │ └── util.h
│ │ ├── Makefile
│ │ ├── README.md
│ │ └── src
│ │ ├── main.c
│ │ ├── math
│ │ │ └── vec.c
│ │ ├── module.c
│ │ └── util.c
│ ├── sample_project_cpp
│ │ ├── class_features.cpp
│ │ ├── classes.cpp
│ │ ├── control_flow.cpp
│ │ ├── edge_cases.cpp
│ │ ├── enum_struct_union.cpp
│ │ ├── exceptions.cpp
│ │ ├── file_io.cpp
│ │ ├── function_chain.cpp
│ │ ├── function_chain.h
│ │ ├── function_types.cpp
│ │ ├── main.cpp
│ │ ├── main.exe
│ │ ├── namespaces.cpp
│ │ ├── raii_example.cpp
│ │ ├── README.md
│ │ ├── sample_project.exe
│ │ ├── stl_usage.cpp
│ │ ├── templates.cpp
│ │ └── types_variable_assignments.cpp
│ ├── sample_project_go
│ │ ├── advanced_types.go
│ │ ├── basic_functions.go
│ │ ├── embedded_composition.go
│ │ ├── error_handling.go
│ │ ├── generics.go
│ │ ├── go.mod
│ │ ├── goroutines_channels.go
│ │ ├── interfaces.go
│ │ ├── packages_imports.go
│ │ ├── README.md
│ │ ├── structs_methods.go
│ │ └── util
│ │ └── helpers.go
│ ├── sample_project_java
│ │ ├── out
│ │ │ └── com
│ │ │ └── example
│ │ │ └── app
│ │ │ ├── annotations
│ │ │ │ └── Logged.class
│ │ │ ├── Main.class
│ │ │ ├── misc
│ │ │ │ ├── Outer.class
│ │ │ │ └── Outer$Inner.class
│ │ │ ├── model
│ │ │ │ ├── Role.class
│ │ │ │ └── User.class
│ │ │ ├── service
│ │ │ │ ├── AbstractGreeter.class
│ │ │ │ ├── GreetingService.class
│ │ │ │ └── impl
│ │ │ │ └── GreetingServiceImpl.class
│ │ │ └── util
│ │ │ ├── CollectionUtils.class
│ │ │ └── IOHelper.class
│ │ ├── README.md
│ │ ├── sources.txt
│ │ └── src
│ │ └── com
│ │ └── example
│ │ └── app
│ │ ├── annotations
│ │ │ └── Logged.java
│ │ ├── Main.java
│ │ ├── misc
│ │ │ └── Outer.java
│ │ ├── model
│ │ │ ├── Role.java
│ │ │ └── User.java
│ │ ├── service
│ │ │ ├── AbstractGreeter.java
│ │ │ ├── GreetingService.java
│ │ │ └── impl
│ │ │ └── GreetingServiceImpl.java
│ │ └── util
│ │ ├── CollectionUtils.java
│ │ └── IOHelper.java
│ ├── sample_project_javascript
│ │ ├── arrays.js
│ │ ├── asyncAwait.js
│ │ ├── classes.js
│ │ ├── dom.js
│ │ ├── errorHandling.js
│ │ ├── events.js
│ │ ├── exporter.js
│ │ ├── fetchAPI.js
│ │ ├── fixtures
│ │ │ └── js
│ │ │ └── accessors.js
│ │ ├── functions.js
│ │ ├── importer.js
│ │ ├── objects.js
│ │ ├── promises.js
│ │ ├── README.md
│ │ └── variables.js
│ ├── sample_project_misc
│ │ ├── index.html
│ │ ├── README.md
│ │ ├── styles.css
│ │ ├── tables.css
│ │ └── tables.html
│ ├── sample_project_php
│ │ ├── classes_objects.php
│ │ ├── database.php
│ │ ├── edgecases.php
│ │ ├── error_handling.php
│ │ ├── file_handling.php
│ │ ├── functions.php
│ │ ├── generators_iterators.php
│ │ ├── globals_superglobals.php
│ │ ├── Inheritance.php
│ │ ├── interface_traits.php
│ │ └── README.md
│ ├── sample_project_ruby
│ │ ├── class_example.rb
│ │ ├── enumerables.rb
│ │ ├── error_handling.rb
│ │ ├── file_io.rb
│ │ ├── inheritance_example.rb
│ │ ├── main.rb
│ │ ├── metaprogramming.rb
│ │ ├── mixins_example.rb
│ │ ├── module_example.rb
│ │ └── tests
│ │ ├── test_mixins.py
│ │ └── test_sample.rb
│ ├── sample_project_rust
│ │ ├── Cargo.toml
│ │ ├── README.md
│ │ └── src
│ │ ├── basic_functions.rs
│ │ ├── concurrency.rs
│ │ ├── error_handling.rs
│ │ ├── generics.rs
│ │ ├── iterators_closures.rs
│ │ ├── lib.rs
│ │ ├── lifetimes_references.rs
│ │ ├── modules.rs
│ │ ├── smart_pointers.rs
│ │ ├── structs_enums.rs
│ │ └── traits.rs
│ ├── sample_project_typescript
│ │ ├── package.json
│ │ ├── README.md
│ │ ├── src
│ │ │ ├── advanced-types.ts
│ │ │ ├── async-promises.ts
│ │ │ ├── classes-inheritance.ts
│ │ │ ├── decorators-metadata.ts
│ │ │ ├── error-validation.ts
│ │ │ ├── functions-generics.ts
│ │ │ ├── index.ts
│ │ │ ├── modules-namespaces.ts
│ │ │ ├── types-interfaces.ts
│ │ │ └── utilities-helpers.ts
│ │ └── tsconfig.json
│ ├── test_cpp_parser.py
│ ├── test_database_validation.py
│ ├── test_end_to_end.py
│ ├── test_graph_indexing_js.py
│ ├── test_graph_indexing.py
│ ├── test_tree_sitter
│ │ ├── __init__.py
│ │ ├── class_instantiation.py
│ │ ├── complex_classes.py
│ │ └── test_file.py
│ └── test_typescript_parser.py
└── website
├── .example.env
├── .gitignore
├── api
│ └── pypi.ts
├── bun.lockb
├── components.json
├── eslint.config.js
├── index.html
├── package-lock.json
├── package.json
├── postcss.config.js
├── public
│ ├── favicon.ico
│ ├── placeholder.svg
│ └── robots.txt
├── README.md
├── src
│ ├── App.css
│ ├── App.tsx
│ ├── assets
│ │ ├── function-calls.png
│ │ ├── graph-total.png
│ │ ├── hero-graph.jpg
│ │ └── hierarchy.png
│ ├── components
│ │ ├── ComparisonTable.tsx
│ │ ├── CookbookSection.tsx
│ │ ├── DemoSection.tsx
│ │ ├── ExamplesSection.tsx
│ │ ├── FeaturesSection.tsx
│ │ ├── Footer.tsx
│ │ ├── HeroSection.tsx
│ │ ├── InstallationSection.tsx
│ │ ├── MoveToTop.tsx
│ │ ├── ShowDownloads.tsx
│ │ ├── ShowStarGraph.tsx
│ │ ├── TestimonialSection.tsx
│ │ ├── ThemeProvider.tsx
│ │ ├── ThemeToggle.tsx
│ │ └── ui
│ │ ├── accordion.tsx
│ │ ├── alert-dialog.tsx
│ │ ├── alert.tsx
│ │ ├── aspect-ratio.tsx
│ │ ├── avatar.tsx
│ │ ├── badge.tsx
│ │ ├── breadcrumb.tsx
│ │ ├── button.tsx
│ │ ├── calendar.tsx
│ │ ├── card.tsx
│ │ ├── carousel.tsx
│ │ ├── chart.tsx
│ │ ├── checkbox.tsx
│ │ ├── collapsible.tsx
│ │ ├── command.tsx
│ │ ├── context-menu.tsx
│ │ ├── dialog.tsx
│ │ ├── drawer.tsx
│ │ ├── dropdown-menu.tsx
│ │ ├── form.tsx
│ │ ├── hover-card.tsx
│ │ ├── input-otp.tsx
│ │ ├── input.tsx
│ │ ├── label.tsx
│ │ ├── menubar.tsx
│ │ ├── navigation-menu.tsx
│ │ ├── orbiting-circles.tsx
│ │ ├── pagination.tsx
│ │ ├── popover.tsx
│ │ ├── progress.tsx
│ │ ├── radio-group.tsx
│ │ ├── resizable.tsx
│ │ ├── scroll-area.tsx
│ │ ├── select.tsx
│ │ ├── separator.tsx
│ │ ├── sheet.tsx
│ │ ├── sidebar.tsx
│ │ ├── skeleton.tsx
│ │ ├── slider.tsx
│ │ ├── sonner.tsx
│ │ ├── switch.tsx
│ │ ├── table.tsx
│ │ ├── tabs.tsx
│ │ ├── textarea.tsx
│ │ ├── toast.tsx
│ │ ├── toaster.tsx
│ │ ├── toggle-group.tsx
│ │ ├── toggle.tsx
│ │ ├── tooltip.tsx
│ │ └── use-toast.ts
│ ├── hooks
│ │ ├── use-mobile.tsx
│ │ └── use-toast.ts
│ ├── index.css
│ ├── lib
│ │ └── utils.ts
│ ├── main.tsx
│ ├── pages
│ │ ├── Index.tsx
│ │ └── NotFound.tsx
│ └── vite-env.d.ts
├── tailwind.config.ts
├── tsconfig.app.json
├── tsconfig.json
├── tsconfig.node.json
├── vercel.json
└── vite.config.ts
```
# Files
--------------------------------------------------------------------------------
/src/codegraphcontext/server.py:
--------------------------------------------------------------------------------
```python
# src/codegraphcontext/server.py
import urllib.parse
import asyncio
import json
import importlib
import stdlibs
import sys
import traceback
import os
import re
from datetime import datetime
from pathlib import Path
from neo4j.exceptions import CypherSyntaxError
from dataclasses import asdict
from typing import Any, Dict, Coroutine, Optional
from .prompts import LLM_SYSTEM_PROMPT
from .core.database import DatabaseManager
from .core.jobs import JobManager, JobStatus
from .core.watcher import CodeWatcher
from .tools.graph_builder import GraphBuilder
from .tools.code_finder import CodeFinder
from .tools.package_resolver import get_local_package_path
from .utils.debug_log import debug_log, info_logger, error_logger, warning_logger, debug_logger
DEFAULT_EDIT_DISTANCE = 2
DEFAULT_FUZZY_SEARCH = False
class MCPServer:
"""
The main MCP Server class.
This class orchestrates all the major components of the application, including:
- Database connection management (`DatabaseManager`)
- Background job tracking (`JobManager`)
- File system watching for live updates (`CodeWatcher`)
- Tool handlers for graph building, code searching, etc.
- The main JSON-RPC communication loop for interacting with an AI assistant.
"""
def __init__(self, loop=None):
"""
Initializes the MCP server and its components.
Args:
loop: The asyncio event loop to use. If not provided, it gets the current
running loop or creates a new one.
"""
try:
# Initialize the database manager and establish a connection early
# to fail fast if credentials are wrong.
self.db_manager = DatabaseManager()
self.db_manager.get_driver()
except ValueError as e:
raise ValueError(f"Database configuration error: {e}")
# Initialize managers for jobs and file watching.
self.job_manager = JobManager()
# Get the current event loop to pass to thread-sensitive components like the graph builder.
if loop is None:
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
self.loop = loop
# Initialize all the tool handlers, passing them the necessary managers and the event loop.
self.graph_builder = GraphBuilder(self.db_manager, self.job_manager, loop)
self.code_finder = CodeFinder(self.db_manager)
self.code_watcher = CodeWatcher(self.graph_builder, self.job_manager)
# Define the tool manifest that will be exposed to the AI assistant.
self._init_tools()
def _init_tools(self):
"""
Defines the complete tool manifest for the LLM.
This dictionary contains the schema for every tool the AI can call,
including its name, description, and input parameters.
"""
self.tools = {
"add_code_to_graph": {
"name": "add_code_to_graph",
"description": "Performs a one-time scan of a local folder to add its code to the graph. Ideal for indexing libraries, dependencies, or projects not being actively modified. Returns a job ID for background processing.",
"inputSchema": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Path to the directory or file to add."},
"is_dependency": {"type": "boolean", "description": "Whether this code is a dependency.", "default": False}
},
"required": ["path"]
}
},
"check_job_status": {
"name": "check_job_status",
"description": "Check the status and progress of a background job.",
"inputSchema": {
"type": "object",
"properties": { "job_id": {"type": "string", "description": "Job ID from a previous tool call"} },
"required": ["job_id"]
}
},
"list_jobs": {
"name": "list_jobs",
"description": "List all background jobs and their current status.",
"inputSchema": {"type": "object", "properties": {}}
},
"find_code": {
"name": "find_code",
"description": "Find relevant code snippets related to a keyword (e.g., function name, class name, or content).",
"inputSchema": {
"type": "object",
"properties": { "query": {"type": "string", "description": "Keyword or phrase to search for"}, "fuzzy_search": {"type": "boolean", "description": "Whether to use fuzzy search", "default": False}, "edit_distance": {"type": "number", "description": "Edit distance for fuzzy search (between 0-2)", "default": 2}},
"required": ["query"]
}
},
"analyze_code_relationships": {
"name": "analyze_code_relationships",
"description": "Analyze code relationships like 'who calls this function' or 'class hierarchy'. Supported query types include: find_callers, find_callees, find_all_callers, find_all_callees, find_importers, who_modifies, class_hierarchy, overrides, dead_code, call_chain, module_deps, variable_scope, find_complexity, find_functions_by_argument, find_functions_by_decorator.",
"inputSchema": {
"type": "object",
"properties": {
"query_type": {"type": "string", "description": "Type of relationship query to run.", "enum": ["find_callers", "find_callees", "find_all_callers", "find_all_callees", "find_importers", "who_modifies", "class_hierarchy", "overrides", "dead_code", "call_chain", "module_deps", "variable_scope", "find_complexity", "find_functions_by_argument", "find_functions_by_decorator"]},
"target": {"type": "string", "description": "The function, class, or module to analyze."},
"context": {"type": "string", "description": "Optional: specific file path for precise results."}
},
"required": ["query_type", "target"]
}
},
"watch_directory": {
"name": "watch_directory",
"description": "Performs an initial scan of a directory and then continuously monitors it for changes, automatically keeping the graph up-to-date. Ideal for projects under active development. Returns a job ID for the initial scan.",
"inputSchema": {
"type": "object",
"properties": { "path": {"type": "string", "description": "Path to directory to watch"} },
"required": ["path"]
}
},
"execute_cypher_query": {
"name": "execute_cypher_query",
"description": "Fallback tool to run a direct, read-only Cypher query against the code graph. Use this for complex questions not covered by other tools. The graph contains nodes representing code structures and relationships between them. **Schema Overview:**\n- **Nodes:** `Repository`, `File`, `Module`, `Class`, `Function`.\n- **Properties:** Nodes have properties like `name`, `path`, `cyclomatic_complexity` (on Function nodes), and `code`.\n- **Relationships:** `CONTAINS` (e.g., File-[:CONTAINS]->Function), `CALLS` (Function-[:CALLS]->Function or File-[:CALLS]->Function), `IMPORTS` (File-[:IMPORTS]->Module), `INHERITS` (Class-[:INHERITS]->Class).",
"inputSchema": {
"type": "object",
"properties": { "cypher_query": {"type": "string", "description": "The read-only Cypher query to execute."} },
"required": ["cypher_query"]
}
},
"add_package_to_graph": {
"name": "add_package_to_graph",
"description": "Add a package to the graph by discovering its location. Supports multiple languages. Returns immediately with a job ID.",
"inputSchema": {
"type": "object",
"properties": {
"package_name": {"type": "string", "description": "Name of the package to add (e.g., 'requests', 'express', 'moment', 'lodash')."},
"language": {"type": "string", "description": "The programming language of the package.", "enum": ["python", "javascript", "typescript", "java", "c", "go", "ruby", "php","cpp"]},
"is_dependency": {"type": "boolean", "description": "Mark as a dependency.", "default": True}
},
"required": ["package_name", "language"]
}
},
"find_dead_code": {
"name": "find_dead_code",
"description": "Find potentially unused functions (dead code) across the entire indexed codebase, optionally excluding functions with specific decorators.",
"inputSchema": {
"type": "object",
"properties": {
"exclude_decorated_with": {"type": "array", "items": {"type": "string"}, "description": "Optional: A list of decorator names (e.g., '@app.route') to exclude from dead code detection.", "default": []}
}
}
},
"calculate_cyclomatic_complexity": {
"name": "calculate_cyclomatic_complexity",
"description": "Calculate the cyclomatic complexity of a specific function to measure its complexity.",
"inputSchema": {
"type": "object",
"properties": {
"function_name": {"type": "string", "description": "The name of the function to analyze."},
"file_path": {"type": "string", "description": "Optional: The full path to the file containing the function for a more specific query."}
},
"required": ["function_name"]
}
},
"find_most_complex_functions": {
"name": "find_most_complex_functions",
"description": "Find the most complex functions in the codebase based on cyclomatic complexity.",
"inputSchema": {
"type": "object",
"properties": {
"limit": {"type": "integer", "description": "The maximum number of complex functions to return.", "default": 10}
}
}
},
"list_indexed_repositories": {
"name": "list_indexed_repositories",
"description": "List all indexed repositories.",
"inputSchema": {
"type": "object",
"properties": {}
}
},
"delete_repository": {
"name": "delete_repository",
"description": "Delete an indexed repository from the graph.",
"inputSchema": {
"type": "object",
"properties": {
"repo_path": {"type": "string", "description": "The path of the repository to delete."}
},
"required": ["repo_path"]
}
},
"visualize_graph_query": {
"name": "visualize_graph_query",
"description": "Generates a URL to visualize the results of a Cypher query in the Neo4j Browser. The user can open this URL in their web browser to see the graph visualization.",
"inputSchema": {
"type": "object",
"properties": {
"cypher_query": {"type": "string", "description": "The Cypher query to visualize."}
},
"required": ["cypher_query"]
}
},
"list_watched_paths": {
"name": "list_watched_paths",
"description": "Lists all directories currently being watched for live file changes.",
"inputSchema": {"type": "object", "properties": {}}
},
"unwatch_directory": {
"name": "unwatch_directory",
"description": "Stops watching a directory for live file changes.",
"inputSchema": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "The absolute path of the directory to stop watching."}
},
"required": ["path"]
}
}
}
def get_database_status(self) -> dict:
"""Returns the current connection status of the Neo4j database."""
return {"connected": self.db_manager.is_connected()}
def execute_cypher_query_tool(self, **args) -> Dict[str, Any]:
"""
Tool implementation for executing a read-only Cypher query.
Important: Includes a safety check to prevent any database modification
by disallowing keywords like CREATE, MERGE, DELETE, etc.
"""
cypher_query = args.get("cypher_query")
if not cypher_query:
return {"error": "Cypher query cannot be empty."}
# Safety Check: Prevent any write operations to the database.
# This check first removes all string literals and then checks for forbidden keywords.
forbidden_keywords = ['CREATE', 'MERGE', 'DELETE', 'SET', 'REMOVE', 'DROP', 'CALL apoc']
# Regex to match single or double quoted strings, handling escaped quotes.
string_literal_pattern = r'"(?:\\.|[^"\\])*"|\'(?:\\.|[^\'\\])*\''
# Remove all string literals from the query.
query_without_strings = re.sub(string_literal_pattern, '', cypher_query)
# Now, check for forbidden keywords in the query without strings.
for keyword in forbidden_keywords:
if re.search(r'\b' + keyword + r'\b', query_without_strings, re.IGNORECASE):
return {
"error": "This tool only supports read-only queries. Prohibited keywords like CREATE, MERGE, DELETE, SET, etc., are not allowed."
}
try:
debug_log(f"Executing Cypher query: {cypher_query}")
with self.db_manager.get_driver().session() as session:
result = session.run(cypher_query)
# Convert results to a list of dictionaries for clean JSON serialization.
records = [record.data() for record in result]
return {
"success": True,
"query": cypher_query,
"record_count": len(records),
"results": records
}
except CypherSyntaxError as e:
debug_log(f"Cypher syntax error: {str(e)}")
return {
"error": "Cypher syntax error.",
"details": str(e),
"query": cypher_query
}
except Exception as e:
debug_log(f"Error executing Cypher query: {str(e)}")
return {
"error": "An unexpected error occurred while executing the query.",
"details": str(e)
}
def find_dead_code_tool(self, **args) -> Dict[str, Any]:
"""Tool to find potentially dead code across the entire project."""
exclude_decorated_with = args.get("exclude_decorated_with", [])
try:
debug_log("Finding dead code.")
results = self.code_finder.find_dead_code(exclude_decorated_with=exclude_decorated_with)
return {
"success": True,
"query_type": "dead_code",
"results": results
}
except Exception as e:
debug_log(f"Error finding dead code: {str(e)}")
return {"error": f"Failed to find dead code: {str(e)}"}
def calculate_cyclomatic_complexity_tool(self, **args) -> Dict[str, Any]:
"""Tool to calculate cyclomatic complexity for a given function."""
function_name = args.get("function_name")
file_path = args.get("file_path")
try:
debug_log(f"Calculating cyclomatic complexity for function: {function_name}")
results = self.code_finder.get_cyclomatic_complexity(function_name, file_path)
response = {
"success": True,
"function_name": function_name,
"results": results
}
if file_path:
response["file_path"] = file_path
return response
except Exception as e:
debug_log(f"Error calculating cyclomatic complexity: {str(e)}")
return {"error": f"Failed to calculate cyclomatic complexity: {str(e)}"}
def find_most_complex_functions_tool(self, **args) -> Dict[str, Any]:
"""Tool to find the most complex functions."""
limit = args.get("limit", 10)
try:
debug_log(f"Finding the top {limit} most complex functions.")
results = self.code_finder.find_most_complex_functions(limit)
return {
"success": True,
"limit": limit,
"results": results
}
except Exception as e:
debug_log(f"Error finding most complex functions: {str(e)}")
return {"error": f"Failed to find most complex functions: {str(e)}"}
def list_indexed_repositories_tool(self, **args) -> Dict[str, Any]:
"""Tool to list indexed repositories."""
try:
debug_log("Listing indexed repositories.")
results = self.code_finder.list_indexed_repositories()
return {
"success": True,
"repositories": results
}
except Exception as e:
debug_log(f"Error listing indexed repositories: {str(e)}")
return {"error": f"Failed to list indexed repositories: {str(e)}"}
def delete_repository_tool(self, **args) -> Dict[str, Any]:
"""Tool to delete a repository from the graph."""
repo_path = args.get("repo_path")
try:
debug_log(f"Deleting repository: {repo_path}")
self.graph_builder.delete_repository_from_graph(repo_path)
return {
"success": True,
"message": f"Repository '{repo_path}' deleted successfully."
}
except Exception as e:
debug_log(f"Error deleting repository: {str(e)}")
return {"error": f"Failed to delete repository: {str(e)}"}
def visualize_graph_query_tool(self, **args) -> Dict[str, Any]:
"""Tool to generate a Neo4j browser visualization URL for a Cypher query."""
cypher_query = args.get("cypher_query")
if not cypher_query:
return {"error": "Cypher query cannot be empty."}
try:
encoded_query = urllib.parse.quote(cypher_query)
visualization_url = f"http://localhost:7474/browser/?cmd=edit&arg={encoded_query}"
return {
"success": True,
"visualization_url": visualization_url,
"message": "Open the URL in your browser to visualize the graph query. The query will be pre-filled for editing."
}
except Exception as e:
debug_log(f"Error generating visualization URL: {str(e)}")
return {"error": f"Failed to generate visualization URL: {str(e)}"}
def list_watched_paths_tool(self, **args) -> Dict[str, Any]:
"""Tool to list all currently watched directory paths."""
try:
paths = self.code_watcher.list_watched_paths()
return {"success": True, "watched_paths": paths}
except Exception as e:
return {"error": f"Failed to list watched paths: {str(e)}"}
def unwatch_directory_tool(self, **args) -> Dict[str, Any]:
"""Tool to stop watching a directory."""
path = args.get("path")
if not path:
return {"error": "Path is a required argument."}
# The watcher class handles the logic of checking if the path is watched
# and returns an error dictionary if not, so we can just call it.
return self.code_watcher.unwatch_directory(path)
def watch_directory_tool(self, **args) -> Dict[str, Any]:
"""
Tool implementation to start watching a directory for changes.
This tool is now smart: it checks if the path exists and if it has already been indexed.
"""
path = args.get("path")
if not path:
return {"error": "Path is a required argument."}
path_obj = Path(path).resolve()
path_str = str(path_obj)
# 1. Validate the path before the try...except block
if not path_obj.is_dir():
return {
"success": True,
"status": "path_not_found",
"message": f"Path '{path_str}' does not exist or is not a directory."
}
try:
# Check if already watching
if path_str in self.code_watcher.watched_paths:
return {"success": True, "message": f"Already watching directory: {path_str}"}
# 2. Check if the repository is already indexed
indexed_repos = self.list_indexed_repositories_tool().get("repositories", [])
is_already_indexed = any(Path(repo["path"]).resolve() == path_obj for repo in indexed_repos)
# 3. Decide whether to perform an initial scan
if is_already_indexed:
# If already indexed, just start the watcher without a scan
self.code_watcher.watch_directory(path_str, perform_initial_scan=False)
return {
"success": True,
"message": f"Path '{path_str}' is already indexed. Now watching for live changes."
}
else:
# If not indexed, perform the scan AND start the watcher
scan_job_result = self.add_code_to_graph_tool(path=path_str, is_dependency=False)
if "error" in scan_job_result:
return scan_job_result
self.code_watcher.watch_directory(path_str, perform_initial_scan=True)
return {
"success": True,
"message": f"Path '{path_str}' was not indexed. Started initial scan and now watching for live changes.",
"job_id": scan_job_result.get("job_id"),
"details": "Use check_job_status to monitor the initial scan."
}
except Exception as e:
error_logger(f"Failed to start watching directory {path}: {e}")
return {"error": f"Failed to start watching directory: {str(e)}"}
def add_code_to_graph_tool(self, **args) -> Dict[str, Any]:
"""
Tool implementation to index a directory of code.
This creates a background job and runs the indexing asynchronously
so the AI assistant can continue to be responsive.
"""
path = args.get("path")
is_dependency = args.get("is_dependency", False)
try:
path_obj = Path(path).resolve()
if not path_obj.exists():
return {
"success": True,
"status": "path_not_found",
"message": f"Path '{path}' does not exist."
}
# Prevent re-indexing the same repository.
indexed_repos = self.list_indexed_repositories_tool().get("repositories", [])
for repo in indexed_repos:
if Path(repo["path"]).resolve() == path_obj:
return {
"success": False,
"message": f"Repository '{path}' is already indexed."
}
# Estimate time and create a job for the user to track.
total_files, estimated_time = self.graph_builder.estimate_processing_time(path_obj)
job_id = self.job_manager.create_job(str(path_obj), is_dependency)
self.job_manager.update_job(job_id, total_files=total_files, estimated_duration=estimated_time)
# Create the coroutine for the background task and schedule it on the main event loop.
coro = self.graph_builder.build_graph_from_path_async(
path_obj, is_dependency, job_id
)
asyncio.run_coroutine_threadsafe(coro, self.loop)
debug_log(f"Started background job {job_id} for path: {str(path_obj)}, is_dependency: {is_dependency}")
return {
"success": True, "job_id": job_id,
"message": f"Background processing started for {str(path_obj)}",
"estimated_files": total_files,
"estimated_duration_seconds": round(estimated_time, 2),
"estimated_duration_human": f"{int(estimated_time // 60)}m {int(estimated_time % 60)}s" if estimated_time >= 60 else f"{int(estimated_time)}s",
"instructions": f"Use 'check_job_status' with job_id '{job_id}' to monitor progress"
}
except Exception as e:
debug_log(f"Error creating background job: {str(e)}")
return {"error": f"Failed to start background processing: {str(e)}"}
def add_package_to_graph_tool(self, **args) -> Dict[str, Any]:
"""Tool to add a package to the graph by auto-discovering its location"""
package_name = args.get("package_name")
language = args.get("language")
is_dependency = args.get("is_dependency", True)
if not language:
return {"error": "The 'language' parameter is required."}
try:
# Check if the package is already indexed
indexed_repos = self.list_indexed_repositories_tool().get("repositories", [])
for repo in indexed_repos:
if repo.get("is_dependency") and (repo.get("name") == package_name or repo.get("name") == f"{package_name}.py"):
return {
"success": False,
"message": f"Package '{package_name}' is already indexed."
}
package_path = get_local_package_path(package_name, language)
if not package_path:
return {"error": f"Could not find package '{package_name}' for language '{language}'. Make sure it's installed."}
if not os.path.exists(package_path):
return {"error": f"Package path '{package_path}' does not exist"}
path_obj = Path(package_path)
total_files, estimated_time = self.graph_builder.estimate_processing_time(path_obj)
job_id = self.job_manager.create_job(package_path, is_dependency)
self.job_manager.update_job(job_id, total_files=total_files, estimated_duration=estimated_time)
coro = self.graph_builder.build_graph_from_path_async(
path_obj, is_dependency, job_id
)
asyncio.run_coroutine_threadsafe(coro, self.loop)
debug_log(f"Started background job {job_id} for package: {package_name} at {package_path}, is_dependency: {is_dependency}")
return {
"success": True, "job_id": job_id, "package_name": package_name,
"discovered_path": package_path,
"message": f"Background processing started for package '{package_name}'",
"estimated_files": total_files,
"estimated_duration_seconds": round(estimated_time, 2),
"estimated_duration_human": f"{int(estimated_time // 60)}m {int(estimated_time % 60)}s" if estimated_time >= 60 else f"{int(estimated_time)}s",
"instructions": f"Use 'check_job_status' with job_id '{job_id}' to monitor progress"
}
except Exception as e:
debug_log(f"Error creating background job for package {package_name}: {str(e)}")
return {"error": f"Failed to start background processing for package '{package_name}': {str(e)}"}
def check_job_status_tool(self, **args) -> Dict[str, Any]:
"""Tool to check job status"""
job_id = args.get("job_id")
if not job_id:
return {"error": "Job ID is a required argument."}
try:
job = self.job_manager.get_job(job_id)
if not job:
return {
"success": True, # Return success to avoid generic error wrapper
"status": "not_found",
"message": f"Job with ID '{job_id}' not found. The ID may be incorrect or the job may have been cleared after a server restart."
}
job_dict = asdict(job)
if job.status == JobStatus.RUNNING:
if job.estimated_time_remaining:
remaining = job.estimated_time_remaining
job_dict["estimated_time_remaining_human"] = (
f"{int(remaining // 60)}m {int(remaining % 60)}s"
if remaining >= 60 else f"{int(remaining)}s"
)
if job.start_time:
elapsed = (datetime.now() - job.start_time).total_seconds()
job_dict["elapsed_time_human"] = (
f"{int(elapsed // 60)}m {int(elapsed % 60)}s"
if elapsed >= 60 else f"{int(elapsed)}s"
)
elif job.status == JobStatus.COMPLETED and job.start_time and job.end_time:
duration = (job.end_time - job.start_time).total_seconds()
job_dict["actual_duration_human"] = (
f"{int(duration // 60)}m {int(duration % 60)}s"
if duration >= 60 else f"{int(duration)}s"
)
job_dict["start_time"] = job.start_time.strftime("%Y-%m-%d %H:%M:%S")
if job.end_time:
job_dict["end_time"] = job.end_time.strftime("%Y-%m-%d %H:%M:%S")
job_dict["status"] = job.status.value
return {"success": True, "job": job_dict}
except Exception as e:
debug_log(f"Error checking job status: {str(e)}")
return {"error": f"Failed to check job status: {str(e)}"}
def list_jobs_tool(self) -> Dict[str, Any]:
"""Tool to list all jobs"""
try:
jobs = self.job_manager.list_jobs()
jobs_data = []
for job in jobs:
job_dict = asdict(job)
job_dict["status"] = job.status.value
job_dict["start_time"] = job.start_time.strftime("%Y-%m-%d %H:%M:%S")
if job.end_time:
job_dict["end_time"] = job.end_time.strftime("%Y-%m-%d %H:%M:%S")
jobs_data.append(job_dict)
jobs_data.sort(key=lambda x: x["start_time"], reverse=True)
return {"success": True, "jobs": jobs_data, "total_jobs": len(jobs_data)}
except Exception as e:
debug_log(f"Error listing jobs: {str(e)}")
return {"error": f"Failed to list jobs: {str(e)}"}
def analyze_code_relationships_tool(self, **args) -> Dict[str, Any]:
"""Tool to analyze code relationships"""
query_type = args.get("query_type")
target = args.get("target")
context = args.get("context")
if not query_type or not target:
return {
"error": "Both 'query_type' and 'target' are required",
"supported_query_types": [
"find_callers", "find_callees", "find_importers", "who_modifies",
"class_hierarchy", "overrides", "dead_code", "call_chain",
"module_deps", "variable_scope", "find_complexity"
]
}
try:
debug_log(f"Analyzing relationships: {query_type} for {target}")
results = self.code_finder.analyze_code_relationships(query_type, target, context)
return {
"success": True, "query_type": query_type, "target": target,
"context": context, "results": results
}
except Exception as e:
debug_log(f"Error analyzing relationships: {str(e)}")
return {"error": f"Failed to analyze relationships: {str(e)}"}
@staticmethod
def _normalize(text: str) -> str:
return text.lower().replace("_", " ").strip()
def find_code_tool(self, **args) -> Dict[str, Any]:
"""Tool to find relevant code snippets"""
query = args.get("query")
fuzzy_search = args.get("fuzzy_search", DEFAULT_FUZZY_SEARCH)
edit_distance = args.get("edit_distance", DEFAULT_EDIT_DISTANCE)
if fuzzy_search:
query = self._normalize(query)
try:
debug_log(f"Finding code for query: {query} with fuzzy_search={fuzzy_search}, edit_distance={edit_distance}")
results = self.code_finder.find_related_code(query, fuzzy_search, edit_distance)
return {"success": True, "query": query, "results": results}
except Exception as e:
debug_log(f"Error finding code: {str(e)}")
return {"error": f"Failed to find code: {str(e)}"}
async def handle_tool_call(self, tool_name: str, args: Dict[str, Any]) -> Dict[str, Any]:
"""
Routes a tool call from the AI assistant to the appropriate handler function.
Args:
tool_name: The name of the tool to execute.
args: A dictionary of arguments for the tool.
Returns:
A dictionary containing the result of the tool execution.
"""
tool_map: Dict[str, Coroutine] = {
"add_package_to_graph": self.add_package_to_graph_tool,
"find_dead_code": self.find_dead_code_tool,
"find_code": self.find_code_tool,
"analyze_code_relationships": self.analyze_code_relationships_tool,
"watch_directory": self.watch_directory_tool,
"execute_cypher_query": self.execute_cypher_query_tool,
"add_code_to_graph": self.add_code_to_graph_tool,
"check_job_status": self.check_job_status_tool,
"list_jobs": self.list_jobs_tool,
"calculate_cyclomatic_complexity": self.calculate_cyclomatic_complexity_tool,
"find_most_complex_functions": self.find_most_complex_functions_tool,
"list_indexed_repositories": self.list_indexed_repositories_tool,
"delete_repository": self.delete_repository_tool,
"visualize_graph_query": self.visualize_graph_query_tool,
"list_watched_paths": self.list_watched_paths_tool,
"unwatch_directory": self.unwatch_directory_tool
}
handler = tool_map.get(tool_name)
if handler:
# Run the synchronous tool function in a separate thread to avoid
# blocking the main asyncio event loop.
return await asyncio.to_thread(handler, **args)
else:
return {"error": f"Unknown tool: {tool_name}"}
async def run(self):
"""
Runs the main server loop, listening for JSON-RPC requests from stdin.
This loop continuously reads lines from stdin, parses them as JSON-RPC
requests, and routes them to the appropriate handlers (e.g., initialize,
tools/list, tools/call). The response is then printed to stdout.
"""
debug_logger("MCP Server is running. Waiting for requests...")
self.code_watcher.start()
loop = asyncio.get_event_loop()
while True:
try:
# Read a request from the standard input.
line = await loop.run_in_executor(None, sys.stdin.readline)
if not line:
debug_logger("Client disconnected (EOF received). Shutting down.")
break
request = json.loads(line.strip())
method = request.get('method')
params = request.get('params', {})
request_id = request.get('id')
response = {}
# Route the request based on the JSON-RPC method.
if method == 'initialize':
response = {
"jsonrpc": "2.0", "id": request_id,
"result": {
"protocolVersion": "2025-03-26",
"serverInfo": {
"name": "CodeGraphContext", "version": "0.1.0",
"systemPrompt": LLM_SYSTEM_PROMPT
},
"capabilities": {"tools": {"listTools": True}},
}
}
elif method == 'tools/list':
# Return the list of tools defined in _init_tools.
response = {
"jsonrpc": "2.0", "id": request_id,
"result": {"tools": list(self.tools.values())}
}
elif method == 'tools/call':
# Execute a tool call and return the result.
tool_name = params.get('name')
args = params.get('arguments', {})
result = await self.handle_tool_call(tool_name, args)
if "error" in result:
response = {
"jsonrpc": "2.0", "id": request_id,
"error": {"code": -32000, "message": "Tool execution error", "data": result}
}
else:
response = {
"jsonrpc": "2.0", "id": request_id,
"result": {"content": [{"type": "text", "text": json.dumps(result, indent=2)}]}
}
elif method == 'notifications/initialized':
# This is a notification, no response needed.
pass
else:
# Handle unknown methods.
if request_id is not None:
response = {
"jsonrpc": "2.0", "id": request_id,
"error": {"code": -32601, "message": f"Method not found: {method}"}
}
# Send the response to standard output if it's not a notification.
if request_id is not None and response:
print(json.dumps(response), flush=True)
except Exception as e:
error_logger(f"Error processing request: {e}\n{traceback.format_exc()}")
request_id = "unknown"
if 'request' in locals() and isinstance(request, dict):
request_id = request.get('id', "unknown")
error_response = {
"jsonrpc": "2.0", "id": request_id,
"error": {"code": -32603, "message": f"Internal error: {str(e)}", "data": traceback.format_exc()}
}
print(json.dumps(error_response), flush=True)
def shutdown(self):
"""Gracefully shuts down the server and its components."""
debug_logger("Shutting down server...")
self.code_watcher.stop()
self.db_manager.close_driver()
```
--------------------------------------------------------------------------------
/src/codegraphcontext/tools/code_finder.py:
--------------------------------------------------------------------------------
```python
# src/codegraphcontext/tools/code_finder.py
import logging
import re
from typing import Any, Dict, List, Literal
from pathlib import Path
from ..core.database import DatabaseManager
logger = logging.getLogger(__name__)
class CodeFinder:
"""Module for finding relevant code snippets and analyzing relationships."""
def __init__(self, db_manager: DatabaseManager):
self.db_manager = db_manager
self.driver = self.db_manager.get_driver()
def format_query(self, find_by: Literal["Class", "Function"], fuzzy_search:bool) -> str:
"""Format the search query based on the search type and fuzzy search settings."""
return f"""
CALL db.index.fulltext.queryNodes("code_search_index", $search_term) YIELD node, score
WITH node, score
WHERE node:{find_by} {'AND node.name CONTAINS $search_term' if not fuzzy_search else ''}
RETURN node.name as name, node.file_path as file_path, node.line_number as line_number,
node.source as source, node.docstring as docstring, node.is_dependency as is_dependency
ORDER BY score DESC
LIMIT 20
"""
def find_by_function_name(self, search_term: str, fuzzy_search: bool) -> List[Dict]:
"""Find functions by name matching using the full-text index."""
with self.driver.session() as session:
if fuzzy_search:
formatted_search_term = f"name:{search_term}"
result = session.run(self.format_query("Function", fuzzy_search), search_term=formatted_search_term)
else:
result = session.run(self.format_query("Function", fuzzy_search), search_term=search_term)
return [dict(record) for record in result]
def find_by_class_name(self, search_term: str, fuzzy_search: bool) -> List[Dict]:
"""Find classes by name matching using the full-text index."""
with self.driver.session() as session:
if fuzzy_search:
formatted_search_term = f"name:{search_term}"
result = session.run(self.format_query("Class", fuzzy_search), search_term=formatted_search_term)
else:
result = session.run(self.format_query("Class", fuzzy_search), search_term=search_term)
return [dict(record) for record in result]
def find_by_variable_name(self, search_term: str) -> List[Dict]:
"""Find variables by name matching"""
with self.driver.session() as session:
result = session.run("""
MATCH (v:Variable)
WHERE v.name CONTAINS $search_term OR v.name =~ $regex_pattern
RETURN v.name as name, v.file_path as file_path, v.line_number as line_number,
v.value as value, v.context as context, v.is_dependency as is_dependency
ORDER BY v.is_dependency ASC, v.name
LIMIT 20
""", search_term=search_term, regex_pattern=f"(?i).*{re.escape(search_term)}.*")
return [dict(record) for record in result]
def find_by_content(self, search_term: str) -> List[Dict]:
"""Find code by content matching in source or docstrings using the full-text index."""
with self.driver.session() as session:
result = session.run("""
CALL db.index.fulltext.queryNodes("code_search_index", $search_term) YIELD node, score
WITH node, score
WHERE node:Function OR node:Class OR node:Variable
RETURN
CASE
WHEN node:Function THEN 'function'
WHEN node:Class THEN 'class'
ELSE 'variable'
END as type,
node.name as name, node.file_path as file_path,
node.line_number as line_number, node.source as source,
node.docstring as docstring, node.is_dependency as is_dependency
ORDER BY score DESC
LIMIT 20
""", search_term=search_term)
return [dict(record) for record in result]
def find_related_code(self, user_query: str, fuzzy_search: bool, edit_distance: int) -> Dict[str, Any]:
"""Find code related to a query using multiple search strategies"""
if fuzzy_search:
user_query_normalized = " ".join(map(lambda x: f"{x}~{edit_distance}", user_query.split(" ")))
else:
user_query_normalized = user_query
results = {
"query": user_query_normalized,
"functions_by_name": self.find_by_function_name(user_query_normalized, fuzzy_search),
"classes_by_name": self.find_by_class_name(user_query_normalized, fuzzy_search),
"variables_by_name": self.find_by_variable_name(user_query), # no fuzzy for variables as they are not using full-text index
"content_matches": self.find_by_content(user_query_normalized)
}
all_results = []
for func in results["functions_by_name"]:
func["search_type"] = "function_name"
func["relevance_score"] = 0.9 if not func["is_dependency"] else 0.7
all_results.append(func)
for cls in results["classes_by_name"]:
cls["search_type"] = "class_name"
cls["relevance_score"] = 0.8 if not cls["is_dependency"] else 0.6
all_results.append(cls)
for var in results["variables_by_name"]:
var["search_type"] = "variable_name"
var["relevance_score"] = 0.7 if not var["is_dependency"] else 0.5
all_results.append(var)
for content in results["content_matches"]:
content["search_type"] = "content"
content["relevance_score"] = 0.6 if not content["is_dependency"] else 0.4
all_results.append(content)
all_results.sort(key=lambda x: x["relevance_score"], reverse=True)
results["ranked_results"] = all_results[:15]
results["total_matches"] = len(all_results)
return results
def find_functions_by_argument(self, argument_name: str, file_path: str = None) -> List[Dict]:
"""Find functions that take a specific argument name."""
with self.driver.session() as session:
if file_path:
query = """
MATCH (f:Function)-[:HAS_PARAMETER]->(p:Parameter)
WHERE p.name = $argument_name AND f.file_path = $file_path
RETURN f.name AS function_name, f.file_path AS file_path, f.line_number AS line_number,
f.docstring AS docstring, f.is_dependency AS is_dependency
ORDER BY f.is_dependency ASC, f.file_path, f.line_number
LIMIT 20
"""
result = session.run(query, argument_name=argument_name, file_path=file_path)
else:
query = """
MATCH (f:Function)-[:HAS_PARAMETER]->(p:Parameter)
WHERE p.name = $argument_name
RETURN f.name AS function_name, f.file_path AS file_path, f.line_number AS line_number,
f.docstring AS docstring, f.is_dependency AS is_dependency
ORDER BY f.is_dependency ASC, f.file_path, f.line_number
LIMIT 20
"""
result = session.run(query, argument_name=argument_name)
return [dict(record) for record in result]
def find_functions_by_decorator(self, decorator_name: str, file_path: str = None) -> List[Dict]:
"""Find functions that have a specific decorator applied to them."""
with self.driver.session() as session:
if file_path:
query = """
MATCH (f:Function)
WHERE f.file_path = $file_path AND $decorator_name IN f.decorators
RETURN f.name AS function_name, f.file_path AS file_path, f.line_number AS line_number,
f.docstring AS docstring, f.is_dependency AS is_dependency, f.decorators AS decorators
ORDER BY f.is_dependency ASC, f.file_path, f.line_number
LIMIT 20
"""
result = session.run(query, decorator_name=decorator_name, file_path=file_path)
else:
query = """
MATCH (f:Function)
WHERE $decorator_name IN f.decorators
RETURN f.name AS function_name, f.file_path AS file_path, f.line_number AS line_number,
f.docstring AS docstring, f.is_dependency AS is_dependency, f.decorators AS decorators
ORDER BY f.is_dependency ASC, f.file_path, f.line_number
LIMIT 20
"""
result = session.run(query, decorator_name=decorator_name)
return [dict(record) for record in result]
def who_calls_function(self, function_name: str, file_path: str = None) -> List[Dict]:
"""Find what functions call a specific function using CALLS relationships with improved matching"""
with self.driver.session() as session:
if file_path:
result = session.run("""
MATCH (caller:Function)-[call:CALLS]->(target:Function {name: $function_name, file_path: $file_path})
OPTIONAL MATCH (caller_file:File)-[:CONTAINS]->(caller)
RETURN DISTINCT
caller.name as caller_function,
caller.file_path as caller_file_path,
caller.line_number as caller_line_number,
caller.docstring as caller_docstring,
caller.is_dependency as caller_is_dependency,
call.line_number as call_line_number,
call.args as call_args,
call.full_call_name as full_call_name,
call.call_type as call_type,
target.file_path as target_file_path
ORDER BY caller.is_dependency ASC, caller.file_path, caller.line_number
LIMIT 20
""", function_name=function_name, file_path=file_path)
results = [dict(record) for record in result]
if not results:
result = session.run("""
MATCH (target:Function {name: $function_name})
MATCH (caller:Function)-[call:CALLS]->(target)
OPTIONAL MATCH (caller_file:File)-[:CONTAINS]->(caller)
RETURN DISTINCT
caller.name as caller_function,
caller.file_path as caller_file_path,
caller.line_number as caller_line_number,
caller.docstring as caller_docstring,
caller.is_dependency as caller_is_dependency,
call.line_number as call_line_number,
call.args as call_args,
call.full_call_name as full_call_name,
call.call_type as call_type,
target.file_path as target_file_path
ORDER BY caller.is_dependency ASC, caller.file_path, caller.line_number
LIMIT 20
""", function_name=function_name)
results = [dict(record) for record in result]
else:
result = session.run("""
MATCH (target:Function {name: $function_name})
MATCH (caller:Function)-[call:CALLS]->(target)
OPTIONAL MATCH (caller_file:File)-[:CONTAINS]->(caller)
RETURN DISTINCT
caller.name as caller_function,
caller.file_path as caller_file_path,
caller.line_number as caller_line_number,
caller.docstring as caller_docstring,
caller.is_dependency as caller_is_dependency,
call.line_number as call_line_number,
call.args as call_args,
call.full_call_name as full_call_name,
call.call_type as call_type,
target.file_path as target_file_path
ORDER BY caller.is_dependency ASC, caller.file_path, caller.line_number
LIMIT 20
""", function_name=function_name)
results = [dict(record) for record in result]
return results
def what_does_function_call(self, function_name: str, file_path: str = None) -> List[Dict]:
"""Find what functions a specific function calls using CALLS relationships"""
with self.driver.session() as session:
if file_path:
# Convert file_path to absolute path
absolute_file_path = str(Path(file_path).resolve())
result = session.run("""
MATCH (caller:Function {name: $function_name, file_path: $absolute_file_path})
MATCH (caller)-[call:CALLS]->(called:Function)
OPTIONAL MATCH (called_file:File)-[:CONTAINS]->(called)
RETURN DISTINCT
called.name as called_function,
called.file_path as called_file_path,
called.line_number as called_line_number,
called.docstring as called_docstring,
called.is_dependency as called_is_dependency,
call.line_number as call_line_number,
call.args as call_args,
call.full_call_name as full_call_name,
call.call_type as call_type
ORDER BY called.is_dependency ASC, called.name
LIMIT 20
""", function_name=function_name, absolute_file_path=absolute_file_path)
else:
result = session.run("""
MATCH (caller:Function {name: $function_name})
MATCH (caller)-[call:CALLS]->(called:Function)
OPTIONAL MATCH (called_file:File)-[:CONTAINS]->(called)
RETURN DISTINCT
called.name as called_function,
called.file_path as called_file_path,
called.line_number as called_line_number,
called.docstring as called_docstring,
called.is_dependency as called_is_dependency,
call.line_number as call_line_number,
call.args as call_args,
call.full_call_name as full_call_name,
call.call_type as call_type
ORDER BY called.is_dependency ASC, called.name
LIMIT 20
""", function_name=function_name)
return [dict(record) for record in result]
def who_imports_module(self, module_name: str) -> List[Dict]:
"""Find what files import a specific module using IMPORTS relationships"""
with self.driver.session() as session:
result = session.run("""
MATCH (file:File)-[imp:IMPORTS]->(module:Module)
WHERE module.name = $module_name OR module.full_import_name CONTAINS $module_name
OPTIONAL MATCH (repo:Repository)-[:CONTAINS]->(file)
WITH file, repo, COLLECT({
imported_module: module.name,
import_alias: module.alias,
full_import_name: module.full_import_name
}) AS imports
RETURN
file.name AS file_name,
file.path AS file_path,
file.relative_path AS file_relative_path,
file.is_dependency AS file_is_dependency,
repo.name AS repository_name,
imports
ORDER BY file.is_dependency ASC, file.path
LIMIT 20
""", module_name=module_name)
return [dict(record) for record in result]
def who_modifies_variable(self, variable_name: str) -> List[Dict]:
"""Find what functions contain or modify a specific variable"""
with self.driver.session() as session:
result = session.run("""
MATCH (var:Variable {name: $variable_name})
MATCH (container)-[:CONTAINS]->(var)
WHERE container:Function OR container:Class OR container:File
OPTIONAL MATCH (file:File)-[:CONTAINS]->(container)
RETURN DISTINCT
CASE
WHEN container:Function THEN container.name
WHEN container:Class THEN container.name
ELSE 'file_level'
END as container_name,
CASE
WHEN container:Function THEN 'function'
WHEN container:Class THEN 'class'
ELSE 'file'
END as container_type,
COALESCE(container.file_path, file.path) as file_path,
container.line_number as container_line_number,
var.line_number as variable_line_number,
var.value as variable_value,
var.context as variable_context,
COALESCE(container.is_dependency, file.is_dependency, false) as is_dependency
ORDER BY is_dependency ASC, file_path, variable_line_number
LIMIT 20
""", variable_name=variable_name)
return [dict(record) for record in result]
def find_class_hierarchy(self, class_name: str, file_path: str = None) -> Dict[str, Any]:
"""Find class inheritance relationships using INHERITS relationships"""
with self.driver.session() as session:
if file_path:
match_clause = "MATCH (child:Class {name: $class_name, file_path: $file_path})"
else:
match_clause = "MATCH (child:Class {name: $class_name})"
parents_query = f"""
{match_clause}
MATCH (child)-[:INHERITS]->(parent:Class)
OPTIONAL MATCH (parent_file:File)-[:CONTAINS]->(parent)
RETURN DISTINCT
parent.name as parent_class,
parent.file_path as parent_file_path,
parent.line_number as parent_line_number,
parent.docstring as parent_docstring,
parent.is_dependency as parent_is_dependency
ORDER BY parent.is_dependency ASC, parent.name
"""
parents_result = session.run(parents_query, class_name=class_name, file_path=file_path)
children_query = f"""
{match_clause}
MATCH (grandchild:Class)-[:INHERITS]->(child)
OPTIONAL MATCH (child_file:File)-[:CONTAINS]->(grandchild)
RETURN DISTINCT
grandchild.name as child_class,
grandchild.file_path as child_file_path,
grandchild.line_number as child_line_number,
grandchild.docstring as child_docstring,
grandchild.is_dependency as child_is_dependency
ORDER BY grandchild.is_dependency ASC, grandchild.name
"""
children_result = session.run(children_query, class_name=class_name, file_path=file_path)
methods_query = f"""
{match_clause}
MATCH (child)-[:CONTAINS]->(method:Function)
RETURN DISTINCT
method.name as method_name,
method.file_path as method_file_path,
method.line_number as method_line_number,
method.args as method_args,
method.docstring as method_docstring,
method.is_dependency as method_is_dependency
ORDER BY method.is_dependency ASC, method.line_number
"""
methods_result = session.run(methods_query, class_name=class_name, file_path=file_path)
return {
"class_name": class_name,
"parent_classes": [dict(record) for record in parents_result],
"child_classes": [dict(record) for record in children_result],
"methods": [dict(record) for record in methods_result]
}
def find_function_overrides(self, function_name: str) -> List[Dict]:
"""Find all implementations of a function across different classes"""
with self.driver.session() as session:
result = session.run("""
MATCH (class:Class)-[:CONTAINS]->(func:Function {name: $function_name})
OPTIONAL MATCH (file:File)-[:CONTAINS]->(class)
RETURN DISTINCT
class.name as class_name,
class.file_path as class_file_path,
func.name as function_name,
func.line_number as function_line_number,
func.args as function_args,
func.docstring as function_docstring,
func.is_dependency as is_dependency,
file.name as file_name
ORDER BY func.is_dependency ASC, class.name
LIMIT 20
""", function_name=function_name)
return [dict(record) for record in result]
def find_dead_code(self, exclude_decorated_with: List[str] = None) -> Dict[str, Any]:
"""Find potentially unused functions (not called by other functions in the project), optionally excluding those with specific decorators."""
if exclude_decorated_with is None:
exclude_decorated_with = []
with self.driver.session() as session:
result = session.run("""
MATCH (func:Function)
WHERE func.is_dependency = false
AND NOT func.name IN ['main', '__init__', '__main__', 'setup', 'run', '__new__', '__del__']
AND NOT func.name STARTS WITH '_test'
AND NOT func.name STARTS WITH 'test_'
AND ALL(decorator_name IN $exclude_decorated_with WHERE NOT decorator_name IN func.decorators)
WITH func
OPTIONAL MATCH (caller:Function)-[:CALLS]->(func)
WHERE caller.is_dependency = false
WITH func, count(caller) as caller_count
WHERE caller_count = 0
OPTIONAL MATCH (file:File)-[:CONTAINS]->(func)
RETURN
func.name as function_name,
func.file_path as file_path,
func.line_number as line_number,
func.docstring as docstring,
func.context as context,
file.name as file_name
ORDER BY func.file_path, func.line_number
LIMIT 50
""", exclude_decorated_with=exclude_decorated_with)
return {
"potentially_unused_functions": [dict(record) for record in result],
"note": "These functions might be unused, but could be entry points, callbacks, or called dynamically"
}
def find_all_callers(self, function_name: str, file_path: str = None) -> List[Dict]:
"""Find all direct and indirect callers of a specific function."""
with self.driver.session() as session:
if file_path:
# Find functions within the specified file_path that call the target function
query = """
MATCH (f:Function)-[:CALLS*]->(target:Function {name: $function_name, file_path: $file_path})
RETURN DISTINCT f.name AS caller_name, f.file_path AS caller_file_path, f.line_number AS caller_line_number, f.is_dependency AS caller_is_dependency
ORDER BY f.is_dependency ASC, f.file_path, f.line_number
LIMIT 50
"""
result = session.run(query, function_name=function_name, file_path=file_path)
else:
# If no file_path (context) is provided, find all callers of the function by name
query = """
MATCH (f:Function)-[:CALLS*]->(target:Function {name: $function_name})
RETURN DISTINCT f.name AS caller_name, f.file_path AS caller_file_path, f.line_number AS caller_line_number, f.is_dependency AS caller_is_dependency
ORDER BY f.is_dependency ASC, f.file_path, f.line_number
LIMIT 50
"""
result = session.run(query, function_name=function_name)
return [dict(record) for record in result]
def find_all_callees(self, function_name: str, file_path: str = None) -> List[Dict]:
"""Find all direct and indirect callees of a specific function."""
with self.driver.session() as session:
if file_path:
query = """
MATCH (caller:Function {name: $function_name, file_path: $file_path})
MATCH (caller)-[:CALLS*]->(f:Function)
RETURN DISTINCT f.name AS callee_name, f.file_path AS callee_file_path, f.line_number AS callee_line_number, f.is_dependency AS callee_is_dependency
ORDER BY f.is_dependency ASC, f.file_path, f.line_number
LIMIT 50
"""
result = session.run(query, function_name=function_name, file_path=file_path)
else:
query = """
MATCH (caller:Function {name: $function_name})
MATCH (caller)-[:CALLS*]->(f:Function)
RETURN DISTINCT f.name AS callee_name, f.file_path AS callee_file_path, f.line_number AS callee_line_number, f.is_dependency AS callee_is_dependency
ORDER BY f.is_dependency ASC, f.file_path, f.line_number
LIMIT 50
"""
result = session.run(query, function_name=function_name)
return [dict(record) for record in result]
def find_function_call_chain(self, start_function: str, end_function: str, max_depth: int = 5) -> List[Dict]:
"""Find call chains between two functions"""
with self.driver.session() as session:
result = session.run(f"""
MATCH path = shortestPath(
(start:Function {{name: $start_function}})-[:CALLS*1..{max_depth}]->(end:Function {{name: $end_function}})
)
WITH path, nodes(path) as func_nodes, relationships(path) as call_rels
RETURN
[node in func_nodes | {{
name: node.name,
file_path: node.file_path,
line_number: node.line_number,
is_dependency: node.is_dependency
}}] as function_chain,
[rel in call_rels | {{
call_line: rel.line_number,
args: rel.args,
full_call_name: rel.full_call_name
}}] as call_details,
length(path) as chain_length
ORDER BY chain_length ASC
LIMIT 10
""", start_function=start_function, end_function=end_function)
return [dict(record) for record in result]
def find_module_dependencies(self, module_name: str) -> Dict[str, Any]:
"""Find all dependencies and dependents of a module"""
with self.driver.session() as session:
importers_result = session.run("""
MATCH (file:File)-[:IMPORTS]->(module:Module {name: $module_name})
OPTIONAL MATCH (repo:Repository)-[:CONTAINS]->(file)
RETURN DISTINCT
file.name as file_name,
file.path as file_path,
file.is_dependency as file_is_dependency,
repo.name as repository_name
ORDER BY file.is_dependency ASC, file.path
LIMIT 20
""", module_name=module_name)
related_imports_result = session.run("""
MATCH (file:File)-[:IMPORTS]->(target_module:Module {name: $module_name})
MATCH (file)-[:IMPORTS]->(other_module:Module)
WHERE other_module <> target_module
RETURN DISTINCT
other_module.name as related_module,
other_module.alias as module_alias,
count(file) as usage_count
ORDER BY usage_count DESC
LIMIT 20
""", module_name=module_name)
return {
"module_name": module_name,
"imported_by_files": [dict(record) for record in importers_result],
"frequently_used_with": [dict(record) for record in related_imports_result]
}
def find_variable_usage_scope(self, variable_name: str) -> Dict[str, Any]:
"""Find the scope and usage patterns of a variable"""
with self.driver.session() as session:
variable_instances = session.run("""
MATCH (var:Variable {name: $variable_name})
OPTIONAL MATCH (container)-[:CONTAINS]->(var)
WHERE container:Function OR container:Class OR container:File
OPTIONAL MATCH (file:File)-[:CONTAINS]->(var)
RETURN DISTINCT
var.name as variable_name,
var.value as variable_value,
var.line_number as line_number,
var.context as context,
COALESCE(var.file_path, file.path) as file_path,
CASE
WHEN container:Function THEN 'function'
WHEN container:Class THEN 'class'
ELSE 'module'
END as scope_type,
CASE
WHEN container:Function THEN container.name
WHEN container:Class THEN container.name
ELSE 'module_level'
END as scope_name,
var.is_dependency as is_dependency
ORDER BY var.is_dependency ASC, file_path, line_number
""", variable_name=variable_name)
return {
"variable_name": variable_name,
"instances": [dict(record) for record in variable_instances]
}
def analyze_code_relationships(self, query_type: str, target: str, context: str = None) -> Dict[str, Any]:
"""Main method to analyze different types of code relationships with fixed return types"""
query_type = query_type.lower().strip()
try:
if query_type == "find_callers":
results = self.who_calls_function(target, context)
return {
"query_type": "find_callers", "target": target, "context": context, "results": results,
"summary": f"Found {len(results)} functions that call '{target}'"
}
elif query_type == "find_callees":
results = self.what_does_function_call(target, context)
return {
"query_type": "find_callees", "target": target, "context": context, "results": results,
"summary": f"Function '{target}' calls {len(results)} other functions"
}
elif query_type == "find_importers":
results = self.who_imports_module(target)
return {
"query_type": "find_importers", "target": target, "results": results,
"summary": f"Found {len(results)} files that import '{target}'"
}
elif query_type == "find_functions_by_argument":
results = self.find_functions_by_argument(target, context)
return {
"query_type": "find_functions_by_argument", "target": target, "context": context, "results": results,
"summary": f"Found {len(results)} functions that take '{target}' as an argument"
}
elif query_type == "find_functions_by_decorator":
results = self.find_functions_by_decorator(target, context)
return {
"query_type": "find_functions_by_decorator", "target": target, "context": context, "results": results,
"summary": f"Found {len(results)} functions decorated with '{target}'"
}
elif query_type in ["who_modifies", "modifies", "mutations", "changes", "variable_usage"]:
results = self.who_modifies_variable(target)
return {
"query_type": "who_modifies", "target": target, "results": results,
"summary": f"Found {len(results)} containers that hold variable '{target}'"
}
elif query_type in ["class_hierarchy", "inheritance", "extends"]:
results = self.find_class_hierarchy(target, context)
return {
"query_type": "class_hierarchy", "target": target, "results": results,
"summary": f"Class '{target}' has {len(results['parent_classes'])} parents, {len(results['child_classes'])} children, and {len(results['methods'])} methods"
}
elif query_type in ["overrides", "implementations", "polymorphism"]:
results = self.find_function_overrides(target)
return {
"query_type": "overrides", "target": target, "results": results,
"summary": f"Found {len(results)} implementations of function '{target}'"
}
elif query_type in ["dead_code", "unused", "unreachable"]:
results = self.find_dead_code()
return {
"query_type": "dead_code", "results": results,
"summary": f"Found {len(results['potentially_unused_functions'])} potentially unused functions"
}
elif query_type == "find_complexity":
limit = int(context) if context and context.isdigit() else 10
results = self.find_most_complex_functions(limit)
return {
"query_type": "find_complexity", "limit": limit, "results": results,
"summary": f"Found the top {len(results)} most complex functions"
}
elif query_type == "find_all_callers":
results = self.find_all_callers(target, context)
return {
"query_type": "find_all_callers", "target": target, "context": context, "results": results,
"summary": f"Found {len(results)} direct and indirect callers of '{target}'"
}
elif query_type == "find_all_callees":
results = self.find_all_callees(target, context)
return {
"query_type": "find_all_callees", "target": target, "context": context, "results": results,
"summary": f"Found {len(results)} direct and indirect callees of '{target}'"
}
elif query_type in ["call_chain", "path", "chain"]:
if '->' in target:
start_func, end_func = target.split('->', 1)
# max_depth can be passed as context, default to 5 if not provided or invalid
max_depth = int(context) if context and context.isdigit() else 5
results = self.find_function_call_chain(start_func.strip(), end_func.strip(), max_depth)
return {
"query_type": "call_chain", "target": target, "results": results,
"summary": f"Found {len(results)} call chains from '{start_func.strip()}' to '{end_func.strip()}' (max depth: {max_depth})"
}
else:
return {
"error": "For call_chain queries, use format 'start_function->end_function'",
"example": "main->process_data"
}
elif query_type in ["module_deps", "module_dependencies", "module_usage"]:
results = self.find_module_dependencies(target)
return {
"query_type": "module_dependencies", "target": target, "results": results,
"summary": f"Module '{target}' is imported by {len(results['imported_by_files'])} files"
}
elif query_type in ["variable_scope", "var_scope", "variable_usage_scope"]:
results = self.find_variable_usage_scope(target)
return {
"query_type": "variable_scope", "target": target, "results": results,
"summary": f"Variable '{target}' has {len(results['instances'])} instances across different scopes"
}
else:
return {
"error": f"Unknown query type: {query_type}",
"supported_types": [
"find_callers", "find_callees", "find_importers", "who_modifies",
"class_hierarchy", "overrides", "dead_code", "call_chain",
"module_deps", "variable_scope", "find_complexity"
]
}
except Exception as e:
return {
"error": f"Error executing relationship query: {str(e)}",
"query_type": query_type,
"target": target
}
def get_cyclomatic_complexity(self, function_name: str, file_path: str = None) -> List[Dict]:
"""Get the cyclomatic complexity of a function."""
with self.driver.session() as session:
if file_path:
# Use ENDS WITH for flexible path matching
query = """
MATCH (f:Function {name: $function_name})
WHERE f.file_path ENDS WITH $file_path
RETURN f.name as function_name, f.file_path as file_path, f.cyclomatic_complexity as complexity
"""
result = session.run(query, function_name=function_name, file_path=file_path)
else:
query = """
MATCH (f:Function {name: $function_name})
RETURN f.name as function_name, f.file_path as file_path, f.cyclomatic_complexity as complexity
"""
result = session.run(query, function_name=function_name)
return [dict(record) for record in result]
def find_most_complex_functions(self, limit: int = 10) -> List[Dict]:
"""Find the most complex functions based on cyclomatic complexity."""
with self.driver.session() as session:
query = """
MATCH (f:Function)
WHERE f.cyclomatic_complexity IS NOT NULL AND f.is_dependency = false
RETURN f.name as function_name, f.file_path as file_path, f.cyclomatic_complexity as complexity, f.line_number as line_number
ORDER BY f.cyclomatic_complexity DESC
LIMIT $limit
"""
result = session.run(query, limit=limit)
return [dict(record) for record in result]
def list_indexed_repositories(self) -> List[Dict]:
"""List all indexed repositories."""
with self.driver.session() as session:
result = session.run("""
MATCH (r:Repository)
RETURN r.name as name, r.path as path, r.is_dependency as is_dependency
ORDER BY r.name
""")
return [dict(record) for record in result]
```
--------------------------------------------------------------------------------
/tests/test_graph_indexing.py:
--------------------------------------------------------------------------------
```python
import pytest
import os
from .conftest import SAMPLE_PROJECT_PATH
# ==============================================================================
# == EXPECTED RELATIONSHIPS
# ==============================================================================
EXPECTED_STRUCTURE = [
("module_a.py", "foo", "Function"),
("module_a.py", "bar", "Function"),
("module_a.py", "outer", "Function"),
("module_b.py", "helper", "Function"),
("module_b.py", "process_data", "Function"),
("module_b.py", "factorial", "Function"),
("advanced_classes.py", "A", "Class"),
("advanced_classes.py", "B", "Class"),
("advanced_classes.py", "C", "Class"),
("module_a.py", "nested", "Function"),
("advanced_calls.py", "square", "Function"),
("advanced_calls.py", "calls", "Function"),
("advanced_calls.py", "Dummy", "Class"),
("advanced_classes2.py", "Base", "Class"),
("advanced_classes2.py", "Mid", "Class"),
("advanced_classes2.py", "Final", "Class"),
("advanced_classes2.py", "Mixin1", "Class"),
("advanced_classes2.py", "Mixin2", "Class"),
("advanced_classes2.py", "Combined", "Class"),
("advanced_classes2.py", "Point", "Class"),
("advanced_classes2.py", "Color", "Class"),
("advanced_classes2.py", "handle", "Function"),
("advanced_functions.py", "with_defaults", "Function"),
("advanced_functions.py", "with_args_kwargs", "Function"),
("advanced_functions.py", "higher_order", "Function"),
("advanced_functions.py", "return_function", "Function"),
("advanced_imports.py", "outer_import", "Function"),
("advanced_imports.py", "use_random", "Function"),
("async_features.py", "fetch_data", "Function"),
("async_features.py", "main", "Function"),
("callbacks_decorators.py", "executor", "Function"),
("callbacks_decorators.py", "square", "Function"),
("callbacks_decorators.py", "log_decorator", "Function"),
("callbacks_decorators.py", "hello", "Function"),
("class_instantiation.py", "A", "Class"),
("class_instantiation.py", "B", "Class"),
("class_instantiation.py", "Fluent", "Class"),
("function_chains.py", "f1", "Function"),
("function_chains.py", "f2", "Function"),
("function_chains.py", "f3", "Function"),
("function_chains.py", "make_adder", "Function"),
("generators.py", "gen_numbers", "Function"),
("generators.py", "agen_numbers", "Function"),
("generators.py", "async_with_example", "Function"),
("generators.py", "AsyncCM", "Class"),
("datatypes.py", "Point", "Class"),
("datatypes.py", "Color", "Class"),
("complex_classes.py", "Base", "Class"),
("complex_classes.py", "Child", "Class"),
("complex_classes.py", "decorator", "Function"),
("complex_classes.py", "decorated_function", "Function"),
("control_flow.py", "choose_path", "Function"),
("control_flow.py", "ternary", "Function"),
("control_flow.py", "try_except_finally", "Function"),
("control_flow.py", "conditional_inner_import", "Function"),
("control_flow.py", "env_based_import", "Function"),
("circular1.py", "func1", "Function"),
("circular2.py", "func2", "Function"),
("cli_and_dunder.py", "run", "Function"),
("comprehensions_generators.py", "double", "Function"),
("context_managers.py", "FileOpener", "Class"),
("context_managers.py", "use_file", "Function"),
("dynamic_dispatch.py", "add", "Function"),
("dynamic_dispatch.py", "sub", "Function"),
("dynamic_dispatch.py", "mul", "Function"),
("dynamic_dispatch.py", "dispatch_by_key", "Function"),
("dynamic_dispatch.py", "dispatch_by_string", "Function"),
("dynamic_dispatch.py", "partial_example", "Function"),
("dynamic_dispatch.py", "C", "Class"),
("dynamic_dispatch.py", "methodcaller_example", "Function"),
("dynamic_dispatch.py", "dynamic_import_call", "Function"),
("dynamic_imports.py", "import_optional", "Function"),
("dynamic_imports.py", "import_by___import__", "Function"),
("dynamic_imports.py", "importlib_runtime", "Function"),
("import_reexports.py", "core_function", "Function"),
("import_reexports.py", "reexport", "Function"),
("mapping_calls.py", "Dispatcher", "Class"),
("mapping_calls.py", "use_dispatcher", "Function"),
("module_c/submodule1.py", "call_helper_twice", "Function"),
("module_c/submodule2.py", "wrapper", "Function"),
("namespace_pkg/ns_module.py", "ns_func", "Function"),
("pattern_matching.py", "matcher", "Function"),
("typing_examples.py", "typed_func", "Function"),
("typing_examples.py", "union_func", "Function"),
("typing_examples.py", "dict_func", "Function"),
]
EXPECTED_INHERITANCE = [
pytest.param("C", "advanced_classes.py", "A", "advanced_classes.py", id="C inherits from A"),
pytest.param("C", "advanced_classes.py", "B", "advanced_classes.py", id="C inherits from B"),
pytest.param("ConcreteThing", "advanced_classes.py", "AbstractThing", "advanced_classes.py", id="ConcreteThing inherits from AbstractThing"),
pytest.param("Mid", "advanced_classes2.py", "Base", "advanced_classes2.py", id="Mid inherits from Base"),
pytest.param("Final", "advanced_classes2.py", "Mid", "advanced_classes2.py", id="Final inherits from Mid"),
pytest.param("Combined", "advanced_classes2.py", "Mixin1", "advanced_classes2.py", id="Combined inherits from Mixin1"),
pytest.param("Combined", "advanced_classes2.py", "Mixin2", "advanced_classes2.py", id="Combined inherits from Mixin2"),
pytest.param("B", "class_instantiation.py", "A", "class_instantiation.py", id="B inherits from A"),
pytest.param("B", "class_instantiation.py", "A", "class_instantiation.py", marks=pytest.mark.skip(reason="Indexer does not support inheritance via super() calls"), id="B inherits from A via super()"),
pytest.param("Child", "complex_classes.py", "Base", "complex_classes.py", id="Child inherits from Base"),
]
EXPECTED_CALLS = [
pytest.param("foo", "module_a.py", None, "helper", "module_b.py", None, id="module_a.foo->module_b.helper"),
pytest.param("foo", "module_a.py", None, "process_data", "module_b.py", None, id="module_a.foo->module_b.process_data"),
pytest.param("factorial", "module_b.py", None, "factorial", "module_b.py", None, id="module_b.factorial->recursive"),
pytest.param("calls", "advanced_calls.py", None, "square", "advanced_calls.py", None, id="advanced_calls.calls->square"),
pytest.param("call_helper_twice", "module_c/submodule1.py", None, "helper", "module_b.py", None, id="submodule1.call_helper_twice->module_b.helper"),
pytest.param("wrapper", "module_c/submodule2.py", None, "call_helper_twice", "module_c/submodule1.py", None, id="submodule2.wrapper->submodule1.call_helper_twice"),
pytest.param("main", "async_features.py", None, "fetch_data", "async_features.py", None, id="async.main->fetch_data"),
pytest.param("func1", "circular1.py", None, "func2", "circular2.py", None, id="circular1.func1->circular2.func2"),
pytest.param("run", "cli_and_dunder.py", None, "with_defaults", "advanced_functions.py", None, id="cli.run->with_defaults"),
pytest.param("use_dispatcher", "mapping_calls.py", None, "call", "mapping_calls.py", None, id="mapping.use_dispatcher->call"),
pytest.param("calls", "advanced_calls.py", None, "method", "advanced_calls.py", "Dummy", marks=pytest.mark.skip(reason="Dynamic call with getattr is not supported"), id="advanced_calls.calls->Dummy.method"),
pytest.param("both", "advanced_classes2.py", "Combined", "m1", "advanced_classes2.py", "Mixin1", id="advanced_classes2.both->m1"),
pytest.param("both", "advanced_classes2.py", "Combined", "m2", "advanced_classes2.py", "Mixin2", id="advanced_classes2.both->m2"),
pytest.param("executor", "callbacks_decorators.py", None, "square", "callbacks_decorators.py", None, marks=pytest.mark.skip(reason="Dynamic call passing function as argument is not supported"), id="callbacks.executor->square"),
pytest.param("reexport", "import_reexports.py", None, "core_function", "import_reexports.py", None, id="reexport->core_function"),
pytest.param("greet", "class_instantiation.py", "B", "greet", "class_instantiation.py", "A", marks=pytest.mark.skip(reason="super() calls are not supported yet"), id="B.greet->A.greet"),
pytest.param("greet", "complex_classes.py", "Child", "greet", "complex_classes.py", "Base", marks=pytest.mark.skip(reason="super() calls are not supported yet"), id="Child.greet->Base.greet"),
pytest.param("class_method", "complex_classes.py", "Child", "greet", "complex_classes.py", "Child", id="Child.class_method->Child.greet"),
pytest.param("use_file", "context_managers.py", None, "__enter__", "context_managers.py", "FileOpener", marks=pytest.mark.skip(reason="Implicit context manager calls not supported"), id="use_file->FileOpener.__enter__"),
pytest.param("partial_example", "dynamic_dispatch.py", None, "add", "dynamic_dispatch.py", None, marks=pytest.mark.skip(reason="Calls via functools.partial not supported yet"), id="partial_example->add"),
pytest.param("async_with_example", "generators.py", None, "__aenter__", "generators.py", "AsyncCM", marks=pytest.mark.skip(reason="Implicit async context manager calls not supported"), id="async_with_example->AsyncCM.__aenter__"),
pytest.param("call", "mapping_calls.py", "Dispatcher", "start", "mapping_calls.py", "Dispatcher", marks=pytest.mark.skip(reason="Dynamic call via dict lookup not supported"), id="Dispatcher.call->start"),
pytest.param("greet", "class_instantiation.py", None, "greet", "class_instantiation.py", "A", marks=pytest.mark.skip(reason="Indexer does not capture calls to methods of instantiated objects within the same file"), id="A.greet called"),
pytest.param("greet", "class_instantiation.py", None, "greet", "class_instantiation.py", "B", marks=pytest.mark.skip(reason="Indexer does not capture calls to methods of instantiated objects within the same file"), id="B.greet called"),
pytest.param("step1", "class_instantiation.py", "Fluent", "step1", "class_instantiation.py", "Fluent", marks=pytest.mark.skip(reason="Indexer does not capture method chaining calls"), id="Fluent.step1 called"),
pytest.param("step2", "class_instantiation.py", "Fluent", "step2", "class_instantiation.py", "Fluent", marks=pytest.mark.skip(reason="Indexer does not capture method chaining calls"), id="Fluent.step2 called"),
pytest.param("step3", "class_instantiation.py", "Fluent", "step3", "class_instantiation.py", "Fluent", marks=pytest.mark.skip(reason="Indexer does not capture method chaining calls"), id="Fluent.step3 called"),
pytest.param("dynamic", "class_instantiation.py", "B", "lambda", "class_instantiation.py", None, marks=pytest.mark.skip(reason="Dynamic attribute assignment and lambda calls not supported"), id="B.dynamic called"),
pytest.param("add_argument", "cli_and_dunder.py", None, "add_argument", None, None, marks=pytest.mark.skip(reason="Calls to external library methods not fully supported"), id="ArgumentParser.add_argument called"),
pytest.param("parse_args", "cli_and_dunder.py", None, "parse_args", None, None, marks=pytest.mark.skip(reason="Calls to external library methods not fully supported"), id="ArgumentParser.parse_args called"),
pytest.param("print", "cli_and_dunder.py", None, "print", None, None, marks=pytest.mark.skip(reason="Built-in function calls not explicitly indexed"), id="print called"),
pytest.param("double", "comprehensions_generators.py", None, "double", "comprehensions_generators.py", None, marks=pytest.mark.skip(reason="Indexer does not capture calls within comprehensions/generators"), id="double called in list comprehension"),
pytest.param("range", "comprehensions_generators.py", None, "range", None, None, marks=pytest.mark.skip(reason="Built-in function calls not explicitly indexed"), id="range called in list comprehension"),
pytest.param("double", "comprehensions_generators.py", None, "double", "comprehensions_generators.py", None, marks=pytest.mark.skip(reason="Indexer does not capture calls within comprehensions/generators"), id="double called in generator expression"),
pytest.param("range", "comprehensions_generators.py", None, "range", None, None, marks=pytest.mark.skip(reason="Built-in function calls not explicitly indexed"), id="range called in generator expression"),
pytest.param("list", "comprehensions_generators.py", None, "list", None, None, marks=pytest.mark.skip(reason="Built-in function calls not explicitly indexed"), id="list called"),
pytest.param("sorted", "comprehensions_generators.py", None, "sorted", None, None, marks=pytest.mark.skip(reason="Built-in function calls not explicitly indexed"), id="sorted called"),
pytest.param("len", "comprehensions_generators.py", None, "len", None, None, marks=pytest.mark.skip(reason="Built-in function calls not explicitly indexed"), id="len called"),
pytest.param("open", "comprehensions_generators.py", None, "open", None, None, marks=pytest.mark.skip(reason="Built-in function calls not explicitly indexed"), id="open called for write"),
pytest.param("write", "comprehensions_generators.py", None, "write", None, None, marks=pytest.mark.skip(reason="Method calls on built-in types not explicitly indexed"), id="write called"),
pytest.param("open", "comprehensions_generators.py", None, "open", None, None, marks=pytest.mark.skip(reason="Built-in function calls not explicitly indexed"), id="open called for read"),
pytest.param("read", "comprehensions_generators.py", None, "read", None, None, marks=pytest.mark.skip(reason="Method calls on built-in types not explicitly indexed"), id="read called"),
pytest.param("ValueError", "control_flow.py", None, "ValueError", None, None, marks=pytest.mark.skip(reason="Built-in exception constructors not explicitly indexed"), id="ValueError called"),
pytest.param("str", "control_flow.py", None, "str", None, None, marks=pytest.mark.skip(reason="Built-in type constructors not explicitly indexed"), id="str called"),
pytest.param("getenv", "control_flow.py", None, "getenv", None, None, marks=pytest.mark.skip(reason="Calls to external library methods not fully supported"), id="os.getenv called"),
pytest.param("dumps", "control_flow.py", None, "dumps", None, None, marks=pytest.mark.skip(reason="Calls to external library methods not fully supported"), id="json.dumps called"),
pytest.param("namedtuple", "datatypes.py", None, "namedtuple", None, None, marks=pytest.mark.skip(reason="Calls to external library functions not fully supported"), id="namedtuple called"),
pytest.param("DISPATCH", "dynamic_dispatch.py", None, "add", "dynamic_dispatch.py", None, marks=pytest.mark.skip(reason="Dynamic dispatch via dictionary lookup not supported"), id="dispatch_by_key calls add dynamically"),
pytest.param("DISPATCH", "dynamic_dispatch.py", None, "sub", "dynamic_dispatch.py", None, marks=pytest.mark.skip(reason="Dynamic dispatch via dictionary lookup not supported"), id="dispatch_by_key calls sub dynamically"),
pytest.param("DISPATCH", "dynamic_dispatch.py", None, "mul", "dynamic_dispatch.py", None, marks=pytest.mark.skip(reason="Dynamic dispatch via dictionary lookup not supported"), id="dispatch_by_key calls mul dynamically"),
pytest.param("get", "dynamic_dispatch.py", None, "get", None, None, marks=pytest.mark.skip(reason="Calls to built-in dictionary methods not explicitly indexed"), id="globals().get called"),
pytest.param("callable", "dynamic_dispatch.py", None, "callable", None, None, marks=pytest.mark.skip(reason="Built-in function calls not explicitly indexed"), id="callable called"),
pytest.param("partial", "dynamic_dispatch.py", None, "partial", None, None, marks=pytest.mark.skip(reason="Calls to external library functions not fully supported"), id="partial called"),
pytest.param("methodcaller", "dynamic_dispatch.py", None, "methodcaller", None, None, marks=pytest.mark.skip(reason="Calls to external library functions not fully supported"), id="methodcaller called"),
pytest.param("method", "dynamic_dispatch.py", "C", "method", "dynamic_dispatch.py", "C", marks=pytest.mark.skip(reason="Dynamic call via operator.methodcaller not supported"), id="C.method called via methodcaller"),
pytest.param("import_module", "dynamic_dispatch.py", None, "import_module", None, None, marks=pytest.mark.skip(reason="Calls to external library functions not fully supported"), id="importlib.import_module called"),
pytest.param("getattr", "dynamic_dispatch.py", None, "getattr", None, None, marks=pytest.mark.skip(reason="Built-in function calls not explicitly indexed"), id="getattr called"),
pytest.param("dumps", "dynamic_imports.py", None, "dumps", None, None, marks=pytest.mark.skip(reason="Calls to external library methods not fully supported"), id="json.dumps called in import_optional"),
pytest.param("__import__", "dynamic_imports.py", None, "__import__", None, None, marks=pytest.mark.skip(reason="Built-in function calls not explicitly indexed"), id="__import__ called"),
pytest.param("getattr", "dynamic_imports.py", None, "getattr", None, None, marks=pytest.mark.skip(reason="Built-in function calls not explicitly indexed"), id="getattr called in import_by___import__"),
pytest.param("import_module", "dynamic_imports.py", None, "import_module", None, None, marks=pytest.mark.skip(reason="Calls to external library functions not fully supported"), id="importlib.import_module called in importlib_runtime"),
pytest.param("getattr", "dynamic_imports.py", None, "getattr", None, None, marks=pytest.mark.skip(reason="Built-in function calls not explicitly indexed"), id="getattr called in importlib_runtime"),
pytest.param("f3", "function_chains.py", None, "f3", "function_chains.py", None, marks=pytest.mark.skip(reason="Indexer does not capture chained function calls"), id="f3 called"),
pytest.param("f2", "function_chains.py", None, "f2", "function_chains.py", None, marks=pytest.mark.skip(reason="Indexer does not capture chained function calls"), id="f2 called"),
pytest.param("f1", "function_chains.py", None, "f1", "function_chains.py", None, marks=pytest.mark.skip(reason="Indexer does not capture chained function calls"), id="f1 called"),
pytest.param("strip", "function_chains.py", None, "strip", None, None, marks=pytest.mark.skip(reason="Method calls on built-in types not explicitly indexed"), id="strip called"),
pytest.param("lower", "function_chains.py", None, "lower", None, None, marks=pytest.mark.skip(reason="Method calls on built-in types not explicitly indexed"), id="lower called"),
pytest.param("replace", "function_chains.py", None, "replace", None, None, marks=pytest.mark.skip(reason="Method calls on built-in types not explicitly indexed"), id="replace called"),
pytest.param("make_adder", "function_chains.py", None, "make_adder", "function_chains.py", None, marks=pytest.mark.skip(reason="Indexer does not capture calls to functions that return functions"), id="make_adder called"),
pytest.param("adder", "function_chains.py", None, "adder", "function_chains.py", None, marks=pytest.mark.skip(reason="Indexer does not capture calls to inner functions returned by other functions"), id="adder called"),
pytest.param("make_adder", "function_chains.py", None, "make_adder", "function_chains.py", None, marks=pytest.mark.skip(reason="Indexer does not capture calls to functions that return functions in a chain"), id="make_adder called in chain"),
pytest.param("adder", "function_chains.py", None, "adder", "function_chains.py", None, marks=pytest.mark.skip(reason="Indexer does not capture calls to inner functions returned by other functions in a chain"), id="adder called in chain"),
pytest.param("sqrt", "import_reexports.py", None, "sqrt", None, None, marks=pytest.mark.skip(reason="Calls to aliased imported functions not fully supported"), id="m.sqrt called"),
pytest.param("Dispatcher", "mapping_calls.py", None, "Dispatcher", "mapping_calls.py", None, marks=pytest.mark.skip(reason="Indexer does not capture class constructor calls"), id="Dispatcher constructor called"),
pytest.param("str", "pattern_matching.py", None, "str", None, None, marks=pytest.mark.skip(reason="Built-in type constructors not explicitly indexed"), id="str called in pattern matching"),
]
EXPECTED_IMPORTS = [
pytest.param("module_a.py", "math", id="module_a imports math"),
pytest.param("module_a.py", "module_b", id="module_a imports module_b"),
pytest.param("advanced_imports.py", "math", id="advanced_imports imports math"),
pytest.param("advanced_imports.py", "random", id="advanced_imports imports random"),
pytest.param("advanced_imports.py", "sys", id="advanced_imports imports sys"),
pytest.param("async_features.py", "asyncio", id="async_features imports asyncio"),
pytest.param("circular1.py", "func2", id="circular1 imports func2"),
pytest.param("circular2.py", "func1", id="circular2 imports func1"),
pytest.param("cli_and_dunder.py", "argparse", id="cli_and_dunder imports argparse"),
pytest.param("cli_and_dunder.py", "advanced_functions", id="cli_and_dunder imports advanced_functions"),
pytest.param("control_flow.py", "os", id="control_flow imports os"),
pytest.param("datatypes.py", "dataclass", id="datatypes imports dataclass"),
pytest.param("datatypes.py", "enum", id="datatypes imports enum"),
pytest.param("datatypes.py", "namedtuple", id="datatypes imports namedtuple"),
pytest.param("dynamic_dispatch.py", "partial", id="dynamic_dispatch imports partial"),
pytest.param("dynamic_dispatch.py", "operator", id="dynamic_dispatch imports operator"),
pytest.param("dynamic_dispatch.py", "importlib", id="dynamic_dispatch imports importlib"),
pytest.param("dynamic_imports.py", "importlib", id="dynamic_imports imports importlib"),
pytest.param("import_reexports.py", "math", marks=pytest.mark.skip(reason="Indexer does not support aliased imports (e.g., 'import math as m')"), id="import_reexports imports math"),
pytest.param("module_c/submodule1.py", "helper", id="submodule1 imports helper"),
pytest.param("module_c/submodule2.py", "call_helper_twice", id="submodule2 imports call_helper_twice"),
pytest.param("typing_examples.py", "List", id="typing_examples imports List"),
pytest.param("typing_examples.py", "Dict", marks=pytest.mark.skip(reason="Indexer does not capture imports after ','"), id="typing_examples imports Dict"),
pytest.param("typing_examples.py", "Union", marks=pytest.mark.skip(reason="Indexer does not capture imports after ','"), id="typing_examples imports Union"),
pytest.param("control_flow.py", "numpy", marks=pytest.mark.skip(reason="Indexer does not capture conditional imports"), id="control_flow imports numpy (conditional)"),
pytest.param("control_flow.py", "ujson", marks=pytest.mark.skip(reason="Indexer does not capture conditional imports"), id="control_flow imports ujson (conditional)"),
pytest.param("control_flow.py", "json", id="control_flow imports json (conditional)"),
pytest.param("dynamic_imports.py", "ujson", marks=pytest.mark.skip(reason="Indexer does not capture conditional imports"), id="dynamic_imports imports ujson (conditional)"),
pytest.param("dynamic_imports.py", "json", id="dynamic_imports imports json (conditional)"),
]
EXPECTED_PARAMETERS = [
pytest.param("foo", "module_a.py", "x", id="foo has parameter x"),
pytest.param("helper", "module_b.py", "x", id="helper has parameter x"),
pytest.param("process_data", "module_b.py", "data", id="process_data has parameter data"),
pytest.param("factorial", "module_b.py", "n", id="factorial has parameter n"),
pytest.param("square", "advanced_calls.py", "x", id="square has parameter x"),
pytest.param("method", "advanced_calls.py", "x", id="Dummy.method has parameter x"),
pytest.param("with_defaults", "advanced_functions.py", "a", id="with_defaults has parameter a"),
pytest.param("with_defaults", "advanced_functions.py", "b", id="with_defaults has parameter b"),
pytest.param("with_defaults", "advanced_functions.py", "c", id="with_defaults has parameter c"),
pytest.param("higher_order", "advanced_functions.py", "func", id="higher_order has parameter func"),
pytest.param("higher_order", "advanced_functions.py", "data", id="higher_order has parameter data"),
pytest.param("return_function", "advanced_functions.py", "x", id="return_function has parameter x"),
pytest.param("executor", "callbacks_decorators.py", "func", id="executor has parameter func"),
pytest.param("executor", "callbacks_decorators.py", "val", id="executor has parameter val"),
pytest.param("square", "callbacks_decorators.py", "x", id="square has parameter x"),
pytest.param("log_decorator", "callbacks_decorators.py", "fn", id="log_decorator has parameter fn"),
pytest.param("hello", "callbacks_decorators.py", "name", id="hello has parameter name"),
pytest.param("greet", "class_instantiation.py", "self", id="A.greet has parameter self"),
pytest.param("greet", "class_instantiation.py", "self", id="B.greet has parameter self"),
pytest.param("step1", "class_instantiation.py", "self", id="Fluent.step1 has parameter self"),
pytest.param("step2", "class_instantiation.py", "self", id="Fluent.step2 has parameter self"),
pytest.param("step3", "class_instantiation.py", "self", id="Fluent.step3 has parameter self"),
pytest.param("run", "cli_and_dunder.py", "argv", id="run has parameter argv"),
pytest.param("greet", "complex_classes.py", "self", id="Base.greet has self"),
pytest.param("greet", "complex_classes.py", "self", id="Child.greet has self"),
pytest.param("static_method", "complex_classes.py", "x", id="Child.static_method has x"),
pytest.param("class_method", "complex_classes.py", "cls", id="Child.class_method has cls"),
pytest.param("class_method", "complex_classes.py", "y", id="Child.class_method has y"),
pytest.param("decorator", "complex_classes.py", "func", id="decorator has func"),
pytest.param("decorated_function", "complex_classes.py", "x", id="decorated_function has x"),
pytest.param("double", "comprehensions_generators.py", "x", id="double has parameter x"),
pytest.param("__enter__", "context_managers.py", "self", id="FileOpener.__enter__ has self"),
pytest.param("__exit__", "context_managers.py", "self", id="FileOpener.__exit__ has self"),
pytest.param("__exit__", "context_managers.py", "exc_type", id="FileOpener.__exit__ has exc_type"),
pytest.param("__exit__", "context_managers.py", "exc_val", id="FileOpener.__exit__ has exc_val"),
pytest.param("__exit__", "context_managers.py", "exc_tb", id="FileOpener.__exit__ has exc_tb"),
pytest.param("choose_path", "control_flow.py", "x", id="choose_path has parameter x"),
pytest.param("ternary", "control_flow.py", "x", id="ternary has parameter x"),
pytest.param("try_except_finally", "control_flow.py", "x", id="try_except_finally has parameter x"),
pytest.param("conditional_inner_import", "control_flow.py", "use_numpy", id="conditional_inner_import has use_numpy"),
pytest.param("add", "dynamic_dispatch.py", "a", id="add has a"),
pytest.param("add", "dynamic_dispatch.py", "b", id="add has b"),
pytest.param("sub", "dynamic_dispatch.py", "a", id="sub has a"),
pytest.param("sub", "dynamic_dispatch.py", "b", id="sub has b"),
pytest.param("mul", "dynamic_dispatch.py", "a", id="mul has a"),
pytest.param("mul", "dynamic_dispatch.py", "b", id="mul has b"),
pytest.param("dispatch_by_key", "dynamic_dispatch.py", "name", id="dispatch_by_key has name"),
pytest.param("dispatch_by_key", "dynamic_dispatch.py", "a", id="dispatch_by_key has a"),
pytest.param("dispatch_by_key", "dynamic_dispatch.py", "b", id="dispatch_by_key has b"),
pytest.param("method", "dynamic_dispatch.py", "x", id="C.method has x"),
pytest.param("methodcaller_example", "dynamic_dispatch.py", "x", id="methodcaller_example has x"),
pytest.param("import_by___import__", "dynamic_imports.py", "name", id="import_by___import__ has name"),
pytest.param("importlib_runtime", "dynamic_imports.py", "name", id="importlib_runtime has name"),
pytest.param("importlib_runtime", "dynamic_imports.py", "attr", id="importlib_runtime has attr"),
pytest.param("f1", "function_chains.py", "x", id="f1 has x"),
pytest.param("f2", "function_chains.py", "x", id="f2 has x"),
pytest.param("f3", "function_chains.py", "x", id="f3 has x"),
pytest.param("make_adder", "function_chains.py", "n", id="make_adder has n"),
pytest.param("gen_numbers", "generators.py", "n", id="gen_numbers has n"),
pytest.param("agen_numbers", "generators.py", "n", id="agen_numbers has n"),
pytest.param("call", "mapping_calls.py", "cmd", id="Dispatcher.call has cmd"),
pytest.param("use_dispatcher", "mapping_calls.py", "cmd", id="use_dispatcher has cmd"),
pytest.param("call_helper_twice", "module_c/submodule1.py", "x", id="call_helper_twice has x"),
pytest.param("wrapper", "module_c/submodule2.py", "x", id="wrapper has x"),
pytest.param("matcher", "pattern_matching.py", "x", id="matcher has x"),
pytest.param("typed_func", "typing_examples.py", "a", marks=pytest.mark.skip(reason="Indexer does not support parameters with type hints"), id="typed_func has a"),
pytest.param("typed_func", "typing_examples.py", "b", marks=pytest.mark.skip(reason="Indexer does not support parameters with type hints"), id="typed_func has b"),
pytest.param("union_func", "typing_examples.py", "x", marks=pytest.mark.skip(reason="Indexer does not support parameters with type hints"), id="union_func has x"),
pytest.param("dict_func", "typing_examples.py", "d", marks=pytest.mark.skip(reason="Indexer does not support parameters with type hints"), id="dict_func has d"),
pytest.param("wrapper", "complex_classes.py", "*args", marks=pytest.mark.skip(reason="Indexer does not capture variadic parameters (*args)"), id="wrapper has *args"),
pytest.param("wrapper", "complex_classes.py", "**kwargs", marks=pytest.mark.skip(reason="Indexer does not capture variadic parameters (**kwargs)"), id="wrapper has **kwargs"),
pytest.param("dispatch_by_string", "dynamic_dispatch.py", "*args", marks=pytest.mark.skip(reason="Indexer does not capture variadic parameters (*args)"), id="dispatch_by_string has *args"),
pytest.param("dispatch_by_string", "dynamic_dispatch.py", "**kwargs", marks=pytest.mark.skip(reason="Indexer does not capture variadic parameters (**kwargs)"), id="dispatch_by_string has **kwargs"),
pytest.param("dynamic_import_call", "dynamic_dispatch.py", "*args", marks=pytest.mark.skip(reason="Indexer does not capture variadic parameters (*args)"), id="dynamic_import_call has *args"),
pytest.param("dynamic_import_call", "dynamic_dispatch.py", "**kwargs", marks=pytest.mark.skip(reason="Indexer does not capture variadic parameters (**kwargs)"), id="dynamic_import_call has **kwargs"),
pytest.param("adder", "function_chains.py", "x", id="adder has x"),
pytest.param("__aenter__", "generators.py", "self", id="AsyncCM.__aenter__ has self"),
pytest.param("__aexit__", "generators.py", "self", id="AsyncCM.__aexit__ has self"),
pytest.param("__aexit__", "generators.py", "exc_type", id="AsyncCM.__aexit__ has exc_type"),
pytest.param("__aexit__", "generators.py", "exc_val", id="AsyncCM.__aexit__ has exc_val"),
pytest.param("__aexit__", "generators.py", "exc_tb", id="AsyncCM.__aexit__ has exc_tb"),
pytest.param("__init__", "mapping_calls.py", "self", id="Dispatcher.__init__ has self"),
pytest.param("start", "mapping_calls.py", "self", id="Dispatcher.start has self"),
pytest.param("stop", "mapping_calls.py", "self", id="Dispatcher.stop has self"),
pytest.param("ns_func", "namespace_pkg/ns_module.py", None, marks=pytest.mark.skip(reason="Functions with no parameters are not explicitly tested for parameter existence"), id="ns_func has no parameters"),
]
EXPECTED_CLASS_METHODS = [
pytest.param("A", "advanced_classes.py", "foo", id="A contains foo"),
pytest.param("B", "advanced_classes.py", "foo", id="B contains foo"),
pytest.param("C", "advanced_classes.py", "bar", id="C contains bar"),
pytest.param("AbstractThing", "advanced_classes.py", "do", id="AbstractThing contains do"),
pytest.param("ConcreteThing", "advanced_classes.py", "do", id="ConcreteThing contains do"),
pytest.param("Dummy", "advanced_calls.py", "method", id="Dummy contains method"),
pytest.param("Mixin1", "advanced_classes2.py", "m1", id="Mixin1 contains m1"),
pytest.param("Mixin2", "advanced_classes2.py", "m2", id="Mixin2 contains m2"),
pytest.param("Combined", "advanced_classes2.py", "both", id="Combined contains both"),
pytest.param("Point", "advanced_classes2.py", "magnitude", id="Point contains magnitude"),
pytest.param("Color", "advanced_classes2.py", "is_primary", id="Color contains is_primary"),
pytest.param("A", "class_instantiation.py", "greet", id="A contains greet"),
pytest.param("B", "class_instantiation.py", "greet", id="B contains greet"),
pytest.param("Fluent", "class_instantiation.py", "step1", id="Fluent contains step1"),
pytest.param("Fluent", "class_instantiation.py", "step2", id="Fluent contains step2"),
pytest.param("Fluent", "class_instantiation.py", "step3", id="Fluent contains step3"),
pytest.param("AsyncCM", "generators.py", "__aenter__", id="AsyncCM contains __aenter__"),
pytest.param("AsyncCM", "generators.py", "__aexit__", id="AsyncCM contains __aexit__"),
pytest.param("Base", "complex_classes.py", "greet", id="Base contains greet"),
pytest.param("Child", "complex_classes.py", "greet", id="Child contains greet"),
pytest.param("Child", "complex_classes.py", "static_method", id="Child contains static_method"),
pytest.param("Child", "complex_classes.py", "class_method", id="Child contains class_method"),
pytest.param("FileOpener", "context_managers.py", "__enter__", id="FileOpener contains __enter__"),
pytest.param("FileOpener", "context_managers.py", "__exit__", id="FileOpener contains __exit__"),
pytest.param("C", "dynamic_dispatch.py", "method", id="C contains method"),
pytest.param("Dispatcher", "mapping_calls.py", "__init__", id="Dispatcher contains __init__"),
pytest.param("Dispatcher", "mapping_calls.py", "start", id="Dispatcher contains start"),
pytest.param("Dispatcher", "mapping_calls.py", "stop", id="Dispatcher contains stop"),
pytest.param("Dispatcher", "mapping_calls.py", "call", id="Dispatcher contains call"),
]
EXPECTED_FUNCTION_CONTAINS = [
pytest.param("return_function", "advanced_functions.py", "inner", id="return_function contains inner"),
pytest.param("log_decorator", "callbacks_decorators.py", "wrapper", id="log_decorator contains wrapper"),
pytest.param("make_adder", "function_chains.py", "adder", id="make_adder contains adder"),
pytest.param("decorator", "complex_classes.py", "wrapper", id="decorator contains wrapper"),
]
EXPECTED_DECORATORS = [
pytest.param("decorated_function", "complex_classes.py", "decorator", "complex_classes.py", id="decorated_function decorated by decorator"),
pytest.param("hello", "callbacks_decorators.py", "log_decorator", "callbacks_decorators.py", id="hello decorated by log_decorator"),
]
EXPECTED_EMPTY_FILES = [
pytest.param("edge_cases/comments_only.py", id="comments_only.py is empty"),
pytest.param("edge_cases/docstring_only.py", id="docstring_only.py is empty"),
pytest.param("edge_cases/empty.py", id="empty.py is empty"),
pytest.param("edge_cases/syntax_error.py", marks=pytest.mark.skip(reason="File with syntax error should be skipped or handled gracefully"), id="syntax_error.py is skipped"),
pytest.param("module_c/__init__.py", id="module_c/__init__.py is empty"),
]
# ==============================================================================
# == TEST IMPLEMENTATIONS
# ==============================================================================
def check_query(graph, query, description):
"""Helper function to execute a Cypher query and assert that a match is found."""
try:
result = graph.query(query)
except Exception as e:
pytest.fail(f"Query failed for {description} with error: {e}\nQuery was:\n{query}")
assert result is not None, f"Query for {description} returned None.\nQuery was:\n{query}"
assert len(result) > 0, f"Query for {description} returned no records.\nQuery was:\n{query}"
assert result[0].get('count', 0) > 0, f"No match found for {description}.\nQuery was:\n{query}"
@pytest.mark.parametrize("file_name, item_name, item_label", EXPECTED_STRUCTURE)
def test_file_contains_item(graph, file_name, item_name, item_label):
"""Verifies that a File node correctly CONTAINS a Function or Class node."""
description = f"CONTAINS from [{file_name}] to [{item_name}]"
abs_file_path = os.path.join(SAMPLE_PROJECT_PATH, file_name)
query = f"""
MATCH (f:File {{path: '{abs_file_path}'}})-[:CONTAINS]->(item:{item_label} {{name: '{item_name}'}})
RETURN count(*) AS count
"""
check_query(graph, query, description)
@pytest.mark.parametrize("child_name, child_file, parent_name, parent_file", EXPECTED_INHERITANCE)
def test_inheritance_relationship(graph, child_name, child_file, parent_name, parent_file):
"""Verifies that an INHERITS relationship exists between two classes."""
description = f"INHERITS from [{child_name}] to [{parent_name}]"
child_path = os.path.join(SAMPLE_PROJECT_PATH, child_file)
parent_path = os.path.join(SAMPLE_PROJECT_PATH, parent_file)
query = f"""
MATCH (child:Class {{name: '{child_name}', file_path: '{child_path}'}})-[:INHERITS]->(parent:Class {{name: '{parent_name}', file_path: '{parent_path}'}})
RETURN count(*) as count
"""
check_query(graph, query, description)
@pytest.mark.parametrize("caller_name, caller_file, caller_class, callee_name, callee_file, callee_class", EXPECTED_CALLS)
def test_function_call_relationship(graph, caller_name, caller_file, caller_class, callee_name, callee_file, callee_class):
"""Verifies that a CALLS relationship exists by checking for nodes first, then the relationship."""
caller_path = os.path.join(SAMPLE_PROJECT_PATH, caller_file)
callee_path = os.path.join(SAMPLE_PROJECT_PATH, callee_file)
# Build match clauses for caller and callee
if caller_class:
caller_match = f"(caller_class:Class {{name: '{caller_class}', file_path: '{caller_path}'}})-[:CONTAINS]->(caller:Function {{name: '{caller_name}'}})"
else:
caller_match = f"(caller:Function {{name: '{caller_name}', file_path: '{caller_path}'}})"
if callee_class:
callee_match = f"(callee_class:Class {{name: '{callee_class}', file_path: '{callee_path}'}})-[:CONTAINS]->(callee:Function {{name: '{callee_name}'}})"
else:
callee_match = f"(callee:Function {{name: '{callee_name}', file_path: '{callee_path}'}})"
# 1. Check that the caller node exists
caller_description = f"existence of caller {caller_class or 'Function'} {{name: '{caller_name}'}} in [{caller_file}]"
caller_query = f"""
MATCH {caller_match}
RETURN count(caller) as count
"""
check_query(graph, caller_query, caller_description)
# 2. Check that the callee node exists
callee_description = f"existence of callee {callee_class or 'Function'} {{name: '{callee_name}'}} in [{callee_file}]"
callee_query = f"""
MATCH {callee_match}
RETURN count(callee) as count
"""
check_query(graph, callee_query, callee_description)
# 3. Check that the CALLS relationship exists between them
relationship_description = f"CALLS from [{caller_name}] to [{callee_name}]"
relationship_query = f"""
MATCH {caller_match}
MATCH {callee_match}
MATCH (caller)-[:CALLS]->(callee)
RETURN count(*) as count
"""
check_query(graph, relationship_query, relationship_description)
@pytest.mark.parametrize("file_name, module_name", EXPECTED_IMPORTS)
def test_import_relationship(graph, file_name, module_name):
"""Verifies that an IMPORTS relationship exists between a file and a module."""
description = f"IMPORTS from [{file_name}] to [{module_name}]"
abs_file_path = os.path.join(SAMPLE_PROJECT_PATH, file_name)
query = f"""
MATCH (f:File {{path: '{abs_file_path}'}})-[:IMPORTS]->(m:Module {{name: '{module_name}'}})
RETURN count(*) as count
"""
check_query(graph, query, description)
@pytest.mark.parametrize("function_name, file_name, parameter_name", EXPECTED_PARAMETERS)
def test_parameter_relationship(graph, function_name, file_name, parameter_name):
"""Verifies that a HAS_PARAMETER relationship exists between a function and a parameter."""
description = f"HAS_PARAMETER from [{function_name}] to [{parameter_name}]"
abs_file_path = os.path.join(SAMPLE_PROJECT_PATH, file_name)
query = f"""
MATCH (f:Function {{name: '{function_name}', file_path: '{abs_file_path}'}})-[:HAS_PARAMETER]->(p:Parameter {{name: '{parameter_name}'}})
RETURN count(*) as count
"""
check_query(graph, query, description)
@pytest.mark.parametrize("class_name, file_name, method_name", EXPECTED_CLASS_METHODS)
def test_class_method_relationship(graph, class_name, file_name, method_name):
"""Verifies that a CONTAINS relationship exists between a class and a method."""
description = f"CONTAINS from [{class_name}] to [{method_name}]"
abs_file_path = os.path.join(SAMPLE_PROJECT_PATH, file_name)
query = f"""
MATCH (c:Class {{name: '{class_name}', file_path: '{abs_file_path}'}})-[:CONTAINS]->(m:Function {{name: '{method_name}'}})
RETURN count(*) as count
"""
check_query(graph, query, description)
@pytest.mark.parametrize("outer_function_name, file_name, inner_function_name", EXPECTED_FUNCTION_CONTAINS)
def test_function_contains_relationship(graph, outer_function_name, file_name, inner_function_name):
"""Verifies that a CONTAINS relationship exists between an outer function and an inner function."""
description = f"CONTAINS from [{outer_function_name}] to [{inner_function_name}]"
abs_file_path = os.path.join(SAMPLE_PROJECT_PATH, file_name)
query = f"""
MATCH (outer:Function {{name: '{outer_function_name}', file_path: '{abs_file_path}'}})-[:CONTAINS]->(inner:Function {{name: '{inner_function_name}'}})
RETURN count(*) as count
"""
check_query(graph, query, description)
```