### Example Usage of GET /ids

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Illustrates how to list all available file IDs using the GET /ids endpoint. This is a primary route for inventorying documents.

```python
import requests

url = "http://localhost:8000/ids"

response = requests.get(url)
print(response.json())
```

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

### Install Pre-commit Formatter

Source: https://github.com/danny-avila/rag_api/blob/main/README.md

Install the pre-commit package and initialize it to enforce code formatting standards, utilizing the black code formatter.

```bash
pip install pre-commit
pre-commit install
```

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

### Install Test Dependencies

Source: https://github.com/danny-avila/rag_api/blob/main/README.md

Before running tests, install all necessary dependencies listed in the test_requirements.txt file using pip.

```bash
pip install -r test_requirements.txt
```

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

### Example Usage of GET /documents

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Illustrates how to retrieve documents by their IDs using the GET /documents endpoint. This is a primary route for accessing specific documents.

```python
import requests

url = "http://localhost:8000/documents"
params = {"ids": "doc123,doc456"}

response = requests.get(url, params=params)
print(response.json())
```

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

### Example Usage of GET /documents/{id}/context

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Demonstrates how to get the full document context for a specific document ID using the GET /documents/{id}/context endpoint.

```python
import requests

url = "http://localhost:8000/documents/doc123/context"

response = requests.get(url)
print(response.json())
```

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

### Example Usage of GET /health

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Demonstrates how to check the health status of the API using the GET /health endpoint. This is a primary route for monitoring.

```python
import requests

url = "http://localhost:8000/health"

response = requests.get(url)
print(response.json())
```

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

### Connection Pooling Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Demonstrates setting up connection pooling for a database. This improves performance by reusing database connections.

```python
import asyncpg

async def create_pool(dsn):
    pool = await asyncpg.create_pool(dsn=dsn, min_size=5, max_size=10)
    return pool

async def main():
    DATABASE_URL = "postgresql://user:password@host:port/database"
    pool = await create_pool(DATABASE_URL)
    print("Connection pool created.")
    # Use the pool to acquire connections for queries
    async with pool.acquire() as connection:
        # Perform database operations
        pass
    await pool.close()
    print("Connection pool closed.")

# asyncio.run(main())
```

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

### Example .env File for RAG API Configuration

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/configuration.md

This snippet shows a comprehensive example of an .env file used to configure the RAG API. It covers settings for the server, vector database (pgvector), embeddings (OpenAI), document processing, batch processing, query filtering, authentication, and collection naming.

```bash
# Server
RAG_HOST=0.0.0.0
RAG_PORT=8000
DEBUG_RAG_API=False

# Vector Database (pgvector)
VECTOR_DB_TYPE=pgvector
DB_HOST=localhost
DB_PORT=5432
POSTGRES_DB=rag_db
POSTGRES_USER=rag_user
POSTGRES_PASSWORD=secure_password
PGVECTOR_CREATE_EXTENSION=True
PG_POOL_PRE_PING=True

# Embeddings (OpenAI)
EMBEDDINGS_PROVIDER=openai
RAG_OPENAI_API_KEY=sk-...
EMBEDDINGS_MODEL=text-embedding-3-small
EMBEDDINGS_DIMENSIONS=1536
EMBEDDINGS_CHUNK_SIZE=200

# Document Processing
CHUNK_SIZE=1500
CHUNK_OVERLAP=100
RAG_UPLOAD_DIR=./uploads/
PDF_EXTRACT_IMAGES=False

# Batch Processing
EMBEDDING_BATCH_SIZE=750
EMBEDDING_MAX_QUEUE_SIZE=3

# Query Filtering
RAG_DISTANCE_THRESHOLD=0.5

# Authentication
JWT_SECRET=your-secret-key-here

# Collection
COLLECTION_NAME=documents
```

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

### Expression Indexes Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Conceptual example of using expression indexes in a database for performance optimization. This speeds up queries involving computed values.

```sql
CREATE INDEX idx_document_length ON documents ((LENGTH(content)));
-- Or for PostgreSQL with pgvector:
-- CREATE INDEX idx_document_embedding ON documents USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);
```

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

### Rollback on Failure Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Conceptual example of implementing rollback logic in a batch processing pipeline. If a step fails, previous successful steps might need to be undone.

```python
class BatchProcessor:
    def __init__(self):
        self.processed_items = []

    def process_item(self, item):
        if item == "fail":
            raise ValueError("Processing failed for this item")
        # Simulate processing
        processed_data = f"processed_{item}"
        self.processed_items.append(processed_data)
        print(f"Processed: {item} -> {processed_data}")
        return processed_data

    def rollback(self, failed_item):
        print(f"Rolling back items processed before {failed_item}...")
        # In a real scenario, this would involve undoing database transactions, deleting partial results, etc.
        # For simplicity, we just clear the list here.
        self.processed_items = []

processor = BatchProcessor()
items_to_process = ["a", "b", "fail", "c"]

try:
    for item in items_to_process:
        processor.process_item(item)
except ValueError as e:
    print(e)
    processor.rollback(item)

print(f"Final processed items: {processor.processed_items}")
```

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

