Python SDK Reference

graph LR
    Q[User Query] --> IC[Intent Classifier]
    IC --> PR[Pre-Retrieval]
    PR --> R{Router}
    R --> VS[VectorStoreReal<br/>ChromaDB]
    R --> CPG[CPGQueryService<br/>DuckDB]
    VS --> RRF[RRF Merger]
    CPG --> RRF
    RRF --> RK[ResultRanker]
    RK --> LLM[LLM Provider]
    LLM --> A[Response]

class CPGQueryService(
    CPGQueryBase,
    SubsystemQueriesMixin,
    CallGraphQueriesMixin,
    SecurityQueriesMixin,
    PerformanceQueriesMixin,
    QualityQueriesMixin,
    SemanticQueriesMixin,
    StatisticsQueriesMixin,
    CommentQueriesMixin,
    ExternalQueriesMixin,
    TypeQueriesMixin,
    PatternQueriesMixin,
    CollectionQueriesMixin,
):
    pass  # All methods inherited from mixins

CPGQueryService(
    db_path: Optional[str] = None,
    allowed_db_paths: Optional[set] = None
)

from src.services.cpg import CPGQueryService

# Auto-resolve DB path from active project
service = CPGQueryService()

# Explicit path
service = CPGQueryService("data/projects/myproject.duckdb")

# Context manager
with CPGQueryService("data/projects/myproject.duckdb") as svc:
    stats = svc.get_database_stats()
    hotspots = svc.get_security_hotspots(limit=20)

# Raw SQL with dict results
rows = service.execute_sql_dict("SELECT name, file FROM nodes_method LIMIT 5")
for row in rows:
    print(f"{row['name']} in {row['file']}")

# Switch project at runtime
service.switch_project("postgresql")

subsystems = service.get_subsystems()
for s in subsystems[:3]:
    print(f"{s['name']}: {s['method_count']} methods")

methods = service.get_methods_by_subsystem("executor", limit=50)

callers = service.get_callers("LWLockAcquire", limit=10)
callees = service.get_callees("CommitTransaction")

# Full call graph with depth control
graph = service.get_call_graph(method_id=123, depth=3, direction="both")

hotspots = service.get_security_hotspots(limit=20)
sources = service.get_taint_sources()
sinks = service.get_taint_sinks()

# Data flow-based SQL injection detection
sqli = service.find_sql_injections_dataflow()
for finding in sqli:
    print(f"{finding['method']} in {finding['file']}: {finding['risk']}")

hotspots = service.get_performance_hotspots(limit=10)
recursive = service.get_recursive_methods()

untested = service.get_methods_without_tests()
complex_methods = service.get_complex_methods()
long_methods = service.get_long_methods(min_lines=150, limit=50)

results = service.search_by_comment_content("TODO", limit=20)
pattern_matches = service.search_methods_by_name_pattern(".*Handler$")

stats = service.get_database_stats()
print(f"Methods: {stats['method_count']}, Files: {stats['file_count']}")

git = service.get_git_state()
print(f"Parsed at commit: {git['commit_hash']}")

todos = service.get_todo_comments(limit=50)
stats = service.get_comment_statistics()
descs = service.get_functions_descriptions()

authors = service.get_git_authors()
hotspots = service.get_git_hotspots(min_churn=10, limit=20)
bus_factor = service.get_bus_factor_candidates()

casts = service.find_cast_operations()
poly = service.find_polymorphic_variables()

findings = service.get_pattern_findings()
stats = service.get_pattern_statistics()
rules = service.get_pattern_rules()

collections = service.get_collection_declarations()
kb = service.find_knowledge_base_collections()

VectorStoreReal(
    persist_directory: Optional[str] = None,
    cache_size: int = 100,
    cache_ttl: int = 3600,
    collection_prefix: str = ""
)

from src.retrieval.vector_store_real import VectorStoreReal

store = VectorStoreReal(persist_directory="chroma_db", collection_prefix="postgresql")
store.initialize_collections()

# Search Q&A pairs
qa = store.retrieve_qa("How does transaction commit work?", top_k=3)

# Search SQL examples with type hint
sql = store.retrieve_sql(
    "find callers",
    keywords=["call_graph"],
    query_type="structural",
    top_k=5
)

# Search auto-generated documentation
docs = store.retrieve_generated_docs("buffer manager", top_k=5, language="en")

# Search code comments by type and subsystem
comments = store.retrieve_comments(
    "memory allocation",
    top_k=10,
    comment_type="docstring",
    subsystem="executor"
)

HybridRetriever(
    vector_store,
    cpg_service,
    config: Optional[HybridRetrievalConfig] = None
)

