### Repominify Development Setup

Source: https://github.com/mikewcasale/repominify/blob/main/README.md

Instructions for setting up the Repominify project for development, including cloning the repository, installing dependencies, and running tests.

```bash
# Clone the repository
git clone https://github.com/mikewcasale/repominify.git
cd repominify

# Install in development mode with test dependencies
pip install -e '.[dev]'

# Run tests
pytest tests/
```

--------------------------------

### Install repominify

Source: https://github.com/mikewcasale/repominify/blob/main/README.md

Provides the command to install the repominify package using pip.

```bash
pip install repominify
```

--------------------------------

### Python Imports and Setup

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

This snippet shows the necessary Python imports and the initialization of an APIRouter for defining API endpoints related to streaming query operations.

```python
import inspect
import json
import os
import traceback

import pandas as pd
import yaml
from fastapi import (
    APIRouter,
    HTTPException,
)
from fastapi.responses import StreamingResponse
from graphrag.config import create_graphrag_config
from graphrag.query.api import (
    global_search_streaming as global_search_streaming_internal,
)
from graphrag.query.api import local_search_streaming as local_search_streaming_internal

from src.api.azure_clients import BlobServiceClientSingleton
from src.api.common import (
    sanitize_name,
    validate_index_file_exist,
)
from src.api.query import _is_index_complete
from src.models import GraphRequest
from src.reporting import ReporterSingleton
from src.utils import query as query_helper

from .query import _get_embedding_description_store, _update_context

query_streaming_route = APIRouter(
    prefix="/query/streaming",
    tags=["Query Streaming Operations"],
)
```

--------------------------------

### Python Imports and Setup

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Imports necessary libraries for API development, data processing, and interacting with Azure services. Sets up the APIRouter for defining API endpoints.

```Python
import inspect
import json
import os
import traceback
from typing import Any

import pandas as pd
import yaml
from azure.identity import DefaultAzureCredential
from azure.search.documents import SearchClient
from azure.search.documents.models import VectorizedQuery
from fastapi import (
    APIRouter,
    HTTPException,
)
from graphrag.config import create_graphrag_config
from graphrag.model.types import TextEmbedder
from graphrag.query.api import global_search, local_search
from graphrag.vector_stores.base import (
    BaseVectorStore,
    VectorStoreDocument,
    VectorStoreSearchResult,
)

from src.api.azure_clients import BlobServiceClientSingleton
from src.api.common import (
    sanitize_name,
    validate_index_file_exist,
)
from src.models import (
    GraphRequest,
    GraphResponse,
    PipelineJob,
)
from src.reporting import ReporterSingleton
from src.typing import PipelineJobState
from src.utils import query as query_helper

query_route = APIRouter(
    prefix="/query",
    tags=["Query Operations"],
)
```

--------------------------------

### Start Indexing Pipeline

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

This Python script uses the `argparse` module to accept an index name as a command-line argument and then asynchronously starts an indexing pipeline using the `_start_indexing_pipeline` function from the `src.api.index` module.

```Python
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.

import argparse
import asyncio

from src import main  # noqa: F401
from src.api.index import _start_indexing_pipeline

parser = argparse.ArgumentParser(description="Kickoff indexing job.")
parser.add_argument("-i", "--index-name", required=True)
args = parser.parse_args()

asyncio.run(
    _start_indexing_pipeline(
        index_name=args.index_name,
    )
)
```

--------------------------------

### Python Imports and Setup

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

This snippet shows the necessary imports for the project, including standard libraries, third-party packages like asyncio, yaml, azure SDKs, and FastAPI, as well as custom modules from src.api, src.models, and src.reporting.

```python
import asyncio
import inspect
import os
import traceback
from time import time
from typing import cast

import yaml
from azure.identity import DefaultAzureCredential
from azure.search.documents.indexes import SearchIndexClient
from datashaper import WorkflowCallbacksManager
from fastapi import (
    APIRouter,
    HTTPException,
    UploadFile,
)
from graphrag.config import create_graphrag_config
from graphrag.index import create_pipeline_config
from graphrag.index.bootstrap import bootstrap
from graphrag.index.run import run_pipeline_with_config
from kubernetes import (
    client,
    config,
)

from src.api.azure_clients import (
    AzureStorageClientManager,
    BlobServiceClientSingleton,
    get_database_container_client,
)
from src.api.common import (
    delete_blob_container,
    sanitize_name,
    validate_blob_container_name,
)
from src.models import (
    BaseResponse,
    IndexNameList,
    IndexStatusResponse,
    PipelineJob,
)
from src.reporting import ReporterSingleton
from src.reporting.load_reporter import load_pipeline_reporter
from src.reporting.pipeline_job_workflow_callbacks import PipelineJobWorkflowCallbacks
from src.reporting.typing import Reporters
from src.typing import PipelineJobState
```

--------------------------------

### Setup Indexing Pipeline API

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

API endpoint to initiate the building of an indexing pipeline. It handles storage validation, prompt uploads, and checks for existing index jobs, ensuring proper state management and error handling.