### JWT Authentication Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Conceptual example of JWT (JSON Web Token) authentication. This involves verifying a token to authenticate incoming requests.

```python
import jwt

SECRET_KEY = "your-super-secret-key"

def verify_jwt(token):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload
    except jwt.ExpiredSignatureError:
        print("Token expired")
        return None
    except jwt.InvalidTokenError:
        print("Invalid token")
        return None

# Example usage:
# token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
# user_info = verify_jwt(token)
# if user_info:
#     print("Authenticated user:", user_info.get('sub'))
```

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

### Document Format Detection Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Conceptual example of automatic document format detection. This utility would identify file types (e.g., PDF, TXT, DOCX) for appropriate processing.

```python
import magic # Example library for file type detection

def detect_document_type(file_path):
    mime = magic.from_file(file_path, mime=True)
    if mime == 'application/pdf':
        return 'pdf'
    elif mime == 'text/plain':
        return 'txt'
    # Add more types as needed
    return 'unknown'

# file_type = detect_document_type('my_document.pdf')
# print(f"Detected type: {file_type}")
```

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

### Debug Route: Get DB Tables

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example of using the /db/tables debug route. This route is available when DEBUG_RAG_API is set to True.

```python
import requests

url = "http://localhost:8000/db/tables"

response = requests.get(url)
print(response.json())
```

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

### Example Usage of POST /query

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Illustrates how to perform a single-file search using the POST /query endpoint. This is a primary route for retrieving information.

```python
import requests

url = "http://localhost:8000/query"

payload = {
    "query": "What is the capital of France?",
    "document_id": "doc123"
}

response = requests.post(url, json=payload)
print(response.json())
```

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

### Debug Route: Get All Records

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example of using the /records/all debug route. This route is available when DEBUG_RAG_API is set to True.

```python
import requests

url = "http://localhost:8000/records/all"

response = requests.get(url)
print(response.json())
```

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

### Vector Search Optimization Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Conceptual example of optimizing vector search, often involving specific database index types or algorithms for faster similarity searches.

```sql
-- Example for pgvector using IVFFlat index
-- CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);

-- Example query using the index
-- SELECT * FROM documents ORDER BY embedding <=> '[1,2,3]' LIMIT 5;
```

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

### Producer-Consumer Pattern Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Illustrates the producer-consumer pattern using queues for asynchronous data processing. This is key for batching pipelines.

```python
import asyncio
import random

async def producer(queue, n):
    for i in range(n):
        item = random.randint(1, 100)
        await queue.put(item)
        print(f"Produced: {item}")
        await asyncio.sleep(random.random())
    await queue.put(None) # Signal end

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"Consumed: {item}")
        await asyncio.sleep(random.random())
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    producer_task = asyncio.create_task(producer(queue, 10))
    consumer_task = asyncio.create_task(consumer(queue))
    await producer_task
    await consumer_task

asyncio.run(main())
```

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

### Debug Route: Get DB Table Columns

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example of using the /db/tables/columns debug route. This route is available when DEBUG_RAG_API is set to True.

```python
import requests

url = "http://localhost:8000/db/tables/columns"

response = requests.get(url)
print(response.json())
```

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

### Clean Install Local Dependencies

Source: https://github.com/danny-avila/rag_api/blob/main/README.md

Commands to perform a clean reinstall of project dependencies, including recreating the virtual environment.

```bash
# Remove existing virtual environment and recreate it
rm -rf venv
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
```

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

### Clean Install Lite Dependencies

Source: https://github.com/danny-avila/rag_api/blob/main/README.md

Commands for a clean reinstall of the lite version of dependencies, excluding sentence_transformers/huggingface.

```bash
rm -rf venv
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.lite.txt
```

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

### Example Usage of POST /embed

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Demonstrates how to upload and embed a file using the POST /embed endpoint. This is a primary route for ingesting documents.

```python
import requests

url = "http://localhost:8000/embed"
files = {"file": open("my_document.pdf", "rb")}

response = requests.post(url, files=files)
print(response.json())
```

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

### Debug Route: Get Records

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example of using the /records debug route. This route is available when DEBUG_RAG_API is set to True.

```python
import requests

url = "http://localhost:8000/records"

response = requests.get(url)
print(response.json())
```

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

### Example Usage of POST /local/embed

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Shows how to embed a file directly from the filesystem using the POST /local/embed endpoint. Requires DEBUG_RAG_API to be True.

```python
import requests

url = "http://localhost:8000/local/embed"
files = {"file": open("my_document.pdf", "rb")}

response = requests.post(url, files=files)
print(response.json())
```

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

