List Sources

The list method returns every source in the project. Each item includes file metadata (ID, name, size, type, origin), current processing status, and a human-readable message. You can optionally filter by file_ids.

Method overview

Python
TypeScript

client.sources.list()

await client.sources.list()

Method signature

Python
TypeScript

client.sources.list(
    file_ids: list[str] | None = None,   # Optional: filter by these file IDs
    timeout: float | None = None
) -> list[PublicSource]

await client.sources.list({
  fileIds?: string[] | null,   // Optional: filter by these file IDs
}): Promise<PublicSource[]>

Parameters

Python
TypeScript

Parameter	Type	Description	Required
`file_ids`	`list[str] \| None`	If provided, only sources whose `file_id` is in this list are returned	No
`timeout`	`float`	Request timeout in seconds	No

Parameter	Type	Description	Required
`fileIds`	`string[] \| null`	If provided, only sources with these `file_id`s are returned	No

When file_ids is omitted, the method returns all sources in the project.

Response

The method returns a list of source objects. Each has:

Property	Type	Description
`status`	`str`	Current processing status (capitalized: e.g. `New`, `Processing`, `Processed`, `Completed`, `Failed`)
`message`	`str`	Human-readable status message
`file_id`	`str`	Unique identifier for the source (use for reprocess, get elements, delete, ask, extract)
`file_name`	`str`	Display name of the source file or identifier
`file_size`	`int`	Size in bytes (0 for URL/GitHub/YouTube)
`file_type`	`str`	File extension or type
`file_source`	`str`	Origin: `local file`, `url`, `github`, or `youtube`
`project_id`	`str`	UUID of the project
`project_name`	`str`	Name of the project
`partition_method`	`str \| null`	Partition method: `auto`, `fast`, `balanced`, `accurate`, or `agentic` (when available)

Status values

Backend returns status with capital letters. Typical values:

Status	Description
`New`	Upload/ingestion accepted; processing has not started yet
`Processing`	Pipeline is running
`Processed`	Source has been processed (intermediate or final state)
`Completed`	Ready for ask, extract, retrieve, and get elements
`Failed` / `Processing failed`	Processing encountered an error; consider reprocessing
`unknown`	Status could not be determined

File Source Types

Source Type	Description	Typical Use Cases
`local file`	Files uploaded directly from your computer	Documents, PDFs, images, spreadsheets
`url`	Content imported from web URLs	Web pages, articles, online documents
`github`	Content imported from GitHub repositories	Code documentation, README files, wikis
`youtube`	Content imported from YouTube videos	Video transcripts, educational content

Code Examples

Basic Usage

Python
TypeScript

from graphor import Graphor

client = Graphor()

# List all sources in the project
sources = client.sources.list()

print(f"Found {len(sources)} sources")

for source in sources:
    print(f"{source.file_name} - {source.status}")

import Graphor from 'graphor';

const client = new Graphor();

// List all sources in the project
const sources = await client.sources.list();

console.log(`Found ${sources.length} sources`);

for (const source of sources) {
  console.log(`${source.file_name} - ${source.status}`);
}

Async Usage

Python
TypeScript

import asyncio
from graphor import AsyncGraphor

async def list_all_sources():
    client = AsyncGraphor()
    
    sources = await client.sources.list()
    
    print(f"Found {len(sources)} sources")
    
    for source in sources:
        print(f"{source.file_name} - {source.status}")
    
    return sources

asyncio.run(list_all_sources())

import Graphor from 'graphor';

const client = new Graphor();

async function listAllSources() {
  const sources = await client.sources.list();

  console.log(`Found ${sources.length} sources`);

  for (const source of sources) {
    console.log(`${source.file_name} - ${source.status}`);
  }

  return sources;
}

await listAllSources();

Filter by file_ids

Python
TypeScript

from graphor import Graphor

client = Graphor()
file_ids = ["file_abc123", "file_def456"]
sources = client.sources.list(file_ids=file_ids)
print(f"Found {len(sources)} sources")

const client = new Graphor();
const fileIds = ['file_abc123', 'file_def456'];
const sources = await client.sources.list({ fileIds });
console.log(`Found ${sources.length} sources`);

Filter by status

Python
TypeScript

from graphor import Graphor

client = Graphor()

sources = client.sources.list()

# Filter by status
completed = [s for s in sources if s.status == "Completed"]
processing = [s for s in sources if s.status == "Processing"]
failed = [s for s in sources if s.status == "Failed"]
new = [s for s in sources if s.status == "New"]