```APIDOC
POST /index
Summary: Build an index
Response Model: BaseResponse
Responses:
  200: { model: BaseResponse }

Parameters:
  - name: storage_name
    in: query
    required: true
    type: string
    description: The name of the storage container.
  - name: index_name
    in: query
    required: true
    type: string
    description: The name of the index to build.
  - name: entity_extraction_prompt
    in: formData
    required: false
    type: file
    description: Optional prompt for entity extraction.
  - name: community_report_prompt
    in: formData
    required: false
    type: file
    description: Optional prompt for community reporting.
  - name: summarize_descriptions_prompt
    in: formData
    required: false
    type: file
    description: Optional prompt for summarizing descriptions.

Error Handling:
  - HTTPException(status_code=500, detail=f"Invalid index name: {index_name}")
  - HTTPException(status_code=500, detail=f"Storage blob container {storage_name} does not exist")
  - HTTPException(status_code=202, detail=f"Index '{index_name}' already exists and has not finished building.")

Functionality:
  - Validates index name against blob container naming rules.
  - Checks for the existence of the specified storage blob container.
  - Reads and decodes content from uploaded prompt files.
  - Manages existing index jobs, preventing new jobs if the existing one is scheduled or running.
  - Resets and updates existing jobs in a FAILED state to allow for rescheduling.
  - Updates job status, progress, and prompts in the database.
```

--------------------------------

### AzureStorageClientManager Cosmos Client Methods

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Provides methods to get the CosmosClient and DatabaseProxy instances managed by the AzureStorageClientManager. The get_cosmos_database_client method initializes the database client if it hasn't been already.

```python
from azure.cosmos import CosmosClient, DatabaseProxy

    def get_cosmos_client(self) -> CosmosClient:
        """
        Returns the Cosmos DB client.

        Returns:
            CosmosClient: The Cosmos DB client.
        """
        return self._cosmos_client

    def get_cosmos_database_client(self, database_name: str) -> DatabaseProxy:
        """
        Returns the Cosmos DB database client.

        Args:
            database_name (str): The name of the database.

        Returns:
            DatabaseProxy: The Cosmos DB database client.
        """
        if not hasattr(self, "_cosmos_database_client"):
            self._cosmos_database_client = self._cosmos_client.get_database_client(
                database=database_name
```

--------------------------------

### Get Database Client

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Retrieves a DatabaseProxy object for a specified Cosmos DB database. It utilizes the CosmosClientSingleton to get an authenticated client instance.

```python
from azure.cosmos import DatabaseProxy

def get_database_client(database_name: str) -> DatabaseProxy:
    client = CosmosClientSingleton.get_instance()
    return client.get_database_client(database_name)
```

--------------------------------

### Get Database Container Client

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Retrieves a ContainerProxy object for a specified Cosmos DB database and container. It first obtains the database client using get_database_client.

```python
from azure.cosmos import ContainerProxy

def get_database_container_client(
    database_name: str,
    container_name: str
) -> ContainerProxy:
    db_client = get_database_client(database_name)
    return db_client.get_container_client(container_name)
```

--------------------------------

### Workflow Logging Callbacks

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Handles logging for workflow start, completion, errors, warnings, and general messages. It formats messages with workflow progress and additional details, sending them to the logger.

```python
def on_workflow_start(self, name: str, instance: object) -> None:
        workflow_progress = (
            f" ({len(self._processed_workflow_steps)}/{self._num_workflow_steps})"
            if self._num_workflow_steps
            else ""
        )  # will take the form "(1/4)"
        message += f"Workflow{workflow_progress}: {name} started."
        details = {
            "workflow_name": name,
            # "workflow_instance": str(instance),
        }
        if self._index_name:
            details["index_name"] = self._index_name
        self._logger.info(
            message, stack_info=False, extra=self._format_details(details=details)
        )

    def on_workflow_end(self, name: str, instance: object) -> None:
        message = f"Index: {self._index_name} -- " if self._index_name else ""
        workflow_progress = (
            f" ({len(self._processed_workflow_steps)}/{self._num_num_workflow_steps})"
            if self._num_workflow_steps
            else ""
        )  # will take the form "(1/4)"
        message += f"Workflow{workflow_progress}: {name} complete."
        details = {
            "workflow_name": name,
            # "workflow_instance": str(instance),
        }
        if self._index_name:
            details["index_name"] = self._index_name
        self._logger.info(
            message, stack_info=False, extra=self._format_details(details=details)
        )
```

--------------------------------

### Handling Workflow Start Event

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

The `on_workflow_start` method is a callback executed when a workflow begins. It updates the internal workflow name and adds the workflow to the list of processed steps, preparing a message that includes the index name if available.

```python
    def on_workflow_start(self, name: str, instance: object) -> None:
        """Execute this callback when a workflow starts."""
        self._workflow_name = name
        self._processed_workflow_steps.append(name)
        message = f"Index: {self._index_name} -- " if self._index_name else ""

```

--------------------------------

### Entity Configuration API

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

This section details the API endpoints for managing entity configurations. It includes methods for creating and deleting entity configurations, which are essential for defining and processing entities within the system. The provided examples demonstrate how to interact with these endpoints using a Python client.

