Introspection API

ninja_introspect

ninja-introspect — Polyglot database introspection engine for Ninja Stack.

__all__ module-attribute

__all__ = [
    "GraphProvider",
    "IntrospectionEngine",
    "IntrospectionProvider",
    "IntrospectionResult",
    "MongoProvider",
    "SQLProvider",
    "VectorProvider",
]

IntrospectionEngine

IntrospectionEngine(project_name: str = 'untitled')

Orchestrates multiple introspection providers and merges their results.

Usage::

engine = IntrospectionEngine(project_name="my-project")
schema = await engine.run([
    "sqlite+aiosqlite:///path/to/db.sqlite",
    "mongodb://localhost:27017/mydb",
])

Source code in libs/ninja-introspect/src/ninja_introspect/engine.py

def __init__(self, project_name: str = "untitled") -> None:
    self.project_name = project_name

project_name instance-attribute

project_name = project_name

run async

run(
    connection_strings: list[str],
    *,
    providers: dict[str, IntrospectionProvider]
    | None = None,
) -> AgenticSchema

Run introspection across all connection strings and merge results.

PARAMETER	DESCRIPTION
connection_strings	List of database connection URIs. TYPE: list[str]
providers	Optional mapping of connection string → provider override. TYPE: dict[str, IntrospectionProvider] | None DEFAULT: None

RETURNS	DESCRIPTION
AgenticSchema	A merged AgenticSchema containing entities and relationships from all sources.

Source code in libs/ninja-introspect/src/ninja_introspect/engine.py

async def run(
    self,
    connection_strings: list[str],
    *,
    providers: dict[str, IntrospectionProvider] | None = None,
) -> AgenticSchema:
    """Run introspection across all connection strings and merge results.

    Args:
        connection_strings: List of database connection URIs.
        providers: Optional mapping of connection string → provider override.

    Returns:
        A merged AgenticSchema containing entities and relationships from all sources.
    """
    all_entities: list[EntitySchema] = []
    all_relationships: list[RelationshipSchema] = []

    for conn_str in connection_strings:
        if providers and conn_str in providers:
            provider = providers[conn_str]
        else:
            provider = _detect_provider(conn_str)

        result: IntrospectionResult = await provider.introspect(conn_str)
        all_entities.extend(result.entities)
        all_relationships.extend(result.relationships)

    return AgenticSchema(
        project_name=self.project_name,
        entities=all_entities,
        relationships=all_relationships,
    )

IntrospectionProvider

Bases: ABC

Protocol for database introspection providers.

Each provider connects to a specific database type, reads its schema, and produces EntitySchema + RelationshipSchema objects.

introspect abstractmethod async

introspect(connection_string: str) -> IntrospectionResult

Connect to the database and extract schema information.

PARAMETER	DESCRIPTION
connection_string	Database connection URI. TYPE: str

RETURNS	DESCRIPTION
IntrospectionResult	IntrospectionResult containing discovered entities and relationships.

Source code in libs/ninja-introspect/src/ninja_introspect/providers/base.py

@abstractmethod
async def introspect(self, connection_string: str) -> IntrospectionResult:
    """Connect to the database and extract schema information.

    Args:
        connection_string: Database connection URI.

    Returns:
        IntrospectionResult containing discovered entities and relationships.
    """

IntrospectionResult dataclass

IntrospectionResult(
    entities: list[EntitySchema] = list(),
    relationships: list[RelationshipSchema] = list(),
)

Result from a single provider's introspection run.

entities class-attribute instance-attribute

entities: list[EntitySchema] = field(default_factory=list)

relationships class-attribute instance-attribute

relationships: list[RelationshipSchema] = field(
    default_factory=list
)

GraphProvider

GraphProvider(sample_size: int = DEFAULT_SAMPLE_SIZE)

Bases: IntrospectionProvider

Introspects Neo4j graph databases — reads node labels, relationship types, and properties.

Source code in libs/ninja-introspect/src/ninja_introspect/providers/graph.py

def __init__(self, sample_size: int = DEFAULT_SAMPLE_SIZE) -> None:
    self.sample_size = sample_size

sample_size instance-attribute

sample_size = sample_size

introspect async

introspect(connection_string: str) -> IntrospectionResult

Source code in libs/ninja-introspect/src/ninja_introspect/providers/graph.py

async def introspect(self, connection_string: str) -> IntrospectionResult:
    driver = AsyncGraphDatabase.driver(connection_string)
    try:
        return await self._run_introspection(driver)
    finally:
        await driver.close()

MongoProvider

MongoProvider(sample_size: int = DEFAULT_SAMPLE_SIZE)

Bases: IntrospectionProvider

Introspects MongoDB databases by sampling documents to infer schema.

Source code in libs/ninja-introspect/src/ninja_introspect/providers/mongo.py

def __init__(self, sample_size: int = DEFAULT_SAMPLE_SIZE) -> None:
    self.sample_size = sample_size

sample_size instance-attribute

sample_size = sample_size

introspect async

introspect(connection_string: str) -> IntrospectionResult

Source code in libs/ninja-introspect/src/ninja_introspect/providers/mongo.py

async def introspect(self, connection_string: str) -> IntrospectionResult:
    client: AsyncIOMotorClient = AsyncIOMotorClient(connection_string)  # type: ignore[type-arg]
    try:
        db_name = client.get_default_database().name  # type: ignore[union-attr]
        db = client[db_name]

        collection_names: list[str] = await db.list_collection_names()
        # Filter out system collections
        collection_names = [c for c in collection_names if not c.startswith("system.")]

        entities: list[EntitySchema] = []
        for coll_name in sorted(collection_names):
            entity = await self._introspect_collection(db, coll_name)
            if entity is not None:
                entities.append(entity)

        return IntrospectionResult(entities=entities)
    finally:
        client.close()

SQLProvider

Bases: IntrospectionProvider

Introspects SQL databases (Postgres, MySQL, SQLite) via SQLAlchemy.

introspect async

introspect(connection_string: str) -> IntrospectionResult

Source code in libs/ninja-introspect/src/ninja_introspect/providers/sql.py

async def introspect(self, connection_string: str) -> IntrospectionResult:
    engine = create_async_engine(connection_string)
    try:
        return await self._run_introspection(engine)
    finally:
        await engine.dispose()

VectorProvider

Bases: IntrospectionProvider

Introspects vector databases (Chroma) — reads collection metadata.

introspect async

introspect(connection_string: str) -> IntrospectionResult

Introspect a Chroma vector database.

PARAMETER	DESCRIPTION
connection_string	For Chroma, this is the persist directory path or an HTTP URL like http://host:port. TYPE: str

Source code in libs/ninja-introspect/src/ninja_introspect/providers/vector.py

async def introspect(self, connection_string: str) -> IntrospectionResult:
    """Introspect a Chroma vector database.

    Args:
        connection_string: For Chroma, this is the persist directory path
            or an HTTP URL like ``http://host:port``.
    """
    client = self._create_client(connection_string)
    collections = client.list_collections()

    entities: list[EntitySchema] = []
    for collection in collections:
        entity = self._introspect_collection(collection)
        entities.append(entity)

    return IntrospectionResult(entities=entities)