### PSQLDatabase Utility

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example structure of the PSQLDatabase utility, likely responsible for managing PostgreSQL database connections and operations.

```python
import asyncpg

class PSQLDatabase:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.pool = None

    async def connect(self):
        self.pool = await asyncpg.create_pool(self.connection_string)

    async def disconnect(self):
        if self.pool:
            await self.pool.close()

    async def execute(self, query: str, *args):
        async with self.pool.acquire() as connection:
            return await connection.execute(query, *args)
```

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

### Memory-Bounded Queues Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Demonstrates creating memory-bounded queues to control resource usage in batch processing pipelines, preventing excessive memory consumption.

```python
import asyncio

async def main():
    # Create a queue with a maximum size of 5 items
    queue = asyncio.Queue(maxsize=5)

    # Attempting to put more items than maxsize will block
    for i in range(10):
        await queue.put(i)
        print(f"Put item {i}, queue size: {queue.qsize()}")

    print("Queue is full, next put will block.")
    # This next put would block if not for the loop finishing
    # await queue.put(10)

    while not queue.empty():
        item = await queue.get()
        print(f"Got item {item}, queue size: {queue.qsize()}")

asyncio.run(main())
```

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

### Example Usage of POST /query_multiple

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Demonstrates searching across multiple files using the POST /query_multiple endpoint. This is a primary route for multi-document retrieval.

```python
import requests

url = "http://localhost:8000/query_multiple"

payload = {
    "query": "Explain the concept of recursion.",
    "document_ids": ["doc123", "doc456"]
}

response = requests.post(url, json=payload)
print(response.json())
```

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

### Event Loop Integration Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Demonstrates integrating asynchronous operations with Python's event loop, essential for non-blocking I/O and concurrency.

```python
import asyncio

async def my_async_task():
    print("Starting async task...")
    await asyncio.sleep(1)
    print("Async task finished.")

async def main():
    await my_async_task()

if __name__ == "__main__":
    asyncio.run(main())
```

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

### Access Control Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Demonstrates access control logic, specifically filtering documents based on a user ID. This ensures users only see their own data.

```python
def filter_by_user(documents, user_id):
    return [doc for doc in documents if doc.get('user_id') == user_id]

all_documents = [
    {'id': 'doc1', 'content': '...', 'user_id': 'user1'},
    {'id': 'doc2', 'content': '...', 'user_id': 'user2'},
    {'id': 'doc3', 'content': '...', 'user_id': 'user1'}
]

user1_docs = filter_by_user(all_documents, 'user1')
print(user1_docs)
```

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

### Query Caching Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Illustrates the concept of query caching to improve performance by storing and reusing results of frequent or expensive queries.

```python
from functools import lru_cache

@lru_cache(maxsize=128)
def expensive_query(param):
    # Simulate a time-consuming database query
    print(f"Executing expensive query for: {param}")
    # Replace with actual database call
    return f"Result for {param}"

print(expensive_query("data1"))
print(expensive_query("data1")) # This call will be cached
print(expensive_query("data2"))
```

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

### Example Usage of DELETE /documents

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Shows how to delete documents by their IDs using the DELETE /documents endpoint. This is a primary route for document management.

```python
import requests

url = "http://localhost:8000/documents"
params = {"ids": "doc123,doc456"}

response = requests.delete(url, params=params)
print(response.json())
```

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

### POST /embed-upload Request Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/endpoints.md

An alternative endpoint for embedding file uploads, functioning identically to the /embed endpoint.

```json
{
  "status": boolean,
  "file_id": string,
  "filename": string,
  "known_type": boolean
}
```

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

### Path Validation Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Illustrates path validation to prevent directory traversal vulnerabilities. Ensures that file paths are safe and within expected directories.

```python
import os

BASE_DIR = "/app/data"

def is_safe_path(base_path, requested_path):
    abs_base = os.path.abspath(base_path)
    abs_requested = os.path.abspath(os.path.join(base_path, requested_path))
    return abs_requested.startswith(abs_base)

user_input_path = "../secrets/config.txt"

if is_safe_path(BASE_DIR, user_input_path):
    print(f"Accessing: {os.path.join(BASE_DIR, user_input_path)}")
else:
    print("Error: Path traversal detected!")
```

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

### Input Validation Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Shows basic input validation, ensuring that data received by the API meets expected criteria before processing.

```python
from pydantic import BaseModel, ValidationError

class Item(BaseModel):
    name: str
    price: float

input_data = {"name": "Gadget", "price": 19.99}

try:
    item = Item(**input_data)
    print(f"Validated item: {item.name}, Price: {item.price}")
except ValidationError as e:
    print(f"Validation error: {e}")

invalid_input = {"name": "Widget", "price": "expensive"}
try:
    item = Item(**invalid_input)
except ValidationError as e:
    print(f"Validation error for invalid input: {e}")
```

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