```APIDOC
POST /config/entity
  Description: Creates a new entity configuration.
  Request Body:
    entity_configuration_name: string (Name of the entity configuration)
    entity_types: list[string] (List of entity types to configure)
    entity_examples: list[string] (Examples of entities)
  Response:
    status_code: 200 OK

DELETE /{entity_config_name}
  Description: Deletes an existing entity configuration.
  Parameters:
    entity_config_name: string (The name of the entity configuration to delete)
  Response:
    status_code: 200 OK
```

--------------------------------

### Create Entity Configuration Fixture

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

This Python fixture generates a default entity configuration for testing purposes. It utilizes `pytest` and includes sample entity types and example text with expected outputs for entity and relationship extraction. The fixture requires a `client` object to be available.

```python
import time
import uuid

import pytest

index_endpoint = "/index"


@pytest.fixture
def create_entity_config(client):
    entity_config_name = f"default-{str(uuid.uuid4())}"
    entity_types = ["ORGANIZATION", "GEO", "PERSON"]
    entity_examples = [
        {
            "entity_types": "ORGANIZATION, PERSON",
            "text": "The Fed is scheduled to meet on Tuesday and Wednesday, with the central bank planning to release its latest policy decision on Wednesday at 2:00 p.m. ET, followed by a press conference where Fed Chair Jerome Powell will take questions. Investors expect the Federal Open Market Committee to hold its benchmark interest rate steady in a range of 5.25%-5.5%.",
            "output": '("entity"{tuple_delimiter}FED{tuple_delimiter}ORGANIZATION{tuple_delimiter}The Fed is the Federal Reserve, which will set interest rates on Tuesday and Wednesday)\n{record_delimiter}\n("entity"{tuple_delimiter}JEROME POWELL{tuple_delimiter}PERSON{tuple_delimiter}Jerome Powell is the chair of the Federal Reserve)\n{record_delimiter}\n("entity"{tuple_delimiter}FEDERAL OPEN MARKET COMMITTEE{tuple_delimiter}ORGANIZATION{tuple_delimiter}The Federal Reserve committee makes key decisions about interest rates and the growth of the United States money supply)\n{record_delimiter}\n("relationship"{tuple_delimiter}JEROME POWELL{tuple_delimiter}FED{tuple_delimiter}Jerome Powell is the Chair of the Federal Reserve and will answer questions at a press conference{tuple_delimiter}9)\n{completion_delimiter}',
        },
        {
            "entity_types": "ORGANIZATION",
            "text": "Arm's (ARM) stock skyrocketed in its opening day on the Nasdaq Thursday. But IPO experts warn that the British chipmaker's debut on the public markets isn't indicative of how other newly listed companies may perform.\n\nArm, a formerly public company, was taken private by SoftBank in 2016. The well-established chip designer says it powers 99% of premium smartphones.",
            "output": '("entity"{tuple_delimiter}ARM{tuple_delimiter}ORGANIZATION{tuple_delimiter}Arm is a stock now listed on the Nasdaq which powers 99% of premium smartphones)\n{record_delimiter}\n("entity"{tuple_delimiter}SOFTBANK{tuple_delimiter}ORGANIZATION{tuple_delimiter}SoftBank is a firm that previously owned Arm)\n{record_delimiter}\n("relationship"{tuple_delimiter}ARM{tuple_delimiter}SOFTBANK{tuple_delimiter}SoftBank formerly owned Arm from 2016 until present{tuple_delimiter}5)\n{completion_delimiter}',
        },
        {
            "entity_types": "ORGANIZATION,GEO,PERSON",
            "text": "Five Americans jailed for years in Iran and widely regarded as hostages are on their way home to the United States.\n\nThe last pieces in a controversial swap mediated by Qatar fell into place when $6bn (£4.8bn) of Iranian funds held in South Korea reached banks in Doha.\n\nIt triggered the departure of the four men and one woman in Tehran, who are also Iranian citizens, on a chartered flight to Qatar's capital.\n\nThey were met by senior US officials and are now on their way to Washington.\n\nThe Americans include 51-year-old businessman Siamak Namazi, who has spent nearly eight years in Tehran's notorious Evin prison, as well as businessman Emad Shargi, 59, and environmentalist Morad Tahbaz, 67, who also holds British nationality.",
        },
    ]
    return {
        "entity_config_name": entity_config_name,
        "entity_types": entity_types,
        "entity_examples": entity_examples,
    }

```

--------------------------------

### AzureStorageClientManager Blob Service Client Methods

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Provides methods to get the synchronous and asynchronous BlobServiceClient instances managed by the AzureStorageClientManager.

```python
from azure.storage.blob import BlobServiceClient
from azure.storage.blob.aio import BlobServiceClient as BlobServiceClientAsync

    def get_blob_service_client(self) -> BlobServiceClient:
        """
        Returns the blob service client.

        Returns:
            BlobServiceClient: The blob service client.
        """
        return self._blob_service_client

    def get_blob_service_client_async(self) -> BlobServiceClientAsync:
        """
        Returns the asynchronous blob service client.

        Returns:
            BlobServiceClientAsync: The asynchronous blob service client.
        """
        return self._blob_service_client_async
```

--------------------------------

### repominify Command Line Usage

Source: https://github.com/mikewcasale/repominify/blob/main/README.md

Demonstrates basic command-line operations for repominify, including specifying output directories and enabling debug logging.

