How to Parse and Process Holehe's JSON Output Data in Python Applications

Holehe outputs one JSON object per line containing service name, registration status, and profile URLs, making it trivial to stream-process results in Python using json.loads on each line.

Holehe is an open-source reconnaissance tool that verifies whether an email address exists across hundreds of online services. When invoked from the command line, it streams results as newline-delimited JSON (NDJSON)—a format optimized for real-time parsing without memory overhead. This article shows you how to consume Holehe's output programmatically in Python, with complete code examples grounded in the actual source code at megadose/holehe.

---

Understanding Holehe's Output Format

Each line emitted by Holehe is a self-contained JSON object with four predictable fields:

Field	Type	Description
service	string	Internal identifier (e.g., github, twitter, instagram)
status	string	"registered" or "unregistered"
url	string | null	Direct profile link when status is "registered"
data	object	Optional service-specific metadata (user IDs, tokens, etc.)

Example output line:

{"service":"github","status":"registered","url":"https://github.com/example_user","data":{}}

The data field varies by service implementation in holehe/modules/—some modules populate it with extra context, others leave it empty.

---

Core Parsing Strategy: Line-by-Line Stream Processing

Holehe's main orchestration logic resides in [holehe/core.py](https://github.com/megadose/holehe/blob/master/holehe/core.py), where results are serialized and flushed to stdout immediately after each service check. This design lets Python consumers process results incrementally rather than buffering everything.

Basic Synchronous Parser

import json
import subprocess
from typing import Iterable, Dict, Any

def stream_holehe(email: str) -> Iterable[Dict[str, Any]]:
    """Execute Holehe and yield parsed JSON objects as they arrive."""
    proc = subprocess.Popen(
        ["holehe", "-f", email],
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        text=True,
        bufsize=1,  # Line-buffered

    )
    
    for line in proc.stdout:
        line = line.strip()
        if not line:
            continue
            
        try:
            result = json.loads(line)
            yield result
        except json.JSONDecodeError:
            # Skip malformed lines (rare, but defensive)

            continue
            
    proc.wait()
    if proc.returncode != 0:
        raise RuntimeError(f"Holehe exited with code {proc.returncode}")

# Usage

for entry in stream_holehe("target@example.com"):
    print(f"{entry['service']}: {entry['status']}")

Key implementation details:

• subprocess.Popen with stdout=subprocess.PIPE captures the stream without blocking
• text=True eliminates manual decode steps
• json.loads per line scales to thousands of services without memory bloat

---

Filtering and Extracting Registered Accounts

Most security workflows only care about services where the email is registered. Here's a focused extractor:

def extract_profiles(email: str) -> Dict[str, str]:
    """Map service names to profile URLs for registered accounts only."""
    profiles = {}
    
    for result in stream_holehe(email):
        if result.get("status") == "registered":
            service = result["service"]
            url = result.get("url") or "N/A"
            profiles[service] = url
            
    return profiles

# Example

found = extract_profiles("alice@example.org")
for service, url in sorted(found.items()):
    print(f"{service:20} → {url}")

This pattern mirrors how threat-intel platforms build digital footprint reports—enumerating active accounts across social media, developer platforms, and SaaS tools.

---

Converting to Tabular Data with Pandas

For analysis workflows, convert the stream to a DataFrame:

import pandas as pd

def holehe_to_dataframe(email: str) -> pd.DataFrame:
    """Collect all Holehe results into a structured DataFrame."""
    records = list(stream_holehe(email))
    
    if not records:
        return pd.DataFrame(columns=["service", "status", "url", "data"])
    
    df = pd.DataFrame(records)
    
    # Normalize column order; expand data dict if needed

    df = df[["service", "status", "url", "data"]]
    
    return df

# Example: Quick statistics

df = holehe_to_dataframe("victim@target.com")
print(f"Total services checked: {len(df)}")
print(f"Registered accounts: {(df['status'] == 'registered').sum()}")
print(df[df["status"] == "registered"][["service", "url"]])

The data column contains nested dictionaries—use pd.json_normalize() if you need to flatten service-specific fields.

---

Asynchronous Processing for High-Volume Checks

For concurrent email checks or integration with async web frameworks:

import asyncio
import json
from typing import AsyncIterator, Dict, Any

