API-Based Ingestion

Connect to external APIs for automated data ingestion.

REST API Ingestion

import requests

def ingest_from_api(api_url, auth_token):
    headers = {'Authorization': f'Bearer {auth_token}'}
    response = requests.get(api_url, headers=headers)
    
    for item in response.json()['data']:
        process_and_index(item)

Slack Integration

from slack_sdk import WebClient

def ingest_slack_messages(channel_id):
    client = WebClient(token=os.environ['SLACK_TOKEN'])
    
    result = client.conversations_history(channel=channel_id)
    
    for message in result['messages']:
        if 'text' in message:
            index_message({
                'content': message['text'],
                'user': message['user'],
                'timestamp': message['ts'],
                'source': 'slack'
            })

Google Drive

from googleapiclient.discovery import build

def ingest_google_drive(folder_id):
    service = build('drive', 'v3')
    
    results = service.files().list(
        q=f"'{folder_id}' in parents",
        fields="files(id, name, mimeType, modifiedTime)"
    ).execute()
    
    for file in results.get('files', []):
        content = download_file(service, file['id'])
        process_and_index(content)

Next: Incremental updates.