### Log HTTP Requests (Text Format)

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/utilities.md

Example of text-formatted log output from the LogMiddleware. This middleware logs requests with method, URL, and status code.

```text
2025-01-15 10:30:45 - root - INFO - Request POST http://localhost:8000/embed - 200
```

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

### Security Middleware Utility

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example of a security middleware utility, responsible for enforcing security policies such as authentication and authorization.

```python
from fastapi import Request, HTTPException

async def security_middleware(request: Request, call_next):
    # Implementation for security checks (e.g., JWT validation, access control)
    # if not is_authenticated(request):
    #     raise HTTPException(status_code=401, detail="Not authenticated")
    response = await call_next(request)
    return response
```

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

### Executor-Based Blocking I/O Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Shows how to perform blocking I/O operations within an asynchronous context using an executor, preventing the event loop from being blocked.

```python
import asyncio
import time

def blocking_io():
    time.sleep(2)
    return "Blocking I/O done"

async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, blocking_io)
    print(result)

asyncio.run(main())
```

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

### Thread Pool Model Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Illustrates the concept of a thread pool model for managing concurrent tasks. This is crucial for handling asynchronous operations efficiently.

```python
import concurrent.futures

def task(n):
    return n * n

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(task, i) for i in range(10)]
    results = [future.result() for future in concurrent.futures.as_completed(futures)]
    print(results)
```

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

### Encoding Conversion Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Illustrates handling different text encodings during document processing. This ensures text is read and stored correctly, regardless of its original encoding.

```python
def read_with_encoding(file_path, encoding='utf-8'):
    try:
        with open(file_path, 'r', encoding=encoding) as f:
            return f.read()
    except UnicodeDecodeError:
        # Try another common encoding if the first fails
        try:
            with open(file_path, 'r', encoding='latin-1') as f:
                return f.read()
        except Exception as e:
            print(f"Could not decode file: {e}")
            return None

# content = read_with_encoding('my_text_file.txt')
# if content:
#     print(content[:100])
```

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

### SafePyPDFLoader Utility

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example of the SafePyPDFLoader utility, likely used for robustly loading PDF files, handling potential errors during parsing.

```python
from typing import List, Dict, Any

class SafePyPDFLoader:
    def __init__(self, file_path: str):
        self.file_path = file_path

    def load(self) -> List[Dict[str, Any]]:
        # Implementation for loading and parsing PDF content safely
        return []
```

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

### POST /text Request Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/endpoints.md

Extract raw text from an uploaded file without generating embeddings. Useful for pre-processing validation.

```json
{
  "text": string,
  "file_id": string,
  "filename": string,
  "known_type": boolean
}
```

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

### Example Usage of POST /text

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Shows how to extract text from a document without embedding it, using the POST /text endpoint. Useful for pre-processing or analysis.

```python
import requests

url = "http://localhost:8000/text"
files = {"file": open("my_document.pdf", "rb")}

response = requests.post(url, files=files)
print(response.json())
```

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

### POST /embed Request Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/endpoints.md

Upload a file to generate embeddings for its content. Supports various file types and allows specifying an entity ID for access control.

```json
{
  "status": boolean,
  "message": string,
  "file_id": string,
  "filename": string,
  "known_type": boolean
}
```

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

### POST /local/embed Request Body Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/endpoints.md

Embed a file from the local filesystem by providing its path and metadata. Path validation prevents directory traversal.

```json
{
  "filepath": string,
  "filename": string,
  "file_content_type": string,
  "file_id": string
}
```

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

### LogMiddleware Utility

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example of a LogMiddleware utility, likely used for intercepting and logging requests or responses in the application's middleware stack.

```python
from fastapi import Request, Response

async def LogMiddleware(request: Request, call_next):
    # Implementation for logging request and response details
    response = await call_next(request)
    print(f"Request: {request.method} {request.url}")
    print(f"Response: {response.status_code}")
    return response
```

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

### Debug Route: Check Index

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example of using the /test/check_index debug route. This route is available when DEBUG_RAG_API is set to True.

```python
import requests

url = "http://localhost:8000/test/check_index"

response = requests.get(url)
print(response.json())
```

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

### AsyncPgVector Class

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example structure of the AsyncPgVector class, an asynchronous wrapper for pgvector operations. It includes methods for document management and querying.

```python
import asyncio
from typing import List, Dict, Any

class AsyncPgVector:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string

    async def connect(self):
        # Implementation for connecting to pgvector
        pass

    async def disconnect(self):
        # Implementation for disconnecting
        pass

    async def add_documents(self, documents: List[Dict[str, Any]]):
        # Implementation to add documents
        pass

    async def query(self, query_text: str, top_k: int = 3) -> List[Dict[str, Any]]:
        # Implementation for querying documents
        return []

    async def delete_documents(self, ids: List[str]):
        # Implementation for deleting documents
        pass
```

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