```bash
# Basic usage
repominify path/to/repomix-output.txt

# Specify output directory
repominify path/to/repomix-output.txt -o output_dir

# Enable debug logging
repominify path/to/repomix-output.txt --debug
```

--------------------------------

### Get Entity

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Retrieves a single entity by its ID from a specified index. It validates the existence of the entity embedding table before reading entity information from Azure Blob Storage.

```python
from fastapi import APIRouter, HTTPException
import pandas as pd
from azure.identity import DefaultAzureCredential

from src.api.common import sanitize_name, validate_index_file_exist
from src.models import EntityResponse
from src.reporting import ReporterSingleton

source_route = APIRouter(
    prefix="/source",
    tags=["Sources"],
)

ENTITY_EMBEDDING_TABLE = "output/create_final_entities.parquet"
storage_account_blob_url = os.environ["STORAGE_ACCOUNT_BLOB_URL"]
storage_account_name = storage_account_blob_url.split("//")[1].split(".")[0]
storage_account_host = storage_account_blob_url.split("//")[1]
storage_options = {
    "account_name": storage_account_name,
    "account_host": storage_account_host,
    "credential": DefaultAzureCredential(),
}

@source_route.get(
    "/entity/{index_name}/{entity_id}",
    summary="Return a single entity.",
    response_model=EntityResponse,
    responses={200: {"model": EntityResponse}},
)
async def get_entity_info(index_name: str, entity_id: int):
    # check for existence of file the query relies on to validate the index is complete
    sanitized_index_name = sanitize_name(index_name)
    validate_index_file_exist(sanitized_index_name, ENTITY_EMBEDDING_TABLE)
    try:
        entity_table = pd.read_parquet(
            f"abfs://{sanitized_index_name}/{ENTITY_EMBEDDING_TABLE}",
            storage_options=storage_options,
        )
        row = entity_table[entity_table.human_readable_id == entity_id]
        return EntityResponse(
            name=row["name"].values[0],
            description=row["description"].values[0],
            text_units=row["text_unit_ids"].values[0].tolist(),
        )
    except Exception:
        reporter = ReporterSingleton().get_instance()
        reporter.on_error("Could not get entity")
```

--------------------------------

### Cosmos Client Singleton

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Provides a singleton instance of the CosmosClient for efficient reuse. It initializes the client using environment variables for the endpoint and DefaultAzureCredential for authentication.

```python
import os

from azure.cosmos import CosmosClient
from azure.identity import DefaultAzureCredential
from environs import Env

class CosmosClientSingleton:
    _instance = None
    _env = Env()

    @classmethod
    def get_instance(cls):
        if cls._instance is None:
            endpoint = os.environ["COSMOS_URI_ENDPOINT"]
            credential = DefaultAzureCredential()
            cls._instance = CosmosClient(endpoint, credential)
        return cls._instance
```

--------------------------------

### Azure Clients and Configuration

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Initialization of Azure client managers and singletons, including BlobServiceClientSingleton and AzureStorageClientManager. It also shows the retrieval of AI Search URL and audience from environment variables.

```python
blob_service_client = BlobServiceClientSingleton.get_instance()
azure_storage_client_manager = (
    AzureStorageClientManager()
)  # TODO: update API to use the AzureStorageClientManager

ai_search_url = os.environ["AI_SEARCH_URL"]
ai_search_audience = os.environ["AI_SEARCH_AUDIENCE"]
```

--------------------------------

### Get Index Job Status

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Retrieves the status of an indexing job for a given index name. It checks for the existence of the index and returns its status, progress, and completion percentage.

```Python
@index_route.get(
    "/status/{index_name}",
    summary="Track the status of an indexing job",
    response_model=IndexStatusResponse,
)
async def get_index_job_status(index_name: str):
    pipelinejob = PipelineJob()  # TODO: fix class so initiliazation is not required
    sanitized_index_name = sanitize_name(index_name)
    if pipelinejob.item_exist(sanitized_index_name):
        pipeline_job = pipelinejob.load_item(sanitized_index_name)
        return IndexStatusResponse(
            status_code=200,
            index_name=pipeline_job.human_readable_index_name,
            storage_name=pipeline_job.human_readable_storage_name,
            status=pipeline_job.status.value,
            percent_complete=pipeline_job.percent_complete,
            progress=pipeline_job.progress,
        )
    raise HTTPException(status_code=404, detail=f"Index '{index_name}' does not exist.")
```

--------------------------------

### Get All Indexes API Endpoint

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Provides an API endpoint to retrieve a list of all available index names. It queries a 'container-store' in a 'graphrag' database to fetch index information.

```APIDOC
@index_route.get(
    "",
    summary="Get all indexes",
    response_model=IndexNameList,
    responses={200: {"model": IndexNameList}},
)
async def get_all_indexes():
    """
    Retrieve a list of all index names.
    """
    items = []
    try:
        container_store_client = get_database_container_client(
            database_name="graphrag", container_name="container-store"
        )
        for item in container_store_client.read_all_items():
            if item["type"] == "index":
                items.append(item["human_readable_name"])
    except Exception:
        reporter = ReporterSingleton().get_instance()
        reporter.on_error("Error retrieving index names")
    return IndexNameList(index_name=items)
```

--------------------------------

### Repomix Project Structure

