Delete Source

The Delete Source endpoint allows you to permanently remove documents from your Graphor project using the REST API. This endpoint provides a secure way to clean up your document collection, automatically updating related flows and removing all associated data including partition nodes and metadata.

Endpoint Overview

HTTP Method

DELETE

Endpoint URL

https://sources.graphorlm.com/delete

Authentication

This endpoint requires authentication using an API token. You must include your API token as a Bearer token in the Authorization header.

Learn how to create and manage API tokens in the API Tokens guide.

Request Format

Headers

Header	Value	Required
`Authorization`	`Bearer YOUR_API_TOKEN`	✅ Yes
`Content-Type`	`application/json`	✅ Yes

Request Body

The request must be sent as JSON with the following field:

Field	Type	Description	Required
`file_id`	string	Unique identifier of the source to delete	Yes

You can obtain file_id from List sources or from Get build status after an ingestion completes.

Important Considerations

Permanent Deletion

Warning: This operation is irreversible

All document content and metadata will be permanently removed
Associated partition nodes and embeddings will be deleted
Flows using this document will be automatically updated
No backup or recovery options are available

Using file_id

Identifier required

Use the source’s unique file_id (returned by list sources or build status)
Use List sources to see all sources and their file_ids

Flow Impact

Automatic flow updates

Dataset nodes using this document will be updated automatically
Successor nodes in affected flows will be marked as outdated
Flow execution may be impacted until nodes are reconfigured
Multiple flows can be affected by a single deletion

Request Example

{
  "file_id": "file_abc123"
}

Response Format

Success Response (200 OK)

{
  "status": "success",
  "message": "Source deleted successfully",
  "file_id": "file_abc123",
  "file_name": "document.pdf",
  "project_id": "550e8400-e29b-41d4-a716-446655440000",
  "project_name": "My Project"
}

Response Fields

Field	Type	Description
`status`	string	Deletion result (typically “success”)
`message`	string	Human-readable confirmation message
`file_id`	string	Unique identifier for the deleted source
`file_name`	string	Name of the deleted file
`project_id`	string	UUID of the project the file was removed from
`project_name`	string	Name of the project

Code Examples

JavaScript/Node.js

