How to Embed Holehe in a Python Application: A Complete Guide for Developers
You can embed Holehe programmatically by importing import_submodules and get_functions from holehe/core.py to discover service modules, then executing checks concurrently with Trio and httpx—bypassing the CLI entirely.
Holehe is an open-source Python tool for email OSINT that checks whether an address is registered across hundreds of web services. While most users run it from the command line, its modular architecture—centered in megadose/holehe—allows direct embedding into scripts, APIs, and data pipelines. This guide shows you how to leverage the same internal APIs that power the CLI.
Understanding Holehe's Core Architecture
The library's design separates discovery, execution, and presentation into distinct layers:
- Module discovery happens in
import_submodules()atholehe/core.py:37 - Function extraction is handled by
get_functions()atholehe/core.py:50 - Concurrent execution uses
launch_module()atholehe/core.py:66with Trio and httpx
Each service module under holehe/modules/ contains a single async function with a standardized signature: async def service_name(email, client, out). This consistency makes the library inherently embeddable.
Basic Embed Pattern: Four Essential Steps
Every embedding approach follows the same sequence:
- Import modules —
import_submodules("holehe.modules")scans the package tree - Select functions —
get_functions(modules, args)filters based on your configuration - Execute concurrently — Trio nursery +
launch_module()runs all checks - Process results — Each function returns a dictionary with
domain,exists,emailrecovery, and other fields
Example 1: Simple Async Script
This standalone example demonstrates the complete flow without Flask or FastAPI overhead:
import httpx
import trio
from holehe.core import import_submodules, get_functions, launch_module
from holehe.instruments import TrioProgress
async def check_email(email: str):
# Load all service modules once
modules = import_submodules("holehe.modules")
# Minimal args namespace matching CLI expectations
class Args:
nopasswordrecovery = False
nocolor = True
onlyused = False
# Extract callable functions from discovered modules
services = get_functions(modules, Args())
# Shared httpx client for connection reuse
client = httpx.AsyncClient(timeout=10)
# Progress tracking (optional)
progress = TrioProgress(len(services))
trio.lowlevel.add_instrument(progress)
results = []
async with trio.open_nursery() as nursery:
for svc in services:
nursery.start_soon(launch_module, svc, email, client, results)
trio.lowlevel.remove_instrument(progress)
await client.aclose()
return results
# Run with: trio.run(check_email, "target@example.com")
if __name__ == "__main__":
email = "demo@example.com"
result = trio.run(check_email, email)
for entry in result:
print(f"{entry['domain']}: exists={entry.get('exists')}")
Key implementation details:
TrioProgressis optional—remove the instrument lines for silent execution- The
Argsclass mimics the subset of CLI arguments thatget_functionsactually reads launch_moduleappends results to the sharedresultslist automatically
Example 2: FastAPI/Flask Web Service
For production APIs, preload modules at startup to eliminate per-request overhead:
from fastapi import FastAPI, HTTPException
import trio
import httpx
from holehe.core import import_submodules, get_functions, launch_module
app = FastAPI()
# Initialize once at module load time
MODULES = import_submodules("holehe.modules")
class Args:
nopasswordrecovery = False
nocolor = True
onlyused = False
SERVICES = get_functions(MODULES, Args())
@app.get("/check/{email}")
async def check_email_endpoint(email: str):
async def run_checks():
client = httpx.AsyncClient(timeout=15)
results = []
async with trio.open_nursery() as nursery:
for svc in SERVICES:
nursery.start_soon(launch_module, svc, email, client, results)
await client.aclose()
return results
# FastAPI doesn't support Trio natively; bridge with trio.run
data = trio.run(run_checks)
return {"email": email, "services": data}
# Flask equivalent (synchronous WSGI interface)
"""
from flask import Flask, request, jsonify
flask_app = Flask(__name__)
@flask_app.route("/api/check")
def flask_check():
email = request.args.get("email")
if not email:
return jsonify({"error": "email required"}), 400
async def run():
client = httpx.AsyncClient(timeout=10)
results = []
async with trio.open_nursery() as nursery:
for svc in SERVICES:
nursery.start_soon(launch_module, svc, email, client, results)
await client.aclose()
return results
return jsonify(trio.run(run))
"""
Performance consideration: Module discovery via import_submodules involves filesystem traversal and dynamic imports. By executing it at import time rather than per-request, you reduce latency from hundreds of milliseconds to near-zero.
Example 3: Selective Service Targeting
Reduce API load and execution time by filtering modules before function extraction:
# Target only social media platforms
social_modules = {
name: mod for name, mod in MODULES.items()
if name.startswith("holehe.modules.social_media")
}
social_services = get_functions(social_modules, Args())
# Or exclude specific categories
non_adult_modules = {
name: mod for name, mod in MODULES.items()
if "adult" not in name
}
# Select specific services by name
target_services = ["twitter", "github", "instagram"]
filtered_modules = {
name: mod for name, mod in MODULES.items()
if any(t in name for t in target_services)
}
The MODULES dictionary uses fully qualified module names as keys (e.g., holehe.modules.social_media.twitter). Inspect list(MODULES.keys()) to see available targets.
Working with Results
Each service function returns a standardized dictionary. The launch_module wrapper in holehe/core.py:66 appends these to your results list:
| Key | Type | Description |
|---|---|---|
domain |
str | Service domain checked |
exists |
bool | Whether email is registered |
emailrecovery |
str/null | Recovery email if exposed |
phoneNumber |
str/null | Phone number if exposed |
others |
dict | Additional service-specific data |
Process results programmatically without the CLI's print_result formatter:
def summarize_findings(results):
hits = [r for r in results if r.get("exists")]
recoverable = [r for r in results if r.get("emailrecovery")]
return {
"total_checked": len(results),
"registered_count": len(hits),
"registered_domains": [r["domain"] for r in hits],
"recovery_exposed": len(recoverable)
}
Handling Custom Instrumentation
Replace or extend the default progress reporting by implementing your own Trio instrument:
from trio import Instrument
class CustomInstrument(Instrument):
def __init__(self, total):
self.total = total
self.completed = 0
def task_exited(self, task):
self.completed += 1
print(f"Progress: {self.completed}/{self.total}")
# Usage
instrument = CustomInstrument(len(services))
trio.lowlevel.add_instrument(instrument)
# ... run nursery ...
trio.lowlevel.remove_instrument(instrument)
The built-in TrioProgress in holehe/instruments.py uses tqdm for terminal output, which may not suit web applications or structured logging environments.
Summary
Embed Holehe in Python applications by following these proven patterns:
- Import
import_submodulesandget_functionsfromholehe/core.pyto leverage the same discovery mechanism as the CLI - Preload service modules at application startup for production deployments
- Use
httpx.AsyncClientwith appropriate timeouts for reliable HTTP handling - Execute checks via Trio nurseries with
launch_modulefor proper concurrency - Filter the modules dictionary before calling
get_functionsto limit scope and reduce API traffic - Process raw result dictionaries directly rather than relying on CLI formatters
Frequently Asked Questions
Can I embed Holehe without Trio?
No—Holehe's architecture is built entirely on Trio for structured concurrency. The launch_module function and all service implementations use Trio primitives. To use asyncio, run Trio inside an asyncio loop with trio.lowlevel.start_guest_run or bridge via anyio, though this adds complexity.
How do I reduce the number of HTTP requests per email check?
Filter the modules dictionary before passing it to get_functions. For example, select only social_media modules or explicitly whitelist target domains. Each module makes independent requests; fewer modules equals fewer connections.
Is holehe.main() suitable for embedding?
Avoid holehe.main() for embedding—it parses sys.argv, writes to stdout, and exits the process. Use the internal APIs (import_submodules, get_functions, launch_module) demonstrated above for programmatic control.
How do I handle rate limiting or add delays between requests?
Holehe does not implement global rate limiting. You can modify launch_module calls to include semaphores, or subclass the httpx client with custom transport layers. For strict rate limits, consider executing modules in batches with trio.sleep between nursery blocks.
Can I add custom service modules to the discovery system?
Yes—place Python files following the standard async signature in holehe/modules/ (or a custom package) and point import_submodules at your package path. The discovery system uses pkgutil.iter_modules and is agnostic to module origin.
Have a question about this repo?
These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:
curl -s "https://instagit.com/install.md" Maintain an open-source project? Get it listed too →