Source: https://github.com/mikewcasale/repominify/blob/main/README.md

Illustrates the directory layout of the repominify project, detailing the purpose of each subdirectory and key files.

```text
repominify/
├── repominify/         # Source code
│   ├── graph.py        # Graph building and analysis
│   ├── parser.py       # Repomix file parsing
│   ├── types.py        # Core types and data structures
│   ├── exporters.py    # Graph export functionality
│   ├── formatters.py   # Text representation formatting
│   ├── dependencies.py # Dependency management
│   ├── logging.py      # Logging configuration
│   ├── stats.py        # Statistics and comparison
│   ├── constants.py    # Shared constants
│   ├── exceptions.py   # Custom exceptions
│   ├── cli.py         # Command-line interface
│   └── __init__.py    # Package initialization
├── tests/             # Test suite
│   ├── test_end2end.py # End-to-end tests
│   └── data/          # Test data files
├── setup.py          # Package configuration
├── LICENSE           # MIT License
└── README.md         # This file
```

--------------------------------

### Get All Data Storage Containers

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Retrieves a list of all data storage containers managed by the system. It queries a 'container-store' in Cosmos DB and filters for items of type 'data'.

```python
import asyncio
import re
from math import ceil
from typing import List

from azure.storage.blob import ContainerClient
from fastapi import (
    APIRouter,
    HTTPException,
    UploadFile,
)

from src.api.azure_clients import (
    AzureStorageClientManager,
    BlobServiceClientSingletonAsync,
)
from src.api.common import (
    delete_blob_container,
    sanitize_name,
    validate_blob_container_name,
)
from src.models import (
    BaseResponse,
    StorageNameList,
)
from src.reporting import ReporterSingleton

azure_storage_client_manager = AzureStorageClientManager()

data_route = APIRouter(
    prefix="/data",
    tags=["Data Management"],
)


@data_route.get(
    "",
    summary="Get all data storage containers.",
    response_model=StorageNameList,
    responses={200: {"model": StorageNameList}},
)
async def get_all_data_storage_containers():
    """
    Retrieve a list of all data storage containers.
    """
    items = []
    try:
        container_store_client = (
            azure_storage_client_manager.get_cosmos_container_client(
                database_name="graphrag", container_name="container-store"
            )
        )
        for item in container_store_client.read_all_items():
            if item["type"] == "data":
                items.append(item["human_readable_name"])
    except Exception:
        reporter = ReporterSingleton().get_instance()
        reporter.on_error("Error getting list of blob containers.")
        raise HTTPException(
            status_code=500, detail="Error getting list of blob containers."
        )
    return StorageNameList(storage_name=items)
```

--------------------------------

### Azure AI Search Embedding Store Initialization

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Initializes and connects a MultiAzureAISearch client for embedding descriptions. It constructs collection names based on provided index names and sets up the connection to the Azure AI Search service using environment variables.

```python
def _get_embedding_description_store(
    entities: Any,
    vector_store_type: str = Any,
    config_args: dict | None = None,
):
    collection_names = [
        f"{index_name}_description_embedding"
        for index_name in config_args.get("index_names", [])
    ]
    ai_search_url = os.environ["AI_SEARCH_URL"]
    description_embedding_store = MultiAzureAISearch(
        collection_name="multi",
        document_collection=None,
        db_connection=None,
    )
    description_embedding_store.connect(url=ai_search_url)
    for collection_name in collection_names:
        description_embedding_store.add_collection(collection_name)
    return description_embedding_store
```

--------------------------------

### AzureStorageClientManager Initialization and Client Retrieval

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Tests the initialization of AzureStorageClientManager and the retrieval of BlobServiceClient, BlobServiceClientAsync, and CosmosClient. It mocks the `from_connection_string` methods of the respective client classes to return mock client objects, ensuring the manager correctly returns these mocked clients.

```python
from unittest.mock import patch

from azure.cosmos import CosmosClient
from azure.storage.blob import BlobServiceClient
from azure.storage.blob.aio import BlobServiceClient as BlobServiceClientAsync

from src.api.azure_clients import AzureStorageClientManager


class TestAzureStorageClientManager:
    @patch("src.api.azure_clients.BlobServiceClientAsync.from_connection_string")
    @patch("src.api.azure_clients.BlobServiceClient.from_connection_string")
    @patch("src.api.azure_clients.CosmosClient.from_connection_string")
    def get_azure_storage_client_manager(self,
        mock_cosmos_client,
        mock_blob_service_client,
        mock_blob_service_client_async,
    ):
        mock_blob_service_client.return_value = BlobServiceClient
        mock_blob_service_client_async.return_value = BlobServiceClientAsync
        mock_cosmos_client.return_value = CosmosClient
        manager = AzureStorageClientManager()
        return manager

    def test_get_blob_service_client(self):
        manager = self.get_azure_storage_client_manager()
        client = manager.get_blob_service_client()
        assert client == BlobServiceClient

    def test_get_blob_service_client_async(self):
        manager = self.get_azure_storage_client_manager()
        client = manager.get_blob_service_client_async()
        assert client == BlobServiceClientAsync

    def test_get_cosmos_client(self):
        manager = self.get_azure_storage_client_manager()
        client = manager.get_cosmos_client()
        assert client == CosmosClient
```