import asyncio
from src.retrieval.hybrid.retriever import HybridRetriever
from src.retrieval.hybrid.models import HybridRetrievalConfig

config = HybridRetrievalConfig(
    vector_weight=0.6,
    graph_weight=0.4,
    final_top_k=10
)

retriever = HybridRetriever(
    vector_store=store,
    cpg_service=service,
    config=config
)

results = asyncio.run(retriever.retrieve(
    query="transaction commit handling",
    mode="hybrid",        # "hybrid", "vector_only", "graph_only"
    query_type="semantic"  # "semantic", "structural", "security"
))

for r in results:
    print(f"[{r.source}] {r.content[:80]}... (score={r.score:.3f})")

@dataclass
class RetrievalResult:
    id: str
    content: str
    score: float
    source: str                                  # "vector", "graph", or "hybrid"
    entity_type: str = "method"                  # "method", "struct", "macro", "type", "caller", "callee"
    metadata: Dict[str, Any] = field(default_factory=dict)
    node_id: Optional[int] = None                # CPG node ID (for deduplication)

@dataclass
class HybridRetrievalConfig:
    vector_weight: float = 0.6
    graph_weight: float = 0.4
    vector_top_k: int = 20
    graph_top_k: int = 20
    final_top_k: int = 10
    min_score_threshold: float = 0.1
    enable_reranking: bool = False

@dataclass
class DomainBoostContext:
    taint_sources: Set[str] = field(default_factory=set)
    taint_sinks: Set[str] = field(default_factory=set)
    target_subsystem: Optional[str] = None
    query_type: Optional[str] = None
    entry_point_boost: float = 1.2
    security_boost: float = 1.5
    subsystem_boost: float = 1.3

ctx = DomainBoostContext.from_config(
    taint_sources={"user_input", "env_var"},
    taint_sinks={"exec_sql", "system"},
    target_subsystem="executor",
    query_type="security"
)

@dataclass
class RelevanceScore:
    total: float
    keyword_match: float = 0.0
    tag_coverage: float = 0.0
    name_match: float = 0.0
    length_bonus: float = 0.0
    semantic_similarity: float = 0.0
    source_confidence: float = 0.0
    retrieval_score: float = 0.0

    def get_breakdown(self) -> Dict[str, float]

score = ranker.rank_result(result, question="find memory leaks")
breakdown = score.get_breakdown()
# {'total': 0.82, 'keyword_match': 0.3, 'semantic_similarity': 0.45, ...}

from src.ranking.result_ranker import ResultRanker

ranker = ResultRanker()

MultiScenarioCopilot()

from src.workflow.orchestration.copilot import MultiScenarioCopilot

copilot = MultiScenarioCopilot()

# Auto-detect scenario from intent
result = copilot.run("Find buffer overflow risks in the parser module")
print(result["answer"])
print(f"Scenario: {result['scenario_id']}, Confidence: {result['confidence']}")

# Force a specific scenario
result = copilot.run(
    "Analyze the executor module",
    context={"scenario_id": "scenario_2"}
)

class MultiScenarioState(TypedDict):
    # Input
    query: str
    context: Optional[Dict[str, Any]]
    language: Optional[str]

    # Intent Classification
    intent: Optional[str]
    scenario_id: Optional[str]
    confidence: Optional[float]
    classification_method: Optional[str]

    # CPG Data
    cpg_results: Optional[List[Dict]]
    subsystems: Optional[List[str]]
    methods: Optional[List[Dict]]
    call_graph: Optional[Any]

    # Output
    answer: Optional[str]
    evidence: Optional[List[str]]
    metadata: Optional[Dict[str, Any]]
    retrieved_functions: Optional[List[str]]

    # Error Handling
    error: Optional[str]
    retry_count: int

    # Workflow Configuration
    enrichment_config: Optional[Dict[str, Any]]
    vector_store: Optional[Any]

    # Multi-tenant scoping
    db_path: Optional[str]
    collection_prefix: Optional[str]

    # Pre-retrieval (Phase E)
    pre_retrieval_results: Optional[List[Dict[str, Any]]]

from src.workflow.state import create_initial_state

state = create_initial_state(
    query="Where is heap_insert defined?",
    context={"subsystem": "executor"}
)

from src.config.cpg_config import CPGConfig

config = CPGConfig()
config.set_cpg_type("postgresql_v2")
title = config.get_code_analyst_title()
# "PostgreSQL 17.6 expert"

from src.domains.registry import DomainRegistry

plugin = DomainRegistry.activate("postgresql")
print(plugin.get_taint_sources())
print(plugin.get_subsystem_names())