async def stream_holehe_async(email: str) -> AsyncIterator[Dict[str, Any]]:
    """Asynchronously stream and parse Holehe output."""
    proc = await asyncio.create_subprocess_exec(
        "holehe", "-f", email,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.DEVNULL,
    )
    
    try:
        async for raw_line in proc.stdout:
            line = raw_line.decode().strip()
            if not line:
                continue
                
            try:
                yield json.loads(line)
            except json.JSONDecodeError:
                continue
    finally:
        await proc.wait()

# Example: Multiple concurrent searches

async def batch_check(emails: list[str]) -> Dict[str, list]:
    """Run Holehe for multiple emails concurrently."""
    results = {}
    
    async def check_one(email: str):
        results[email] = [
            r for r in await async_collect(stream_holehe_async(email))
            if r["status"] == "registered"
        ]
    
    async def async_collect(aiter):
        """Helper to collect async iterator to list."""
        return [item async for item in aiter]
    
    await asyncio.gather(*[check_one(e) for e in emails])
    return results

# Usage

# asyncio.run(batch_check(["a@x.com", "b@y.com"]))

Performance note: The bottleneck is typically network I/O across Holehe's service modules, not Python parsing. Async wrappers excel when orchestrating multiple Holehe instances or integrating with async databases.

---

Accessing Module-Specific Metadata

Individual service implementations in holehe/modules/ control what populates the data field. For example, [holehe/modules/social_media/twitter.py](https://github.com/megadose/holehe/blob/master/holehe/modules/social_media/twitter.py) might include rate-limit headers, while [holehe/modules/software/github.py](https://github.com/megadose/holehe/blob/master/holehe/modules/software/github.py) could expose repository counts.

Inspect the data field defensively:

def inspect_metadata(result: Dict[str, Any]) -> None:
    """Pretty-print service-specific metadata if present."""
    service = result["service"]
    data = result.get("data") or {}
    
    if not data:
        print(f"{service}: no extended metadata")
        return
    
    print(f"{service} metadata:")
    for key, value in data.items():
        print(f"  {key}: {value}")

# Usage within stream

for result in stream_holehe("test@example.com"):
    if result["status"] == "registered":
        inspect_metadata(result)

---

Error Handling and Edge Cases

Scenario	Handling Strategy
Non-zero exit code	Check proc.returncode; Holehe may fail on network errors
Empty output	Email format invalid or all services failed
Malformed JSON lines	Skip with try/except json.JSONDecodeError
Missing url field	Treat as None; some services don't expose direct links

Robust production code should log skipped lines for debugging:

import logging

logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger("holehe.parser")

def safe_parse_line(line: str) -> Dict[str, Any] | None:
    """Parse a line with error logging."""
    try:
        return json.loads(line)
    except json.JSONDecodeError as e:
        logger.warning(f"JSON parse failed: {e!r} on line: {line[:200]}")
        return None

---

Summary

• Holehe outputs NDJSON—one {"service":..., "status":..., "url":..., "data":...} object per line
• Stream with subprocess.Popen and json.loads per line for memory efficiency
• Filter on status == "registered" to isolate active accounts
• Convert to pandas for analytical workflows or async iterate for scalable pipelines
• Inspect data fields for module-specific extensions, with defensive handling for missing keys

---

Frequently Asked Questions

How does Holehe format its output for programmatic consumption?

Holehe prints one JSON object per line to stdout, with fields service, status, url, and data. This newline-delimited JSON format allows streaming parsers to process results as they arrive without buffering the complete output.

Can I import Holehe as a Python library instead of using subprocess?

Yes. Install with pip install holehe and import from holehe.core or individual modules. However, the CLI's NDJSON output remains the cleanest integration point for external applications, as it avoids threading complications with Holehe's async HTTP client.

What information appears in the data field of Holehe's output?

The data dictionary contains service-specific metadata added by individual modules in holehe/modules/. Contents vary—some modules populate user IDs, others include response headers or verification tokens. Always access defensively with .get() since many services return empty data.

How can I process multiple email addresses efficiently?

Use asyncio to run multiple Holehe subprocesses concurrently, or parallelize with concurrent.futures.ProcessPoolExecutor for CPU-bound post-processing. The parsing itself is I/O-bound; the bottleneck is Holehe's network requests to hundreds of services per email.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}