print(f"Completed: {len(completed)}")
print(f"Processing: {len(processing)}")
print(f"Failed: {len(failed)}")
print(f"New: {len(new)}")

import Graphor from 'graphor';

const client = new Graphor();

const sources = await client.sources.list();

// Filter by status
const completed = sources.filter((s) => s.status === 'Completed');
const processing = sources.filter((s) => s.status === 'Processing');
const failed = sources.filter((s) => s.status === 'Failed');
const newSources = sources.filter((s) => s.status === 'New');

console.log(`Completed: ${completed.length}`);
console.log(`Processing: ${processing.length}`);
console.log(`Failed: ${failed.length}`);
console.log(`New: ${newSources.length}`);

Filter by File Type

Python
TypeScript

from graphor import Graphor

client = Graphor()

sources = client.sources.list()

# Filter by file type
pdf_files = [s for s in sources if s.file_type == "pdf"]
docx_files = [s for s in sources if s.file_type == "docx"]
images = [s for s in sources if s.file_type in ("png", "jpg", "jpeg")]

print(f"PDFs: {len(pdf_files)}")
print(f"Word docs: {len(docx_files)}")
print(f"Images: {len(images)}")

# List all PDF files
for pdf in pdf_files:
    size_mb = pdf.file_size / (1024 * 1024)
    print(f"  {pdf.file_name} ({size_mb:.2f} MB)")

import Graphor from 'graphor';

const client = new Graphor();

const sources = await client.sources.list();

// Filter by file type
const pdfFiles = sources.filter((s) => s.file_type === 'pdf');
const docxFiles = sources.filter((s) => s.file_type === 'docx');
const images = sources.filter((s) => ['png', 'jpg', 'jpeg'].includes(s.file_type));

console.log(`PDFs: ${pdfFiles.length}`);
console.log(`Word docs: ${docxFiles.length}`);
console.log(`Images: ${images.length}`);

// List all PDF files
for (const pdf of pdfFiles) {
  const sizeMb = pdf.file_size / (1024 * 1024);
  console.log(`  ${pdf.file_name} (${sizeMb.toFixed(2)} MB)`);
}

Filter by Source Type

Python
TypeScript

from graphor import Graphor

client = Graphor()

sources = client.sources.list()

# Filter by source type
local_files = [s for s in sources if s.file_source == "local file"]
url_sources = [s for s in sources if s.file_source == "url"]
github_sources = [s for s in sources if s.file_source == "github"]
youtube_sources = [s for s in sources if s.file_source == "youtube"]

print(f"Local files: {len(local_files)}")
print(f"URL sources: {len(url_sources)}")
print(f"GitHub repos: {len(github_sources)}")
print(f"YouTube videos: {len(youtube_sources)}")

import Graphor from 'graphor';

const client = new Graphor();

const sources = await client.sources.list();

// Filter by source type
const localFiles = sources.filter((s) => s.file_source === 'local file');
const urlSources = sources.filter((s) => s.file_source === 'url');
const githubSources = sources.filter((s) => s.file_source === 'github');
const youtubeSources = sources.filter((s) => s.file_source === 'youtube');

console.log(`Local files: ${localFiles.length}`);
console.log(`URL sources: ${urlSources.length}`);
console.log(`GitHub repos: ${githubSources.length}`);
console.log(`YouTube videos: ${youtubeSources.length}`);

Error Handling

Python
TypeScript

import graphor
from graphor import Graphor

client = Graphor()

try:
    sources = client.sources.list()
    print(f"Found {len(sources)} sources")
    
except graphor.AuthenticationError as e:
    print(f"Invalid API key: {e}")
    
except graphor.PermissionDeniedError as e:
    print(f"Access denied to project: {e}")
    
except graphor.RateLimitError as e:
    print(f"Rate limit exceeded. Please wait and retry: {e}")
    
except graphor.APIConnectionError as e:
    print(f"Connection error: {e}")
    
except graphor.InternalServerError as e:
    print(f"Server error: {e}")

import Graphor from 'graphor';

const client = new Graphor();