# Check active domain
domain = DomainRegistry.get_active_or_none()
if domain:
    print(f"Active domain: {domain.name}")

from src.config import get_unified_config

cfg = get_unified_config()

# Retrieval weights
cfg.retrieval_weights.default_vector_weight   # 0.6
cfg.retrieval_weights.final_top_k             # 10

# Reranking
cfg.reranking.boost_domain_match              # 1.3

# Timeouts (never hardcode)
cfg.timeouts.external_api                     # 30
cfg.timeouts.llm_generation                   # 60

# Batch processing (never hardcode)
cfg.batch_processing.extraction_default_limit # 100
cfg.batch_processing.vector_store_batch       # 100

# Cache
cfg.cache.embedding_cache_size                # 100
cfg.cache.embedding_cache_ttl                 # 3600

@dataclass(frozen=True)
class ProjectContext:
    project_id: UUID
    group_id: UUID
    project_name: str
    db_path: str
    domain: Optional[str] = None
    language: Optional[str] = None
    collection_prefix: str = ""

# In a FastAPI route handler
@router.get("/methods")
async def list_methods(ctx: ProjectContext = Depends(get_project_context)):
    cpg = ProjectScopedServices.get_cpg(ctx)
    return cpg.get_database_stats()

from src.api.services.project_services import ProjectScopedServices

# Initialize from config on startup
ProjectScopedServices.init_from_config()

# Get project-scoped services
cpg = ProjectScopedServices.get_cpg(ctx)
vs = ProjectScopedServices.get_vector_store(ctx)

# Health check
status = ProjectScopedServices.health_check()
# {'cpg_cache_size': 2, 'vector_cache_size': 1, 'max_cache': 10, ...}

from fastapi import Depends
from src.api.dependencies import (
    get_project_context,
    get_cpg_service,
    RequireAdmin,
    require_permission,
)
from src.api.auth.permissions import Permission

@router.post("/admin/reset")
async def reset_cache(user=Depends(RequireAdmin)):
    ProjectScopedServices.reset()

@router.get("/query")
async def run_query(
    ctx=Depends(get_project_context),
    cpg=Depends(get_cpg_service),
    user=Depends(require_permission(Permission.QUERY_EXECUTE)),
):
    return cpg.get_database_stats()

from src.api.errors import LocalizedHTTPException

raise LocalizedHTTPException(
    status_code=401,
    detail_key="invalid_credentials",
    lang="en"
)

# With format parameters
raise LocalizedHTTPException(
    status_code=404,
    detail_key="user_not_found",
    lang="ru",
    username="john"
)

from src.llm.base_provider import LLMProviderError

try:
    response = provider.generate(prompt)
except LLMProviderError as e:
    logger.error(f"LLM generation failed: {e}")

from src.workflow.error_handling import AgentExecutionError

try:
    result = copilot.run("analyze security")
except AgentExecutionError as e:
    logger.error(f"Agent failed: {e}")

from src.project_manager import DatabaseNotConfiguredError
from src.llm.base_provider import LLMProviderError
from src.workflow.error_handling import AgentExecutionError

try:
    service = CPGQueryService()
    result = copilot.run(query)
except DatabaseNotConfiguredError:
    print("No project configured. Use: python -m src.cli import /path/to/source")
except LLMProviderError as e:
    print(f"LLM provider error: {e}")
except AgentExecutionError as e:
    print(f"Workflow error: {e}")

from src.security import HardeningScanner, HardeningCategory, HardeningSeverity

scanner = HardeningScanner(cpg_service=cpg_service, language="c")

findings = scanner.scan_all(limit_per_check=50)
# Returns: [HardeningFinding(d3fend_id='D3-VI', severity='high', ...)]

findings = scanner.scan_by_d3fend_id(["D3-VI", "D3-NPC", "D3-TL"])
# Returns: Findings for Variable Initialization, Null Pointer Checking, Trusted Library

findings = scanner.scan_by_category(HardeningCategory.MEMORY_SAFETY)
# Returns: Findings for all memory safety checks

findings = scanner.scan_by_severity(HardeningSeverity.HIGH)
# Returns: Findings with CRITICAL or HIGH severity

scores = scanner.get_compliance_score(findings)
# Returns: {
#     'overall_score': 85.3,
#     'total_findings': 12,
#     'by_category': {'initialization': 3, 'pointer_safety': 5, ...},
#     'by_d3fend': {'D3-VI': 3, 'D3-NPC': 5, ...},
#     'by_severity': {'high': 2, 'medium': 6, 'low': 4},
#     'category_scores': {'initialization': 70, 'pointer_safety': 50, ...},
#     'd3fend_scores': {'D3-VI': 70, 'D3-NPC': 50, ...}
# }