### Pydantic Model: CleanupMethod

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example definition of the CleanupMethod Pydantic model, likely used for specifying document cleanup strategies.

```python
from pydantic import BaseModel

class CleanupMethod(BaseModel):
    method: str
```

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

### Get or Create PostgreSQL Connection Pool

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/utilities.md

Retrieves the existing asyncpg connection pool or creates a new one if it doesn't exist. The pool is lazily created on the first call.

```python
@classmethod
async def get_pool(cls) -> asyncpg.pool.Pool
```

```python
pool = await PSQLDatabase.get_pool()
async with pool.acquire() as conn:
    result = await conn.fetchval("SELECT 1")
```

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

### Log HTTP Requests (JSON Format)

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/utilities.md

Example of JSON-formatted log output from the LogMiddleware. This format includes structured metadata like timestamp, level, request details, and response status code.

```json
{
  "message": "Request POST http://localhost:8000/embed - 200",
  "http_req": {"method": "POST", "url": "http://localhost:8000/embed"},
  "http_res": {"status_code": 200},
  "timestamp": "2025-01-15T10:30:45.123456",
  "level": "INFO"
}
```

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

### SafePyPDFLoader lazy_load Method Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/utilities.md

Demonstrates how to use the lazy_load method of SafePyPDFLoader to iterate over PDF documents. This method is useful for processing large PDFs efficiently by yielding documents one by one.

```python
loader = SafePyPDFLoader("document.pdf", extract_images=True)
for doc in loader.lazy_load():
    print(doc.page_content)
```

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

### JWT Authentication Header Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/endpoints.md

When JWT authentication is enabled, include a valid JWT token in the Authorization header. Tokens are verified using HS256.

```http
Authorization: Bearer <token>
```

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

### PDF Special Handling Example

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Highlights the need for special handling when processing PDF documents, which may contain complex structures, images, or scanned text requiring OCR.

```python
from PyPDF2 import PdfReader # Example library

def extract_pdf_text(file_path):
    text = ""
    try:
        reader = PdfReader(file_path)
        for page in reader.pages:
            text += page.extract_text() + "\n"
    except Exception as e:
        print(f"Error processing PDF: {e}")
        # Potentially fallback to OCR or other methods
    return text

# pdf_text = extract_pdf_text('complex_document.pdf')
# print(pdf_text[:200])
```

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

### Build PostgreSQL Search Path

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/vector-stores.md

Utility function to construct a PostgreSQL search_path, ensuring the 'public' schema is appended. This is important for resolving pgvector types and functions, even if they are installed in the public schema.

```python
def _build_search_path(schemas: List[str]) -> str
```

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

### GET /health

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/document-routes.md

Check the health status of the API and its backing vector database.

```APIDOC
## GET /health

### Description
Check the health of the API and backing vector database. Supports both pgvector and MongoDB.

### Method
GET

### Endpoint
/health

### Parameters
None

### Response
#### Success Response (200)
- **status** (str) - API status, either "UP" or "DOWN"

#### Response Example
```json
{"status": "UP"}
```
```

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

### Run RAG API Locally

Source: https://github.com/danny-avila/rag_api/blob/main/README.md

Use these commands to run the RAG API locally after setting up dependencies. Ensure you are in a virtual environment.

```bash
pip install -r requirements.txt
uvicorn main:app
```

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

### Get Full Document Context

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Retrieves the full context of a specific document.

```APIDOC
## GET /documents/{id}/context

### Description
Retrieves the complete content and context of a specific document.

### Method
GET

### Endpoint
/documents/{id}/context

### Parameters
#### Path Parameters
- **id** (string) - Required - The ID of the document to retrieve context for.
```

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

### GET /documents

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/endpoints.md

Retrieves documents based on a list of provided document IDs.

```APIDOC
## GET /documents

### Description
Retrieve documents by their IDs.

### Method
GET

### Endpoint
/documents

### Parameters
#### Query Parameters
- **ids** (list[string]) - Required - Document IDs to retrieve

### Response
#### Success Response (200)
- **page_content** (string) - The content of the document chunk.
- **metadata** (object) - Metadata associated with the document chunk.

#### Response Example
```json
[
  {
    "page_content": "Content of document 1.",
    "metadata": {}
  },
  {
    "page_content": "Content of document 2.",
    "metadata": {}
  }
]
```

### Status Codes
- `200`: Documents retrieved
- `404`: One or more IDs not found
- `500`: Retrieval error
```

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

### GET /ids

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/endpoints.md

Retrieves a list of all unique file IDs currently indexed in the vector store.

```APIDOC
## GET /ids

### Description
Get all unique file IDs in the vector store.

### Method
GET

### Endpoint
/ids

### Response
#### Success Response (200)
- **array[string]** - A list of all unique `file_id` values currently indexed.

#### Response Example
```json
[
  "file1",
  "file2",
  "file3"
]
```

### Status Codes
- `200`: IDs retrieved
- `500`: Retrieval error
```

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