try {
  const sources = await client.sources.list();
  console.log(`Found ${sources.length} sources`);
} catch (err) {
  if (err instanceof Graphor.AuthenticationError) {
    console.log(`Invalid API key: ${err.message}`);
  } else if (err instanceof Graphor.PermissionDeniedError) {
    console.log(`Access denied to project: ${err.message}`);
  } else if (err instanceof Graphor.RateLimitError) {
    console.log(`Rate limit exceeded. Please wait and retry: ${err.message}`);
  } else if (err instanceof Graphor.APIConnectionError) {
    console.log(`Connection error: ${err.message}`);
  } else if (err instanceof Graphor.InternalServerError) {
    console.log(`Server error: ${err.message}`);
  } else {
    throw err;
  }
}

Advanced Examples

Source Analysis

Analyze your project’s sources with detailed statistics:

Python
TypeScript

from graphor import Graphor
from collections import defaultdict

client = Graphor()

def analyze_sources():
    """Analyze sources by status, type, and size."""
    sources = client.sources.list()
    
    status_counts = defaultdict(int)
    type_counts = defaultdict(int)
    source_counts = defaultdict(int)
    total_size = 0
    
    for source in sources:
        status_counts[source.status] += 1
        type_counts[source.file_type] += 1
        source_counts[source.file_source] += 1
        total_size += source.file_size
    
    return {
        "total_sources": len(sources),
        "total_size_mb": round(total_size / (1024 * 1024), 2),
        "by_status": dict(status_counts),
        "by_type": dict(type_counts),
        "by_source": dict(source_counts)
    }

# Usage
analysis = analyze_sources()
print(f"Total sources: {analysis['total_sources']}")
print(f"Total size: {analysis['total_size_mb']} MB")
print(f"By status: {analysis['by_status']}")
print(f"By type: {analysis['by_type']}")
print(f"By source: {analysis['by_source']}")

import Graphor from 'graphor';

const client = new Graphor();

async function analyzeSources() {
  const sources = await client.sources.list();

  const statusCounts: Record<string, number> = {};
  const typeCounts: Record<string, number> = {};
  const sourceCounts: Record<string, number> = {};
  let totalSize = 0;

  for (const source of sources) {
    statusCounts[source.status] = (statusCounts[source.status] ?? 0) + 1;
    typeCounts[source.file_type] = (typeCounts[source.file_type] ?? 0) + 1;
    sourceCounts[source.file_source] = (sourceCounts[source.file_source] ?? 0) + 1;
    totalSize += source.file_size;
  }

  return {
    totalSources: sources.length,
    totalSizeMb: Math.round((totalSize / (1024 * 1024)) * 100) / 100,
    byStatus: statusCounts,
    byType: typeCounts,
    bySource: sourceCounts,
  };
}

// Usage
const analysis = await analyzeSources();
console.log(`Total sources: ${analysis.totalSources}`);
console.log(`Total size: ${analysis.totalSizeMb} MB`);
console.log('By status:', analysis.byStatus);
console.log('By type:', analysis.byType);
console.log('By source:', analysis.bySource);

Status Monitoring

Monitor the processing status of your documents:

Python
TypeScript

from graphor import Graphor

client = Graphor()

def monitor_processing_status():
    """Monitor and report on processing status."""
    sources = client.sources.list()
    
    processing = [s for s in sources if s.status == "Processing"]
    failed = [s for s in sources if s.status == "Failed"]
    completed = [s for s in sources if s.status == "Completed"]
    new = [s for s in sources if s.status == "New"]
    
    print("=" * 50)
    print("Processing Status Report")
    print("=" * 50)
    print(f"Completed: {len(completed)}")
    print(f"Processing: {len(processing)}")
    print(f"New: {len(new)}")
    print(f"Failed: {len(failed)}")
    print("=" * 50)
    
    # List files currently processing
    if processing:
        print("\nCurrently Processing:")
        for source in processing:
            print(f"  - {source.file_name} ({source.partition_method})")
    
    # List failed files that need attention
    if failed:
        print("\nFailed Files (need attention):")
        for source in failed:
            print(f"  - {source.file_name}: {source.message}")
    
    return {
        "completed": completed,
        "processing": processing,
        "new": new,
        "failed": failed
    }

# Usage
status = monitor_processing_status()

import Graphor from 'graphor';

const client = new Graphor();