const deleteSource = async (apiToken, fileId) => {
  const response = await fetch('https://sources.graphorlm.com/delete', {
    method: 'DELETE',
    headers: {
      'Authorization': `Bearer ${apiToken}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({ file_id: fileId })
  });

  if (response.ok) {
    const result = await response.json();
    console.log('Deletion successful:', result);
    return result;
  } else {
    const error = await response.text();
    throw new Error(`Deletion failed: ${response.status} ${error}`);
  }
};

// Usage
deleteSource('grlm_your_api_token_here', 'file_abc123')
  .then(result => console.log('Source deleted:', result.file_id))
  .catch(error => console.error('Error:', error));

Python

import requests

def delete_source(api_token, file_id):
    url = "https://sources.graphorlm.com/delete"
    headers = {
        "Authorization": f"Bearer {api_token}",
        "Content-Type": "application/json"
    }
    response = requests.delete(url, headers=headers, json={"file_id": file_id}, timeout=60)
    response.raise_for_status()
    return response.json()

# Usage - with confirmation prompt for safety
def safe_delete_source(api_token, file_id):
    """Delete source with confirmation prompt for safety."""
    print(f"⚠️  WARNING: This will permanently delete source '{file_id}'")
    confirmation = input("Type 'DELETE' to confirm: ")
    if confirmation == 'DELETE':
        try:
            result = delete_source(api_token, file_id)
            print(f"✅ Source deleted successfully: {result.get('file_name', file_id)}")
            return result
        except requests.exceptions.RequestException as e:
            print(f"❌ Error deleting source: {e}")
            raise
    else:
        print("❌ Deletion cancelled")
        return None

# Usage
result = delete_source("grlm_your_api_token_here", "file_abc123")

cURL

curl -X DELETE https://sources.graphorlm.com/delete \
  -H "Authorization: Bearer grlm_your_api_token_here" \
  -H "Content-Type: application/json" \
  -d '{"file_id":"file_abc123"}'

Error Responses

Common Error Codes

Status Code	Error Type	Description
`400`	Bad Request	Invalid request format or missing file_id
`401`	Unauthorized	Invalid or missing API token
`403`	Forbidden	Access denied to the specified project
`404`	Not Found	File not found in the project
`500`	Internal Server Error	Server-side error during deletion

Error Response Format

{
  "detail": "Source not found"
}

Error Examples

File Not Found (404)

{
  "detail": "Source not found"
}

Cause: The specified file doesn’t exist in your projectSolutions:

Verify the exact file_id (obtain it from List sources or Get build status)
Use the List Sources endpoint to check available sources
Ensure the source wasn’t already deleted

Invalid API Token (401)

{
  "detail": "Invalid authentication credentials"
}

Cause: API token is invalid, expired, or malformedSolutions:

Check your API token format (should start with “grlm_”)
Verify the token hasn’t been revoked in the dashboard
Ensure correct Authorization header format

Access Denied (403)

{
  "detail": "Permission denied"
}

Cause: Insufficient permissions or API token doesn’t have accessSolutions:

Verify you’re using the correct API token for this project
Check that you have delete permissions
Contact your project administrator

Invalid Request (400)

{
  "detail": "Invalid input: file_id is required"
}

Cause: Missing file_id in request bodySolutions:

Include file_id in the JSON body (required)
Get file_id from List sources or Get build status

Best Practices

Pre-Deletion Verification

Always verify before deletion

def verify_and_delete(api_token, file_id):
    # 1. List available sources first
    sources = list_sources(api_token)
    by_id = {s["file_id"]: s for s in sources}
    if file_id not in by_id:
        print(f"❌ Source '{file_id}' not found in project")
        return None
    file_info = by_id[file_id]
    print(f"Source to delete: {file_info['file_name']} ({file_id})")
    print(f"Size: {file_info['file_size']} bytes, Type: {file_info['file_type']}, Status: {file_info['status']}")
    confirm = input("Proceed with deletion? (yes/no): ")
    if confirm.lower() == 'yes':
        return delete_source(api_token, file_id)
    print("Deletion cancelled")
    return None

Check for active processing

Before deleting documents that are currently processing:

Wait for processing to complete when possible
Processing documents may be in an inconsistent state
Consider the impact on any dependent workflows
Monitor processing status using the List Sources endpoint

Review flow dependencies

Before deletion, consider:

Which flows use this document
Impact on downstream processing
Whether alternative documents can serve the same purpose
Need to reconfigure affected flows after deletion

Safety Measures

Implement confirmation prompts in interactive applications
Log all deletion operations for audit trails
Use descriptive variable names to avoid accidental deletions
Test with non-production data when implementing deletion features
Consider soft deletion patterns for critical applications

Error Handling

Implement retry logic for transient network errors (not for 404/403 errors)
Validate file_id before making deletion requests
Handle batch operations carefully to avoid partial failures
Provide clear error messages to end users

Integration Examples

Batch Deletion with Safety Checks

import requests
import time
from typing import List, Dict

def safe_batch_delete(api_token: str, file_ids: List[str], confirm_each: bool = True) -> Dict:
    """Safely delete multiple sources by file_id."""
    print("🔍 Verifying sources exist before deletion...")
    try:
        r = requests.get(
            "https://sources.graphorlm.com",
            headers={"Authorization": f"Bearer {api_token}"},
            timeout=30
        )
        sources = r.json() if r.status_code == 200 else []
        by_id = {s["file_id"]: s for s in sources}
    except Exception as e:
        print(f"❌ Error fetching source list: {e}")
        return {"error": "Could not verify sources"}
    missing = [fid for fid in file_ids if fid not in by_id]
    if missing:
        print(f"❌ Sources not found: {missing}")
        return {"error": f"Missing: {missing}"}
    print(f"✅ All {len(file_ids)} sources found")
    total_size = sum(by_id[f]["file_size"] for f in file_ids)
    print(f"\n📋 Deletion summary: {len(file_ids)} sources, {total_size:,} bytes total")
    for fid in file_ids:
        info = by_id[fid]
        print(f"  - {info['file_name']} ({fid})")
    if confirm_each:
        confirm = input(f"\n⚠️  Delete {len(file_ids)} sources permanently? (type 'DELETE'): ")
        if confirm != 'DELETE':
            print("❌ Batch deletion cancelled")
            return {"cancelled": True}
    results = {"successful": [], "failed": []}
    for i, file_id in enumerate(file_ids, 1):
        print(f"\n[{i}/{len(file_ids)}] Deleting {file_id}...")
        try:
            response = requests.delete(
                "https://sources.graphorlm.com/delete",
                headers={"Authorization": f"Bearer {api_token}", "Content-Type": "application/json"},
                json={"file_id": file_id},
                timeout=60
            )
            if response.status_code == 200:
                results["successful"].append({"file_id": file_id, "result": response.json()})
                print("  ✅ Deleted successfully")
            else:
                results["failed"].append({"file_id": file_id, "error": response.text})
                print(f"  ❌ Failed: {response.text}")
        except Exception as e:
            results["failed"].append({"file_id": file_id, "error": str(e)})
            print(f"  ❌ Error: {e}")
        if i < len(file_ids):
            time.sleep(0.5)
    print(f"\n🏁 Complete: {len(results['successful'])} successful, {len(results['failed'])} failed")
    return results

# Usage
results = safe_batch_delete("grlm_your_token", ["file_id_1", "file_id_2"])

Project Cleanup Tool

class ProjectCleanupTool {
  constructor(apiToken) {
    this.apiToken = apiToken;
    this.baseUrl = 'https://sources.graphorlm.com';
  }

  async getSourcesByStatus(status = null) {
    const response = await fetch(this.baseUrl, {
      headers: { 'Authorization': `Bearer ${this.apiToken}` }
    });
    if (!response.ok) throw new Error(`Failed to fetch sources: ${response.status}`);
    const sources = await response.json();
    return status ? sources.filter(s => s.status === status) : sources;
  }

  async deleteSource(fileId) {
    const response = await fetch(`${this.baseUrl}/delete`, {
      method: 'DELETE',
      headers: {
        'Authorization': `Bearer ${this.apiToken}`,
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({ file_id: fileId })
    });
    if (!response.ok) {
      const error = await response.text();
      throw new Error(`Deletion failed: ${response.status} ${error}`);
    }
    return response.json();
  }

  async cleanupFailedSources(interactive = true) {
    console.log('🔍 Finding failed sources...');
    const failedSources = await this.getSourcesByStatus('Failed');
    if (failedSources.length === 0) {
      console.log('✅ No failed sources found');
      return { deleted: 0, skipped: 0 };
    }
    console.log(`📋 Found ${failedSources.length} failed sources:`);
    failedSources.forEach((source, i) => {
      console.log(`  ${i + 1}. ${source.file_name} (${source.file_id})`);
    });
    if (interactive) {
      const readline = require('readline').createInterface({ input: process.stdin, output: process.stdout });
      const answer = await new Promise(resolve => {
        readline.question(`\n⚠️  Delete all ${failedSources.length} failed sources? (yes/no): `, resolve);
      });
      readline.close();
      if (answer.toLowerCase() !== 'yes') {
        console.log('❌ Cleanup cancelled');
        return { deleted: 0, skipped: failedSources.length };
      }
    }
    let deleted = 0;
    for (const source of failedSources) {
      try {
        await this.deleteSource(source.file_id);
        console.log(`✅ Deleted: ${source.file_name}`);
        deleted++;
        await new Promise(resolve => setTimeout(resolve, 500));
      } catch (error) {
        console.log(`❌ Failed to delete ${source.file_name}: ${error.message}`);
      }
    }
    console.log(`\n🏁 Cleanup complete: ${deleted} sources deleted`);
    return { deleted, skipped: failedSources.length - deleted };
  }

  async cleanupOldSources(daysOld = 30, fileTypes = []) {
    console.log(`🔍 Finding sources older than ${daysOld} days...`);
    
    const allSources = await this.getSourcesByStatus();
    const cutoffDate = new Date();
    cutoffDate.setDate(cutoffDate.getDate() - daysOld);

    const oldSources = allSources.filter(source => {
      const createdAt = new Date(source.created_at || 0);
      const matchesAge = createdAt < cutoffDate;
      const matchesType = fileTypes.length === 0 || fileTypes.includes(source.file_type);
      return matchesAge && matchesType;
    });

    if (oldSources.length === 0) {
      console.log(`✅ No sources older than ${daysOld} days found`);
      return { deleted: 0 };
    }

    console.log(`📋 Found ${oldSources.length} old sources`);
    // Implementation continues...
  }
}

// Usage
const cleanup = new ProjectCleanupTool('grlm_your_token');

// Clean up failed sources
cleanup.cleanupFailedSources()
  .then(result => console.log('Cleanup result:', result))
  .catch(error => console.error('Cleanup error:', error));

File Lifecycle Management

from datetime import datetime, timedelta
from typing import List, Dict, Optional
import requests

class DocumentLifecycleManager:
    def __init__(self, api_token: str):
        self.api_token = api_token
        self.base_url = "https://sources.graphorlm.com"
    
    def get_headers(self):
        return {
            "Authorization": f"Bearer {self.api_token}",
            "Content-Type": "application/json"
        }
    
    def list_sources(self) -> List[Dict]:
        response = requests.get(self.base_url, headers=self.get_headers())
        response.raise_for_status()
        return response.json()
    
    def delete_source(self, file_id: str) -> Dict:
        response = requests.delete(
            f"{self.base_url}/delete",
            headers=self.get_headers(),
            json={"file_id": file_id}
        )
        response.raise_for_status()
        return response.json()
    
    def archive_by_criteria(self, 
                          max_age_days: Optional[int] = None,
                          file_types: Optional[List[str]] = None,
                          min_file_size: Optional[int] = None,
                          max_file_size: Optional[int] = None,
                          status_filter: Optional[str] = None,
                          dry_run: bool = True) -> Dict:
        """
        Archive (delete) sources based on multiple criteria.
        """
        
        print("🔍 Analyzing sources for archival...")
        sources = self.list_sources()
        
        candidates = []
        
        for source in sources:
            # Age check
            if max_age_days:
                # Note: This would need actual creation date from source
                # For demo purposes, we'll skip this check
                pass
            
            # File type check
            if file_types and source['file_type'] not in file_types:
                continue
            
            # Size checks
            size = source['file_size']
            if min_file_size and size < min_file_size:
                continue
            if max_file_size and size > max_file_size:
                continue
            
            # Status check
            if status_filter and source['status'] != status_filter:
                continue
            
            candidates.append(source)
        
        print(f"📋 Found {len(candidates)} sources matching criteria:")
        
        total_size = 0
        for source in candidates:
            size = source['file_size']
            total_size += size
            size_mb = size / (1024 * 1024)
            print(f"  - {source['file_name']} ({source['file_id']}, {size_mb:.1f}MB, {source['status']})")
        
        total_size_mb = total_size / (1024 * 1024)
        print(f"\nTotal size to be freed: {total_size_mb:.1f}MB")
        
        if dry_run:
            print("\n🔬 DRY RUN - No files will be deleted")
            return {
                "dry_run": True,
                "candidates": len(candidates),
                "total_size_mb": total_size_mb
            }
        
        # Confirm deletion
        confirm = input(f"\n⚠️  Delete {len(candidates)} sources? (type 'DELETE'): ")
        if confirm != 'DELETE':
            print("❌ Archival cancelled")
            return {"cancelled": True}
        
        # Perform deletions
        deleted = []
        failed = []
        
        for source in candidates:
            try:
                result = self.delete_source(source['file_id'])
                deleted.append(source['file_id'])
                print(f"✅ Archived: {source['file_name']}")
            except Exception as e:
                failed.append({"file_id": source['file_id'], "error": str(e)})
                print(f"❌ Failed: {source['file_name']} - {e}")
        return {
            "deleted": deleted,
            "failed": failed,
            "total_deleted": len(deleted),
            "total_failed": len(failed),
            "size_freed_mb": sum(s['file_size'] for s in candidates if s['file_id'] in deleted) / (1024 * 1024)
        }

# Usage examples
manager = DocumentLifecycleManager("grlm_your_token")

# Archive large files over 50MB
result = manager.archive_by_criteria(
    min_file_size=50 * 1024 * 1024,  # 50MB
    dry_run=True  # Check what would be deleted first
)

# Archive failed documents
result = manager.archive_by_criteria(
    status_filter="Failed",
    dry_run=False
)

# Archive specific file types
result = manager.archive_by_criteria(
    file_types=["tmp", "temp", "test"],
    dry_run=True
)

Troubleshooting

Source not found errors

Causes: Source doesn’t exist, wrong file_id, or already deletedSolutions:

Use List sources to get valid file_ids
Verify you’re using the correct project/API token
The source may have been deleted by another process

Authentication failures

Causes: Invalid token, token revoked, or wrong project accessSolutions:

Verify API token format (should start with “grlm_”)
Check token status in the Graphor dashboard
Ensure token has delete permissions for the project
Try regenerating the API token if needed

Deletion timeouts

Causes: Large files, complex cleanup operations, or server loadSolutions:

Increase request timeout (60+ seconds recommended)
Retry the operation after a short delay
Contact support for persistent timeout issues
Consider deleting files during off-peak hours

Flow synchronization issues

Causes: Flows updating asynchronously after deletionSolutions:

Allow time for flow updates to propagate
Refresh flow data in your application
Reconfigure affected flows manually if needed
Monitor flow status after bulk deletions

Network connectivity

Causes: DNS issues, firewall restrictions, or network timeoutsSolutions:

Test connectivity to sources.graphorlm.com
Check firewall allows outbound HTTPS traffic
Verify DNS resolution is working
Try from a different network if issues persist

Next Steps

After successfully deleting your documents:

List sources

Verify the deletion and view remaining sources (with their file_ids)

Upload & ingest

Ingest new files, URLs, GitHub, or YouTube (async)

Reprocess source

Re-process remaining sources with a different partition method (async)

Get build status

Poll status for async ingestion or reprocess

Get Started

Data API Options

RAG Pipelines API

Endpoint Overview

HTTP Method

Endpoint URL

Authentication

Request Format

Headers

Request Body

Important Considerations

Request Example

Response Format

Success Response (200 OK)

Response Fields

Code Examples

JavaScript/Node.js

Python

cURL

Error Responses

Common Error Codes

Error Response Format

Error Examples

Best Practices

Pre-Deletion Verification

Safety Measures

Error Handling

Integration Examples

Batch Deletion with Safety Checks

Project Cleanup Tool

File Lifecycle Management

Troubleshooting

Next Steps

List sources

Upload & ingest

Reprocess source

Get build status

Get Started

Data API Options

RAG Pipelines API

​Endpoint Overview

HTTP Method

Endpoint URL

​Authentication

​Request Format

​Headers

​Request Body

​Important Considerations

​Request Example

​Response Format

​Success Response (200 OK)

​Response Fields

​Code Examples

​JavaScript/Node.js

​Python

​cURL

​Error Responses

​Common Error Codes

​Error Response Format

​Error Examples

​Best Practices

​Pre-Deletion Verification

​Safety Measures

​Error Handling

​Integration Examples

​Batch Deletion with Safety Checks

​Project Cleanup Tool

​File Lifecycle Management

​Troubleshooting

​Next Steps

List sources

Upload & ingest

Reprocess source

Get build status

Endpoint Overview

Authentication

Request Format

Headers

Request Body

Important Considerations

Request Example

Response Format

Success Response (200 OK)

Response Fields

Code Examples

JavaScript/Node.js

Python

cURL

Error Responses

Common Error Codes

Error Response Format

Error Examples

Best Practices

Pre-Deletion Verification

Safety Measures

Error Handling

Integration Examples

Batch Deletion with Safety Checks

Project Cleanup Tool

File Lifecycle Management

Troubleshooting

Next Steps