### Enum: EmbeddingsProvider

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example definition of the EmbeddingsProvider enum, listing various supported embedding providers.

```python
from enum import Enum

class EmbeddingsProvider(str, Enum):
    OPENAI = "openai"
    COHERE = "cohere"
    HUGGINGFACE = "huggingface"
    AZURE_OPENAI = "azure_openai"
    BEDROCK = "bedrock"
    VERTEX_AI = "vertex_ai"
    LOCAL = "local"
    SENTENCE_TRANSFORMERS = "sentence-transformers"
```

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

### Get All IDs from Vector Store

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/vector-stores.md

Retrieves all unique file IDs currently stored in the vector store.

```python
def get_all_ids(self) -> list[str]:
    # ... implementation details ...
    pass
```

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

### GET /documents/{id}/context

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/endpoints.md

Loads the full context of a document identified by its ID, reconstructing it from all its chunks.

```APIDOC
## GET /documents/{id}/context

### Description
Load full document context for a single ID (all chunks from that file).

### Method
GET

### Endpoint
/documents/{id}/context

### Parameters
#### Path Parameters
- **id** (string) - Required - Document ID

### Response
#### Success Response (200)
- **string** - The full document text reconstructed from all chunks for that ID, with pages delineated.

#### Response Example
```
This is the full text of the document.
Page 1 content...
---
Page 2 content...
```

### Status Codes
- `200`: Document loaded
- `404`: Document not found
- `400`: Invalid request or loading error
```

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

### Enum: ERROR_MESSAGES

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example definition of the ERROR_MESSAGES enum, likely containing predefined error message strings.

```python
from enum import Enum

class ERROR_MESSAGES(str, Enum):
    FILE_NOT_FOUND = "File not found."
    INVALID_INPUT = "Invalid input provided."
```

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

### Pydantic Model: StoreDocument

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example definition of the StoreDocument Pydantic model, used for storing document information.

```python
from pydantic import BaseModel

class StoreDocument(BaseModel):
    id: str
    content: str
    metadata: dict
```

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

### Pydantic Model: DocumentModel

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example definition of the DocumentModel Pydantic model, representing a document within the system.

```python
from pydantic import BaseModel

class DocumentModel(BaseModel):
    id: str
    content: str
    metadata: dict
```

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

### Create Vector Store Instance

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/vector-stores.md

Use this function to create a vector store instance. Specify the connection string, embeddings, collection name, and desired mode (sync, async, or atlas-mongo). Optional parameters allow for search index configuration, extension creation, and connection pooling behavior.

```python
def get_vector_store(
    connection_string: str,
    embeddings: Embeddings,
    collection_name: str,
    mode: str = "sync",
    search_index: Optional[str] = None,
    create_extension: bool = True,
    pool_pre_ping: bool = True,
    pool_recycle: int = -1,
    schema: Optional[str] = None,
)
```

```python
vector_store = get_vector_store(
    connection_string="postgresql://user:pass@localhost/db",
    embeddings=embeddings,
    collection_name="documents",
    mode="async",
    pool_pre_ping=True,
    pool_recycle=1800
)
```

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

### GET /documents Response Body

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/endpoints.md

Represents the structure of retrieved documents. Each document includes its content and associated metadata.

```json
[
  {
    "page_content": string,
    "metadata": object
  }
]
```

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

### Minimal Environment Variables Configuration

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/README.md

Configure essential RAG API settings using environment variables for server port, vector database connection, embeddings model, and document processing parameters.

```bash
# Server
RAG_PORT=8000

# Vector Database
VECTOR_DB_TYPE=pgvector
DB_HOST=localhost
POSTGRES_DB=rag_db
POSTGRES_USER=rag_user
POSTGRES_PASSWORD=password

# Embeddings (OpenAI)
RAG_OPENAI_API_KEY=sk-...
EMBEDDINGS_MODEL=text-embedding-3-small

# Document Processing
CHUNK_SIZE=1500
CHUNK_OVERLAP=100
EMBEDDING_BATCH_SIZE=750
```

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

### AtlasMongoVector Class

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example structure of the AtlasMongoVector class, designed for interacting with MongoDB Atlas vector search capabilities.

```python
from typing import List, Dict, Any

class AtlasMongoVector:
    def __init__(self, connection_string: str, collection_name: str):
        self.connection_string = connection_string
        self.collection_name = collection_name

    def connect(self):
        # Implementation for connecting to MongoDB Atlas
        pass

    def add_documents(self, documents: List[Dict[str, Any]]):
        # Implementation to add documents to MongoDB Atlas
        pass

    def query(self, query_text: str, top_k: int = 3) -> List[Dict[str, Any]]:
        # Implementation for querying documents in MongoDB Atlas
        return []

    def delete_documents(self, ids: List[str]):
        # Implementation for deleting documents from MongoDB Atlas
        pass
```

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