async function monitorProcessingStatus() {
  const sources = await client.sources.list();

  const processing = sources.filter((s) => s.status === 'Processing');
  const failed = sources.filter((s) => s.status === 'Failed');
  const completed = sources.filter((s) => s.status === 'Completed');
  const newSources = sources.filter((s) => s.status === 'New');

  console.log('='.repeat(50));
  console.log('Processing Status Report');
  console.log('='.repeat(50));
  console.log(`Completed: ${completed.length}`);
  console.log(`Processing: ${processing.length}`);
  console.log(`New: ${newSources.length}`);
  console.log(`Failed: ${failed.length}`);
  console.log('='.repeat(50));

  // List files currently processing
  if (processing.length > 0) {
    console.log('\nCurrently Processing:');
    for (const source of processing) {
      console.log(`  - ${source.file_name} (${source.partition_method})`);
    }
  }

  // List failed files that need attention
  if (failed.length > 0) {
    console.log('\nFailed Files (need attention):');
    for (const source of failed) {
      console.log(`  - ${source.file_name}: ${source.message}`);
    }
  }

  return { completed, processing, new: newSources, failed };
}

// Usage
const status = await monitorProcessingStatus();

Find Source by Name

Search for a specific source by filename:

Python
TypeScript

from graphor import Graphor

client = Graphor()

def find_source(file_name: str):
    """Find a source by exact file name."""
    sources = client.sources.list()
    
    for source in sources:
        if source.file_name == file_name:
            return source
    
    return None

def search_sources(query: str):
    """Search sources by partial name match."""
    sources = client.sources.list()
    
    matches = [s for s in sources if query.lower() in s.file_name.lower()]
    return matches

# Usage
# Find exact match
source = find_source("document.pdf")
if source:
    print(f"Found: {source.file_name} - {source.status}")
else:
    print("Source not found")

# Search by partial name
matches = search_sources("report")
print(f"Found {len(matches)} sources matching 'report'")
for match in matches:
    print(f"  - {match.file_name}")

import Graphor from 'graphor';

const client = new Graphor();

async function findSource(fileName: string) {
  const sources = await client.sources.list();
  return sources.find((s) => s.file_name === fileName) ?? null;
}

async function searchSources(query: string) {
  const sources = await client.sources.list();
  return sources.filter((s) => s.file_name.toLowerCase().includes(query.toLowerCase()));
}

// Usage
// Find exact match
const source = await findSource('document.pdf');
if (source) {
  console.log(`Found: ${source.file_name} - ${source.status}`);
} else {
  console.log('Source not found');
}

// Search by partial name
const matches = await searchSources('report');
console.log(`Found ${matches.length} sources matching 'report'`);
for (const match of matches) {
  console.log(`  - ${match.file_name}`);
}

Project Health Check

Perform a comprehensive health check of your project:

Python
TypeScript

from graphor import Graphor
from datetime import datetime
import graphor

client = Graphor()

def project_health_check():
    """Perform a comprehensive health check of the project."""
    try:
        sources = client.sources.list()
        
        health_report = {
            "timestamp": datetime.now().isoformat(),
            "total_sources": len(sources),
            "status_summary": {},
            "issues": [],
            "recommendations": []
        }
        
        # Analyze status distribution
        for source in sources:
            status = source.status or "unknown"
            health_report["status_summary"][status] = health_report["status_summary"].get(status, 0) + 1
            
            # Identify issues
            if source.status == "Failed":
                health_report["issues"].append(f"Failed processing: {source.file_name}")
            elif source.status == "unknown":
                health_report["issues"].append(f"Unknown status: {source.file_name}")
        
        # Generate recommendations
        failed_count = health_report["status_summary"].get("Failed", 0)
        if failed_count > 0:
            health_report["recommendations"].append(
                f"Reprocess {failed_count} failed documents using client.sources.reprocess(file_id)"
            )
        
        processing_count = health_report["status_summary"].get("Processing", 0)
        if processing_count > 5:
            health_report["recommendations"].append(
                "Monitor processing queue - high volume detected"
            )
        
        new_count = health_report["status_summary"].get("New", 0)
        if new_count > 0:
            health_report["recommendations"].append(
                f"{new_count} documents awaiting processing"
            )
        
        return health_report
        
    except graphor.APIStatusError as e:
        return {
            "error": str(e),
            "timestamp": datetime.now().isoformat()
        }

# Usage
health = project_health_check()
print(f"Project Health Report")
print(f"Timestamp: {health['timestamp']}")
print(f"Total Sources: {health['total_sources']}")
print(f"Status Summary: {health['status_summary']}")

if health.get("issues"):
    print(f"\nIssues:")
    for issue in health["issues"]:
        print(f"  - {issue}")