--------------------------------

### Get Community Report

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Retrieves a single community report by its ID from a specified index. It validates the existence of the necessary data file before attempting to read the report from Azure Blob Storage.

```python
from fastapi import APIRouter, HTTPException
import pandas as pd
from azure.identity import DefaultAzureCredential

from src.api.common import sanitize_name, validate_index_file_exist
from src.models import ReportResponse
from src.reporting import ReporterSingleton

source_route = APIRouter(
    prefix="/source",
    tags=["Sources"],
)

COMMUNITY_REPORT_TABLE = "output/create_final_community_reports.parquet"
storage_account_blob_url = os.environ["STORAGE_ACCOUNT_BLOB_URL"]
storage_account_name = storage_account_blob_url.split("//")[1].split(".")[0]
storage_account_host = storage_account_blob_url.split("//")[1]
storage_options = {
    "account_name": storage_account_name,
    "account_host": storage_account_host,
    "credential": DefaultAzureCredential(),
}

@source_route.get(
    "/report/{index_name}/{report_id}",
    summary="Return a single community report.",
    response_model=ReportResponse,
    responses={200: {"model": ReportResponse}},
)
async def get_report_info(index_name: str, report_id: str):
    # check for existence of file the query relies on to validate the index is complete
    sanitized_index_name = sanitize_name(index_name)
    validate_index_file_exist(sanitized_index_name, COMMUNITY_REPORT_TABLE)
    try:
        report_table = pd.read_parquet(
            f"abfs://{sanitized_index_name}/{COMMUNITY_REPORT_TABLE}",
            storage_options=storage_options,
        )
        row = report_table[report_table.community == report_id]
        return ReportResponse(text=row["full_content"].values[0])
    except Exception:
        reporter = ReporterSingleton().get_instance()
        reporter.on_error("Could not get report.")
        raise HTTPException(
            status_code=500,
            detail=f"Error retrieving report '{report_id}' from index '{index_name}'.",
        )
```

--------------------------------

### Logger Initialization with Azure Monitor

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

The `__init_logger` method sets up the OpenTelemetry logger provider and configures the Azure Monitor Log Exporter. It includes retry logic for logger initialization and ensures the logger is properly configured with handlers and level.

```python
    def __init_logger(self, connection_string, max_logger_init_retries: int = 10):
        max_retry = max_logger_init_retries
        while not (hasattr(self, "_logger")):
            if max_retry == 0:
                raise Exception(
                    "Failed to create logger. Could not disambiguate logger name."
                )

            # generate a unique logger name
            current_time = str(time.time())
            unique_hash = hashlib.sha256(current_time.encode()).hexdigest()
            self._logger_name = f"{self.__class__.__name__}-{unique_hash}"
            if self._logger_name not in logging.Logger.manager.loggerDict:
                # attach azure monitor log exporter to logger provider
                logger_provider = LoggerProvider()
                set_logger_provider(logger_provider)
                exporter = AzureMonitorLogExporter(connection_string=connection_string)
                get_logger_provider().add_log_record_processor(
                    BatchLogRecordProcessor(
                        exporter=exporter,
                        schedule_delay_millis=60000,
                    )
                )
                # instantiate new logger
                self._logger = logging.getLogger(self._logger_name)
                self._logger.propagate = False
                # remove any existing handlers
                self._logger.handlers.clear()
                # fetch handler from logger provider and attach to class
                self._logger.addHandler(LoggingHandler())
                # set logging level
                self._logger.setLevel(logging.DEBUG)

            # reduce sentinel counter value
            max_retry -= 1
```

--------------------------------

### Async Blob Service Client Singleton

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Provides a singleton instance of the asynchronous BlobServiceClient for efficient reuse. It initializes the client using environment variables for the storage account URL and DefaultAzureCredential for authentication. It also includes a method to extract the storage account name from the URL.

```python
import os

from azure.storage.blob.aio import BlobServiceClient as BlobServiceClientAsync
from azure.identity import DefaultAzureCredential
from environs import Env

class BlobServiceClientSingletonAsync:
    _instance = None
    _env = Env()

    @classmethod
    def get_instance(cls):
        if cls._instance is None:
            account_url = os.environ["STORAGE_ACCOUNT_BLOB_URL"]
            credential = DefaultAzureCredential()
            cls._instance = BlobServiceClientAsync(account_url, credential=credential)
        return cls._instance

    @classmethod
    def get_storage_account_name(cls):
        account_url = os.environ["STORAGE_ACCOUNT_BLOB_URL"]
        return account_url.split("//")[1].split(".")[0]
```

--------------------------------

### ConsoleWorkflowCallbacks on_workflow_start

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Callback executed when a workflow begins. It updates the internal workflow name, adds the workflow to the processed list, and logs a message indicating the workflow's start, including progress information.

```python
def on_workflow_start(self, name: str, instance: object) -> None:
        """Execute this callback when a workflow starts."""
        self._workflow_name = name
        self._processed_workflow_steps.append(name)
        message = f"Index: {self._index_name} -- " if self._index_name else ""
        workflow_progress = (
            f" ({len(self._processed_workflow_steps)}/{self._num_workflow_steps})"
            if self._num_workflow_steps
            else ""
        )  # will take the form "(1/4)"
        message += f"Workflow{workflow_progress}: {name} started."
        details = {
            "workflow_name": name,
            # "workflow_instance": str(instance),
        }
        if self._index_name:
            details["index_name"] = self._index_name
        self._logger.info(

```