### Pydantic Model: QueryMultipleBody

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example definition of the QueryMultipleBody Pydantic model, used for the request body of multi-document queries.

```python
from pydantic import BaseModel

class QueryMultipleBody(BaseModel):
    query: str
    document_ids: list[str]
```

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

### Pydantic Model: QueryRequestBody

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example definition of the QueryRequestBody Pydantic model, used for the request body of single-document queries.

```python
from pydantic import BaseModel

class QueryRequestBody(BaseModel):
    query: str
    document_id: str
```

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

### Pydantic Model: DocumentResponse

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example definition of the DocumentResponse Pydantic model, used for API responses related to documents.

```python
from pydantic import BaseModel

class DocumentResponse(BaseModel):
    id: str
    content: str
    metadata: dict
```

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

### Health Check using cURL

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/README.md

Verify the operational status of the RAG API by sending a GET request to the /health endpoint.

```bash
curl http://localhost:8000/health
```

```json
{"status": "UP"}
```

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

### Synchronous Document Preparation

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/document-routes.md

Synchronously prepares raw documents by splitting, cleaning, and adding metadata. Runs within an executor for blocking I/O.

```python
def _prepare_documents_sync(
    data: Iterable[Document],
    file_id: str,
    user_id: str,
    clean_content: bool
) → List[Document]
```

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

### Module Map

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/INDEX.md

This map outlines the project's directory structure and the corresponding documentation files for each module.

```text
app/config.py              → [configuration.md](configuration.md)
app/models.py              → [types.md](types.md)
app/constants.py           → [types.md](types.md)
app/middleware.py          → [api-reference/utilities.md](api-reference/utilities.md)

app/routes/document_routes.py    → [api-reference/document-routes.md](api-reference/document-routes.md)
app/routes/pgvector_routes.py    → [api-reference/endpoints.md](api-reference/endpoints.md)

app/services/database.py                 → [api-reference/utilities.md](api-reference/utilities.md)
app/services/mongo_client.py             → [api-reference/utilities.md](api-reference/utilities.md)
app/services/vector_store/factory.py     → [api-reference/vector-stores.md](api-reference/vector-stores.md)
app/services/vector_store/async_pg_vector.py  → [api-reference/vector-stores.md](api-reference/vector-stores.md)
app/services/vector_store/extended_pg_vector.py → [api-reference/vector-stores.md](api-reference/vector-stores.md)
app/services/vector_store/atlas_mongo_vector.py → [api-reference/vector-stores.md](api-reference/vector-stores.md)

app/utils/document_loader.py  → [api-reference/utilities.md](api-reference/utilities.md)
app/utils/health.py          → [api-reference/utilities.md](api-reference/utilities.md)

main.py                  → [architecture.md](architecture.md)
```

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

### ExtendedPgVector Class

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example structure of the ExtendedPgVector class, an optimized version of the pgvector wrapper. It focuses on efficient pgvector operations.

```python
from typing import List, Dict, Any

class ExtendedPgVector:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string

    def connect(self):
        # Implementation for connecting to pgvector
        pass

    def add_documents(self, documents: List[Dict[str, Any]]):
        # Implementation to add documents
        pass

    def query(self, query_text: str, top_k: int = 3) -> List[Dict[str, Any]]:
        # Implementation for querying documents
        return []

    def delete_documents(self, ids: List[str]):
        # Implementation for deleting documents
        pass
```

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

### Initialize FastAPI Router

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/document-routes.md

Initializes a FastAPI router instance. This router will be used to register various API endpoints.

```python
router = APIRouter()
```

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

### Initialize Embeddings Provider

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/architecture.md

Selects and initializes an embeddings provider based on environment configuration. Ensure the EMBEDDINGS_PROVIDER environment variable is set to a supported provider.

```python
EMBEDDINGS_PROVIDER = EmbeddingsProvider(get_env_variable("EMBEDDINGS_PROVIDER", "openai"))
embeddings = init_embeddings(EMBEDDINGS_PROVIDER, EMBEDDINGS_MODEL, EMBEDDINGS_DIMENSIONS)
```

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

### Enable pgvector Extension in PostgreSQL

Source: https://github.com/danny-avila/rag_api/blob/main/README.md

After creating the database and role, switch to the rag_api database and enable the pgvector extension. This is crucial for RAG API functionality. Note that master users may not have superuser privileges required for certain role creations.

```sql
\c rag_api
```

```sql
create extension vector;
```

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

### Enum: VectorDBType

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/DOCUMENT_MAP.txt

Example definition of the VectorDBType enum, specifying available vector database types like pgvector and MongoDB Atlas.