if health.get("recommendations"):
    print(f"\nRecommendations:")
    for rec in health["recommendations"]:
        print(f"  - {rec}")

import Graphor from 'graphor';

const client = new Graphor();

async function projectHealthCheck() {
  try {
    const sources = await client.sources.list();

    const statusSummary: Record<string, number> = {};
    const issues: string[] = [];
    const recommendations: string[] = [];

    // Analyze status distribution
    for (const source of sources) {
      const status = source.status || 'unknown';
      statusSummary[status] = (statusSummary[status] ?? 0) + 1;

      if (source.status === 'Failed') {
        issues.push(`Failed processing: ${source.file_name}`);
      } else if (source.status === 'unknown') {
        issues.push(`Unknown status: ${source.file_name}`);
      }
    }

    // Generate recommendations
    const failedCount = statusSummary['Failed'] ?? 0;
    if (failedCount > 0) {
      recommendations.push(
        `Reprocess ${failedCount} failed documents using client.sources.reprocess(file_id)`,
      );
    }

    const processingCount = statusSummary['Processing'] ?? 0;
    if (processingCount > 5) {
      recommendations.push('Monitor processing queue - high volume detected');
    }

    const newCount = statusSummary['New'] ?? 0;
    if (newCount > 0) {
      recommendations.push(`${newCount} documents awaiting processing`);
    }

    return {
      timestamp: new Date().toISOString(),
      totalSources: sources.length,
      statusSummary,
      issues,
      recommendations,
    };
  } catch (err) {
    return {
      error: err instanceof Graphor.APIError ? err.message : String(err),
      timestamp: new Date().toISOString(),
    };
  }
}

// Usage
const health = await projectHealthCheck();
console.log('Project Health Report');
console.log(`Timestamp: ${health.timestamp}`);
if ('totalSources' in health) {
  console.log(`Total Sources: ${health.totalSources}`);
  console.log('Status Summary:', health.statusSummary);

  if (health.issues?.length) {
    console.log('\nIssues:');
    for (const issue of health.issues) {
      console.log(`  - ${issue}`);
    }
  }

  if (health.recommendations?.length) {
    console.log('\nRecommendations:');
    for (const rec of health.recommendations) {
      console.log(`  - ${rec}`);
    }
  }
}

Async Batch Operations

Use the list to perform batch operations efficiently:

Python
TypeScript

import asyncio
from graphor import AsyncGraphor
import graphor

async def reprocess_failed_sources(method: str = "balanced"):
    """Find and reprocess all failed sources."""
    client = AsyncGraphor(timeout=300.0)
    
    # Get all sources
    sources = await client.sources.list()
    
    # Find failed sources
    failed = [s for s in sources if s.status == "Failed"]
    
    if not failed:
        print("No failed sources to reprocess")
        return []
    
    print(f"Found {len(failed)} failed sources to reprocess")
    
    # Reprocess each failed source (returns build_id; poll get_build_status to wait)
    results = []
    for source in failed:
        try:
            print(f"Reprocessing: {source.file_name} ({source.file_id})...")
            build_id = await client.sources.reprocess(
                file_id=source.file_id,
                partition_method=method
            )
            results.append({"file_id": source.file_id, "build_id": build_id, "status": "scheduled"})
            print(f"  OK - build_id: {build_id}")
        except graphor.APIStatusError as e:
            results.append({"file_id": source.file_id, "status": "failed", "error": str(e)})
            print(f"  FAIL - {source.file_id}: {e}")
    
    return results

# Usage
results = asyncio.run(reprocess_failed_sources("balanced"))

import Graphor from 'graphor';

const client = new Graphor({ timeout: 300 * 1000 });

async function reprocessFailedSources(method = 'balanced') {
  // Get all sources
  const sources = await client.sources.list();

  // Find failed sources
  const failed = sources.filter((s) => s.status === 'Failed');

  if (failed.length === 0) {
    console.log('No failed sources to reprocess');
    return [];
  }

  console.log(`Found ${failed.length} failed sources to reprocess`);

  // Reprocess each failed source (returns build_id; poll getBuildStatus to wait)
  const results: { fileId: string; buildId?: string; status: string; error?: string }[] = [];
  for (const source of failed) {
    try {
      console.log(`Reprocessing: ${source.file_name} (${source.file_id})...`);
      const buildId = await client.sources.reprocess({
        fileId: source.file_id,
        partitionMethod: method as any,
      });
      results.push({ fileId: source.file_id, buildId, status: 'scheduled' });
      console.log(`  OK - build_id: ${buildId}`);
    } catch (err) {
      const message = err instanceof Graphor.APIError ? err.message : String(err);
      results.push({ fileId: source.file_id, status: 'failed', error: message });
      console.log(`  FAIL - ${source.file_id}: ${message}`);
    }
  }

  return results;
}