--------------------------------

### AzureStorageClientManager Initialization

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Initializes the AzureStorageClientManager by setting up environment variables for blob storage URL and Cosmos DB endpoint, and creating instances of BlobServiceClient, BlobServiceClientAsync, and CosmosClient using DefaultAzureCredential.

```python
import os

from azure.cosmos import ContainerProxy, CosmosClient, DatabaseProxy
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient
from azure.storage.blob.aio import BlobServiceClient as BlobServiceClientAsync
from environs import Env

ENDPOINT_ERROR_MSG = "Could not find connection string in environment variables"

from dotenv import load_dotenv

load_dotenv()

class AzureStorageClientManager:
    """
    Manages the Azure storage clients for blob storage and Cosmos DB.

    Attributes:
        azure_storage_blob_url (str): The blob endpoint for azure storage.
        cosmos_uri_endpoint (str): The uri endpoint for the Cosmos DB.
        _blob_service_client (BlobServiceClient): The blob service client.
        _blob_service_client_async (BlobServiceClientAsync): The asynchronous blob service client.
        _cosmos_client (CosmosClient): The Cosmos DB client.
        _cosmos_database_client (DatabaseProxy): The Cosmos DB database client.
        _cosmos_container_client (ContainerProxy): The Cosmos DB container client.
    """

    def __init__(self) -> None:
        self._env = Env()
        self.azure_storage_blob_url = self._env.str(
            "STORAGE_ACCOUNT_BLOB_URL", ENDPOINT_ERROR_MSG
        )
        self.cosmos_uri_endpoint = self._env.str(
            "COSMOS_URI_ENDPOINT", ENDPOINT_ERROR_MSG
        )
        credential = DefaultAzureCredential()
        self._blob_service_client = BlobServiceClient(
            account_url=os.environ["STORAGE_ACCOUNT_BLOB_URL"], credential=credential
        )
        self._blob_service_client_async = BlobServiceClientAsync(
            account_url=os.environ["STORAGE_ACCOUNT_BLOB_URL"], credential=credential
        )
        self._cosmos_client = CosmosClient(
            url=os.environ["COSMOS_URI_ENDPOINT"], credential=credential
        )
```

--------------------------------

### Get Relationship Information

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Retrieves a single relationship's details from a specified index. It ensures that both the relationships and entity embedding files exist before proceeding. The endpoint requires the index name and relationship ID as path parameters.

```APIDOC
GET /relationship/{index_name}/{relationship_id}

Summary: Return a single relationship.

Parameters:
  - index_name (str): The name of the index to retrieve the relationship from.
  - relationship_id (int): The unique identifier of the relationship.

Responses:
  200 OK:
    description: Successfully retrieved relationship information.
    content:
      application/json:
        schema:
          $ref: "#/components/schemas/RelationshipResponse"
  500 Internal Server Error:
    description: An error occurred during retrieval or processing.
    content:
      application/json:
        schema:
          type: object
          properties:
            detail: { type: string }

Example:
GET /relationship/my_index/54321
```

```Python
from fastapi import HTTPException
import pandas as pd
from src.reporting.reporter_singleton import ReporterSingleton
from .utils import sanitize_name, validate_index_file_exist
from .constants import COVARIATES_TABLE, RELATIONSHIPS_TABLE, ENTITY_EMBEDDING_TABLE
from .schemas import ClaimResponse, RelationshipResponse
from .storage import storage_options

@source_route.get(
    "/relationship/{index_name}/{relationship_id}",
    summary="Return a single relationship.",
    response_model=RelationshipResponse,
    responses={200: {"model": RelationshipResponse}},
)
async def get_relationship_info(index_name: str, relationship_id: int):
    # check for existence of file the query relies on to validate the index is complete
    sanitized_index_name = sanitize_name(index_name)
    validate_index_file_exist(sanitized_index_name, RELATIONSHIPS_TABLE)
    validate_index_file_exist(sanitized_index_name, ENTITY_EMBEDDING_TABLE)
    try:
        relationship_table = pd.read_parquet(
            f"abfs://{sanitized_index_name}/{RELATIONSHIPS_TABLE}",
            storage_options=storage_options,
        )
        entity_table = pd.read_parquet(
            f"abfs://{sanitized_index_name}/{ENTITY_EMBEDDING_TABLE}",
            storage_options=storage_options,
        )
        row = relationship_table[
            relationship_table.human_readable_id == str(relationship_id)
        ]
        return RelationshipResponse(
            source=row["source"].values[0],
            source_id=entity_table[
                entity_table.name == row["source"].values[0]
            ].human_readable_id.values[0],
            target=row["target"].values[0],
            target_id=entity_table[
                entity_table.name == row["target"].values[0]
            ].human_readable_id.values[0],
            description=row["description"].values[0],
            text_units=[
                x[0] for x in row["text_unit_ids"].to_list()
            ],  # extract text_unit_ids from a list of panda series
        )
    except Exception:
        reporter = ReporterSingleton().get_instance()
        reporter.on_error("Could not get relationship.")
        raise HTTPException(
            status_code=500,
            detail=f"Error retrieving relationship '{relationship_id}' from index '{index_name}'.",
        )
```