report = scanner.get_remediation_report(findings)
print(report)
# # D3FEND Source Code Hardening Report
# ## Summary
# - **Overall Compliance Score**: 85.3%
# - **Total Findings**: 12
# ...

summary = scanner.get_checks_summary()
# Returns: {
#     'total_checks': 22,
#     'language': 'c',
#     'by_category': {...},
#     'by_d3fend': {...},
#     'domain_checks': 10
# }

from src.security import HardeningCategory

class HardeningCategory(Enum):
    INITIALIZATION = "initialization"           # D3-VI
    CREDENTIAL_MANAGEMENT = "credential_mgmt"   # D3-CS
    INTEGER_SAFETY = "integer_safety"           # D3-IRV
    POINTER_SAFETY = "pointer_safety"           # D3-PV, D3-NPC, D3-MBSV
    MEMORY_SAFETY = "memory_safety"             # D3-RN
    LIBRARY_SAFETY = "library_safety"           # D3-TL
    TYPE_SAFETY = "type_safety"                 # D3-VTV
    DOMAIN_VALIDATION = "domain_validation"     # D3-DLV
    OPERATIONAL_VALIDATION = "operational"      # D3-OLV

from src.security import HardeningSeverity

class HardeningSeverity(Enum):
    CRITICAL = "critical"  # Directly exploitable
    HIGH = "high"          # Significant security risk
    MEDIUM = "medium"      # Moderate security risk
    LOW = "low"            # Minor security concern
    INFO = "info"          # Best practice recommendation

from src.security import HardeningCheck

@dataclass
class HardeningCheck:
    id: str                    # "D3-VI-001"
    d3fend_id: str             # "D3-VI"
    d3fend_name: str           # "Variable Initialization"
    category: HardeningCategory
    severity: HardeningSeverity
    description: str
    cpgql_query: str           # SQL query for CPG database
    cwe_ids: List[str]         # ["CWE-457"]
    language_scope: List[str]  # ["c", "cpp"] or ["*"]
    indicators: List[str]
    good_patterns: List[str]
    remediation: str
    example_code: str
    confidence_weight: float   # 0.0-1.0

check = get_check_by_id("D3-VI-001")
if check.applies_to_language("c"):
    print("Applies to C code")

from src.security import HardeningFinding

@dataclass
class HardeningFinding:
    finding_id: str      # Unique ID
    check_id: str        # "D3-VI-001"
    d3fend_id: str       # "D3-VI"
    category: str        # "initialization"
    severity: str        # "high"
    method_name: str     # "process_input"
    filename: str        # "src/input.c"
    line_number: int     # 142
    code_snippet: str    # "int x; use(x);"
    description: str
    cwe_ids: List[str]
    remediation: str
    confidence: float    # 0.0-1.0
    metadata: Dict

finding_dict = finding.to_dict()
# Returns: {'finding_id': 'a1b2c3', 'd3fend_id': 'D3-VI', ...}

finding = HardeningFinding.from_check_and_row(check, row, confidence=0.9)

from src.security import (
    get_check_by_id,
    get_checks_by_category,
    get_checks_by_d3fend_id,
    get_all_checks,
    get_checks_for_language,
    D3FEND_TECHNIQUES,
    D3FEND_TECHNIQUE_IDS,
)

check = get_check_by_id("D3-VI-001")

memory_checks = get_checks_by_category(HardeningCategory.MEMORY_SAFETY)

null_checks = get_checks_by_d3fend_id("D3-NPC")

all_checks = get_all_checks()
print(f"Total checks: {len(all_checks)}")

c_checks = get_checks_for_language("c")

# Available D3FEND technique IDs
D3FEND_TECHNIQUE_IDS = [
    "D3-VI",   # Variable Initialization
    "D3-CS",   # Credential Scrubbing
    "D3-IRV",  # Integer Range Validation
    "D3-PV",   # Pointer Validation
    "D3-RN",   # Reference Nullification
    "D3-TL",   # Trusted Library
    "D3-VTV",  # Variable Type Validation
    "D3-MBSV", # Memory Block Start Validation
    "D3-NPC",  # Null Pointer Checking
    "D3-DLV",  # Domain Logic Validation
    "D3-OLV",  # Operational Logic Validation
]

# D3FEND technique metadata
D3FEND_TECHNIQUES = {
    "D3-VI": {
        "name": "Variable Initialization",
        "description": "Setting variables to a known value before use",
        "url": "https://next.d3fend.mitre.org/technique/d3f:VariableInitialization",
    },
    # ... other techniques
}