// Usage
const results = await reprocessFailedSources('balanced');

Source Management Class

A complete class for managing sources:

Python
TypeScript

from graphor import Graphor
import graphor
from dataclasses import dataclass
from typing import Optional

@dataclass
class SourceSummary:
    total: int
    completed: int
    processing: int
    failed: int
    new: int
    total_size_mb: float

class SourceManager:
    def __init__(self, api_key: Optional[str] = None):
        self.client = Graphor(api_key=api_key) if api_key else Graphor()
        self._cache = None
    
    def refresh(self):
        """Refresh the sources cache."""
        self._cache = self.client.sources.list()
        return self._cache
    
    @property
    def sources(self):
        """Get sources (cached)."""
        if self._cache is None:
            self.refresh()
        return self._cache
    
    def get_summary(self) -> SourceSummary:
        """Get a summary of all sources."""
        sources = self.sources
        
        total_size = sum(s.file_size for s in sources)
        
        return SourceSummary(
            total=len(sources),
            completed=len([s for s in sources if s.status == "Completed"]),
            processing=len([s for s in sources if s.status == "Processing"]),
            failed=len([s for s in sources if s.status == "Failed"]),
            new=len([s for s in sources if s.status == "New"]),
            total_size_mb=round(total_size / (1024 * 1024), 2)
        )
    
    def find_by_name(self, name: str):
        """Find a source by exact name."""
        for source in self.sources:
            if source.file_name == name:
                return source
        return None
    
    def search(self, query: str):
        """Search sources by partial name match."""
        return [s for s in self.sources if query.lower() in s.file_name.lower()]
    
    def filter_by_status(self, status: str):
        """Filter sources by status."""
        return [s for s in self.sources if s.status == status]
    
    def filter_by_type(self, file_type: str):
        """Filter sources by file type."""
        return [s for s in self.sources if s.file_type == file_type]
    
    def get_failed(self):
        """Get all failed sources."""
        return self.filter_by_status("Failed")
    
    def get_processing(self):
        """Get all processing sources."""
        return self.filter_by_status("Processing")

# Usage
manager = SourceManager()

# Get summary
summary = manager.get_summary()
print(f"Total: {summary.total}, Completed: {summary.completed}, Failed: {summary.failed}")

# Find a specific source
source = manager.find_by_name("document.pdf")
if source:
    print(f"Found: {source.file_name} - {source.status}")

# Search sources
matches = manager.search("report")
print(f"Found {len(matches)} matches for 'report'")

# Get failed sources
failed = manager.get_failed()
print(f"Failed sources: {len(failed)}")

import Graphor from 'graphor';

interface SourceSummary {
  total: number;
  completed: number;
  processing: number;
  failed: number;
  new: number;
  totalSizeMb: number;
}

type PublicSource = Awaited<ReturnType<Graphor['sources']['list']>>[number];

class SourceManager {
  private client: Graphor;
  private _cache: PublicSource[] | null = null;

  constructor(apiKey?: string) {
    this.client = apiKey ? new Graphor({ apiKey }) : new Graphor();
  }

  async refresh() {
    this._cache = await this.client.sources.list();
    return this._cache;
  }

  async getSources() {
    if (this._cache === null) {
      await this.refresh();
    }
    return this._cache!;
  }

  async getSummary(): Promise<SourceSummary> {
    const sources = await this.getSources();

    const totalSize = sources.reduce((sum, s) => sum + s.file_size, 0);

    return {
      total: sources.length,
      completed: sources.filter((s) => s.status === 'Completed').length,
      processing: sources.filter((s) => s.status === 'Processing').length,
      failed: sources.filter((s) => s.status === 'Failed').length,
      new: sources.filter((s) => s.status === 'New').length,
      totalSizeMb: Math.round((totalSize / (1024 * 1024)) * 100) / 100,
    };
  }

  async findByName(name: string) {
    const sources = await this.getSources();
    return sources.find((s) => s.file_name === name) ?? null;
  }