```python
from enum import Enum

class VectorDBType(str, Enum):
    PGVECTOR = "pgvector"
    ATLAS_MONGO = "atlas-mongo"
```

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

### RAG API Project Structure

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/architecture.md

Illustrates the directory and file organization of the RAG API project, highlighting the purpose of key modules and files.

```tree
rag_api/
├── main.py                          # FastAPI app initialization and lifespan
├── requirements.txt                 # Python dependencies
├── README.md                        # User documentation
│
├── app/
│   ├── __init__.py
│   ├── config.py                   # Environment configuration and embeddings initialization
│   ├── constants.py                # Error message enums
│   ├── middleware.py               # JWT authentication middleware
│   ├── models.py                   # Pydantic request/response models
│   │
│   ├── routes/
│   │   ├── __init__.py
│   │   ├── document_routes.py      # Main endpoints: /embed, /query, /documents
│   │   └── pgvector_routes.py      # Debug endpoints (DEBUG_RAG_API only)
│   │
│   ├── services/
│   │   ├── __init__.py
│   │   ├── database.py             # PostgreSQL connection pool and indexes
│   │   ├── mongo_client.py         # MongoDB health check
│   │   │
│   │   └── vector_store/
│   │       ├── __init__.py
│   │       ├── factory.py          # Vector store creation and connection management
│   │       ├── async_pg_vector.py  # Async pgvector wrapper
│   │       ├── extended_pg_vector.py  # Optimized pgvector with expression indexes
│   │       └── atlas_mongo_vector.py  # MongoDB Atlas vector search
│   │
│   └── utils/
│       ├── __init__.py
│       ├── document_loader.py      # File format detection and document loaders
│       └── health.py               # Health check dispatcher
│
└── tests/                           # Test suite
```

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

### Create Database and Role in PostgreSQL

Source: https://github.com/danny-avila/rag_api/blob/main/README.md

Use these SQL commands to create a dedicated database and role for the RAG API within your RDS Postgres instance. Ensure you are connected to the correct instance before execution.

```sql
create database rag_api;
```

```sql
create role rag;
```

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

### PSQLDatabase Class Definition

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/utilities.md

Defines a singleton class for managing PostgreSQL connection pooling and operations using asyncpg.

```python
class PSQLDatabase:
    pool = None

    @classmethod
    async def get_pool(cls):
        """Get or create asyncpg connection pool."""

    @classmethod
    async def close_pool(cls):
        """Close the connection pool."""
```

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

### Get All Vector Store IDs

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/vector-stores.md

Retrieves all unique file IDs from the vector store. Use when you need a complete list of all stored document identifiers.

```python
async def get_all_ids(self, executor=None) -> list[str]:
    # Get all unique file IDs in the vector store.
    pass

ids = await vector_store.get_all_ids(executor=request.app.state.thread_pool)
```

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

### Store Documents in Vector DB

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/document-routes.md

Prepares and stores documents in the vector database. Handles batching and rollback on failure. Supports optional text cleaning.

```python
async def store_data_in_vector_db(
    data: Iterable[Document],
    file_id: str,
    user_id: str = "",
    clean_content: bool = False,
    executor=None
) → dict
```

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

### Get Document Context

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/README.md

Obtain the complete content of a document, including page markers, using its ID. This provides detailed context for a specific document.

```APIDOC
## GET /documents/{id}/context

### Description
Get full document context with page markers.

### Method
GET

### Endpoint
/documents/{id}/context

### Parameters
#### Path Parameters
- **id** (string) - Required - The ID of the document.
```

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

### Upload and Embed a File

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/document-routes.md

Use this endpoint to upload a file and generate embeddings for its content. Ensure the file is provided as multipart/form-data. The response includes processing status and file details.

```python
files = {"file": ("document.pdf", open("document.pdf", "rb"))}
data = {"file_id": "doc_123", "entity_id": "user_1"}
response = await client.post("/embed", files=files, data=data)
result = response.json()
```

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

### GET /ids Response Body

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/endpoints.md

Represents the response body for retrieving all unique file IDs. It returns a list of strings, where each string is a file ID.

```json
[
string
]
```

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

### SQL Query Optimization for Metadata Filtering

Source: https://github.com/danny-avila/rag_api/blob/main/_autodocs/api-reference/vector-stores.md

Demonstrates how to rewrite SQL queries to leverage B-tree expression indexes for faster metadata filtering, improving performance from O(N) to O(log N).

```sql
SELECT * FROM langchain_pg_embedding
WHERE langchain_pg_embedding_json_path_match(cmetadata, '$.file_id', 'file_123')
-- Requires sequential scan: O(N)
```

```sql
SELECT * FROM langchain_pg_embedding
WHERE (cmetadata->>'file_id') = 'file_123'
-- Uses index: idx_langchain_pg_embedding_file_id, O(log N)
```