--------------------------------

### PipelineJob Class Methods

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

This section covers the core methods of the PipelineJob class, including creating new jobs, loading existing ones from a database, checking for item existence, calculating completion percentage, and serializing the job data.

```Python
instance._epoch_request_time = int(time())
        instance._human_readable_index_name = human_readable_index_name
        instance._sanitized_index_name = sanitize_name(human_readable_index_name)
        instance._human_readable_storage_name = human_readable_storage_name
        instance._sanitized_storage_name = sanitize_name(human_readable_storage_name)
        instance._entity_extraction_prompt = entity_extraction_prompt
        instance._community_report_prompt = community_report_prompt
        instance._summarize_descriptions_prompt = summarize_descriptions_prompt
        instance._all_workflows = kwargs.get("all_workflows", [])
        instance._completed_workflows = kwargs.get("completed_workflows", [])
        instance._failed_workflows = kwargs.get("failed_workflows", [])
        instance._status = PipelineJobState(
            kwargs.get("status", PipelineJobState.SCHEDULED.value)
        )
        instance._percent_complete = kwargs.get("percent_complete", 0.0)
        instance._progress = kwargs.get("progress", "")

        # Create the item in the database
        instance.update_db()
        return instance
```

```Python
    @classmethod
    def load_item(cls, id: str) -> "PipelineJob":
        """
        This method loads an existing pipeline job from the database and returns
        it as an instance of the PipelineJob class.

        Args:
            id (str): The ID of the pipeline job.

        Returns:
            PipelineJob: The loaded pipeline job instance.
        """
        try:
            db_item = PipelineJob._jobs_container().read_item(item=id, partition_key=id)
        except CosmosHttpResponseError:
            raise ValueError(
                f"Pipeline job with ID {id} does not exist. "
                "Use PipelineJob.create_item() to create a new pipeline job."
            )
        instance = cls.__new__(cls, **db_item)
        instance._id = db_item.get("id")
        instance._epoch_request_time = db_item.get("epoch_request_time")
        instance._index_name = db_item.get("index_name")
        instance._human_readable_index_name = db_item.get("human_readable_index_name")
        instance._sanitized_index_name = db_item.get("sanitized_index_name")
        instance._human_readable_storage_name = db_item.get(
            "human_readable_storage_name"
        )
        instance._sanitized_storage_name = db_item.get("sanitized_storage_name")
        instance._entity_extraction_prompt = db_item.get("entity_extraction_prompt")
        instance._community_report_prompt = db_item.get("community_report_prompt")
        instance._summarize_descriptions_prompt = db_item.get(
            "summarize_descriptions_prompt"
        )
        instance._all_workflows = db_item.get("all_workflows", [])
        instance._completed_workflows = db_item.get("completed_workflows", [])
        instance._failed_workflows = db_item.get("failed_workflows", [])
        instance._status = PipelineJobState(db_item.get("status"))
        instance._percent_complete = db_item.get("percent_complete", 0.0)
        instance._progress = db_item.get("progress", "")
        return instance
```

```Python
    @staticmethod
    def item_exist(id: str) -> bool:
        try:
            PipelineJob._jobs_container().read_item(item=id, partition_key=id)
            return True
        except CosmosHttpResponseError:
            return False
```

```Python
    def calculate_percent_complete(self) -> float:
        """
        This method calculates the percentage of completion of the pipeline job.

        Returns:
            float: The percentage of completion.
        """
        if len(self.completed_workflows) == 0 or len(self.all_workflows) == 0:
            return 0.0
        return round(
            (len(self.completed_workflows) / len(self.all_workflows)) * 100, ndigits=2
        )
```

```Python
    def dump_model(self) -> dict:
        model = {
            "id": self._id,
            "epoch_request_time": self._epoch_request_time,
            "human_readable_index_name": self._human_readable_index_name,
            "sanitized_index_name": self._sanitized_index_name,
            "human_readable_storage_name": self._human_readable_storage_name,
            "sanitized_storage_name": self._sanitized_storage_name,
            "all_workflows": self._all_workflows,
            "completed_workflows": self._completed_workflows,
            "failed_workflows": self._failed_workflows,
            "status": self._status.value,
            "percent_complete": self._percent_complete,
            "progress": self._progress,
        }
        if self._entity_extraction_prompt:
            model["entity_extraction_prompt"] = self._entity_extraction_prompt
        if self._community_report_prompt:
            model["community_report_prompt"] = self._community_report_prompt
        if self._summarize_descriptions_prompt:
            model["summarize_descriptions_prompt"] = self._summarize_descriptions_prompt
        return model
```

```Python
    def update_db(self):
        
```

--------------------------------

### FastAPI Index Router Initialization

Source: https://github.com/mikewcasale/repominify/blob/main/tests/data/graphrag-accelerator.txt

Defines the FastAPI router for index operations, setting the base path to '/index' and applying the 'Index Operations' tag for Swagger UI documentation.

```python
index_route = APIRouter(
    prefix="/index",
    tags=["Index Operations"],
)
```