  async search(query: string) {
    const sources = await this.getSources();
    return sources.filter((s) => s.file_name.toLowerCase().includes(query.toLowerCase()));
  }

  async filterByStatus(status: string) {
    const sources = await this.getSources();
    return sources.filter((s) => s.status === status);
  }

  async filterByType(fileType: string) {
    const sources = await this.getSources();
    return sources.filter((s) => s.file_type === fileType);
  }

  async getFailed() {
    return this.filterByStatus('Failed');
  }

  async getProcessing() {
    return this.filterByStatus('Processing');
  }
}

// Usage
const manager = new SourceManager();

// Get summary
const summary = await manager.getSummary();
console.log(`Total: ${summary.total}, Completed: ${summary.completed}, Failed: ${summary.failed}`);

// Find a specific source
const source = await manager.findByName('document.pdf');
if (source) {
  console.log(`Found: ${source.file_name} - ${source.status}`);
}

// Search sources
const matches = await manager.search('report');
console.log(`Found ${matches.length} matches for 'report'`);

// Get failed sources
const failed = await manager.getFailed();
console.log(`Failed sources: ${failed.length}`);

Continuous Monitoring

Set up continuous monitoring of your sources:

Python
TypeScript

import time
from graphor import Graphor
import graphor

client = Graphor()

def continuous_monitoring(interval_seconds: int = 60, max_iterations: int = None):
    """Continuously monitor source processing status."""
    iteration = 0
    
    while max_iterations is None or iteration < max_iterations:
        try:
            sources = client.sources.list()
            
            processing = len([s for s in sources if s.status == "Processing"])
            failed = len([s for s in sources if s.status == "Failed"])
            completed = len([s for s in sources if s.status == "Completed"])
            
            print(f"[{time.strftime('%H:%M:%S')}] "
                  f"Completed: {completed} | Processing: {processing} | Failed: {failed}")
            
            # Alert if new failures detected
            if failed > 0:
                failed_sources = [s for s in sources if s.status == "Failed"]
                print(f"  Warning - Failed sources: {[s.file_name for s in failed_sources]}")
            
            time.sleep(interval_seconds)
            iteration += 1
            
        except graphor.APIConnectionError as e:
            print(f"[{time.strftime('%H:%M:%S')}] Connection error: {e}")
            time.sleep(interval_seconds)
        except KeyboardInterrupt:
            print("\nMonitoring stopped")
            break

# Usage (monitor every 30 seconds, 10 times)
# continuous_monitoring(interval_seconds=30, max_iterations=10)

import Graphor from 'graphor';

const client = new Graphor();

async function continuousMonitoring(intervalSeconds = 60, maxIterations?: number) {
  let iteration = 0;

  while (maxIterations === undefined || iteration < maxIterations) {
    try {
      const sources = await client.sources.list();

      const processing = sources.filter((s) => s.status === 'Processing').length;
      const failed = sources.filter((s) => s.status === 'Failed').length;
      const completed = sources.filter((s) => s.status === 'Completed').length;

      const time = new Date().toLocaleTimeString();
      console.log(
        `[${time}] Completed: ${completed} | Processing: ${processing} | Failed: ${failed}`,
      );

      // Alert if new failures detected
      if (failed > 0) {
        const failedSources = sources.filter((s) => s.status === 'Failed');
        console.log(
          `  Warning - Failed sources: ${failedSources.map((s) => s.file_name).join(', ')}`,
        );
      }

      await new Promise((r) => setTimeout(r, intervalSeconds * 1000));
      iteration++;
    } catch (err) {
      if (err instanceof Graphor.APIConnectionError) {
        const time = new Date().toLocaleTimeString();
        console.log(`[${time}] Connection error: ${err.message}`);
        await new Promise((r) => setTimeout(r, intervalSeconds * 1000));
      } else {
        throw err;
      }
    }
  }
}

// Usage (monitor every 30 seconds, 10 times)
// await continuousMonitoring(30, 10);

Error Reference

Error Type	Status Code	Description
`AuthenticationError`	401	Invalid or missing API key
`PermissionDeniedError`	403	Access denied to the specified project
`RateLimitError`	429	Too many requests, please retry after waiting
`InternalServerError`	≥500	Server-side error retrieving sources
`APIConnectionError`	N/A	Network connectivity issues
`APITimeoutError`	N/A	Request timed out

Best Practices

Performance Optimization

Cache results: Store the response locally when making multiple queries
Filter client-side: The SDK returns all sources; filter in your code as needed
Use async: For applications that need to perform other work while waiting

Python
TypeScript

# Example: Cache sources for multiple operations
sources = client.sources.list()

# Now perform multiple filter operations without re-fetching
pdfs = [s for s in sources if s.file_type == "pdf"]
completed = [s for s in sources if s.status == "Completed"]
large_files = [s for s in sources if s.file_size > 10 * 1024 * 1024]

// Example: Cache sources for multiple operations
const sources = await client.sources.list();

// Now perform multiple filter operations without re-fetching
const pdfs = sources.filter((s) => s.file_type === 'pdf');
const completed = sources.filter((s) => s.status === 'Completed');
const largeFiles = sources.filter((s) => s.file_size > 10 * 1024 * 1024);

Data Management

Track processing times: Monitor how long documents take to process
Identify patterns: Look for file types or sizes that frequently fail
Plan capacity: Use file counts and sizes for storage planning

Error Handling

Implement retries: Handle temporary network issues with the SDK’s built-in retry mechanism
Monitor status: Regularly check for failed processing jobs
Graceful degradation: Have fallback plans when the API is unavailable

Python
TypeScript

from graphor import Graphor

# Configure retries
client = Graphor(max_retries=5)

# Or per-request
sources = client.with_options(max_retries=5).sources.list()

import Graphor from 'graphor';

// Configure retries
const client = new Graphor({ maxRetries: 5 });

// Or per-request
const sources = await client.sources.list({}, { maxRetries: 5 });

Troubleshooting

Slow response times

Causes: Large number of sources, server load, or network issuesSolutions:

Implement request timeouts
Use response caching for non-critical applications
Consider filtering client-side after initial fetch

Python
TypeScript

client = Graphor(timeout=60.0)  # Increase timeout

const client = new Graphor({ timeout: 60 * 1000 }); // Increase timeout

Empty response

Causes: No sources in project, wrong API key, or permission issuesSolutions:

Verify you have uploaded documents to your project
Check that your API key is correct and active
Ensure you’re accessing the right project

Inconsistent status information

Causes: Processing lag, system sync issues, or database inconsistenciesSolutions:

Wait a few minutes and retry the request
Call sources.list() again to refresh the data
Contact support if inconsistencies persist

Authentication errors

Causes: Invalid token, expired token, or revoked accessSolutions:

Verify API key format and validity
Check token hasn’t been revoked in dashboard
Generate a new API key if necessary

Next steps

After listing your sources:

Upload

Ingest new files, URLs, GitHub repos, or YouTube videos

Reprocess source

Reprocess a source with a different partition method

Get elements

Retrieve parsed elements from a source

Delete source

Remove a source by file_id

Get Started

Data SDK Options

Method overview

Method signature

Parameters

Response

Status values

File Source Types

Code Examples

Basic Usage

Async Usage

Filter by file_ids

Filter by status

Filter by File Type

Filter by Source Type

Error Handling

Advanced Examples

Source Analysis

Status Monitoring

Find Source by Name

Project Health Check

Async Batch Operations

Source Management Class

Continuous Monitoring

Error Reference

Best Practices

Performance Optimization

Data Management

Error Handling

Troubleshooting

Next steps

Upload

Reprocess source

Get elements

Delete source

​Method overview

​Method signature

​Parameters

​Response

​Status values

​File Source Types

​Code Examples

​Basic Usage

​Async Usage

​Filter by file_ids

​Filter by status

​Filter by File Type

​Filter by Source Type

​Error Handling

​Advanced Examples

​Source Analysis

​Status Monitoring

​Find Source by Name

​Project Health Check

​Async Batch Operations

​Source Management Class

​Continuous Monitoring

​Error Reference

​Best Practices

​Performance Optimization

​Data Management

​Error Handling

​Troubleshooting

​Next steps

Upload

Reprocess source

Get elements

Delete source

Method overview

Method signature

Parameters

Response

Status values

File Source Types

Code Examples

Basic Usage

Async Usage

Filter by file_ids

Filter by status

Filter by File Type

Filter by Source Type

Error Handling

Advanced Examples

Source Analysis

Status Monitoring

Find Source by Name

Project Health Check

Async Batch Operations

Source Management Class

Continuous Monitoring

Error Reference

Best Practices

Performance Optimization

Data Management

Error Handling

Troubleshooting

Next steps