How InsightEngine Queries Private Databases Using SQLAlchemy: Async Architecture Explained

InsightEngine uses an async SQLAlchemy helper in InsightEngine/utils/db.py to query private MySQL or PostgreSQL databases, wrapping raw SQL execution in fetch_all() while higher-level tools like MediaCrawlerDB provide semantic search methods for AI agents.

The InsightEngine component of the bettafish repository (666ghj/bettafish) provides AI agents with secure, asynchronous access to private databases. By leveraging SQLAlchemy 2.x async capabilities, the system abstracts complex connection management behind simple utility functions while maintaining full control over raw SQL execution for private data queries.

Architecture Overview

The database query flow follows a layered architecture that separates connection management from business logic. When an AI agent needs to query private databases using SQLAlchemy, the request flows through four distinct layers: configuration management in InsightEngine/utils/config.py, engine initialization in InsightEngine/utils/db.py, low-level query execution via fetch_all(), and high-level semantic wrappers in InsightEngine/tools/search.py.

This design ensures that private database credentials remain isolated in environment variables while providing both async-native and synchronous-compatible interfaces for agent tools.

Database Configuration and Connection Setup

Environment-Based Configuration

All database connection parameters are centralized in InsightEngine/utils/config.py using a Pydantic settings class. The system reads environment variables including DB_DIALECT, DB_HOST, DB_PORT, DB_USER, DB_PASSWORD, and DB_NAME to construct connection strings without hardcoding sensitive credentials in the source code.

Dynamic URL Construction

The _build_database_url() function in InsightEngine/utils/db.py (lines 28-46) dynamically constructs dialect-specific database URLs. For MySQL databases, it uses the aiomysql driver, while PostgreSQL connections utilize asyncpg. The function also supports a fallback to a complete DATABASE_URL environment variable if provided, overriding individual connection parameters.

# From InsightEngine/utils/db.py - conceptual usage

from InsightEngine.utils.db import _build_database_url

# Automatically selects aiomysql for MySQL or asyncpg for PostgreSQL

url = _build_database_url(
    dialect="mysql",
    host="localhost",
    port=3306,
    user="agent",
    password="secret",
    database="private_data"
)

AsyncEngine Initialization

The get_async_engine() function (lines 49-58 of db.py) implements a singleton pattern to create and cache a single AsyncEngine instance. This engine is configured with production-ready connection pool settings including pool_pre_ping=True to verify connections before use and pool_recycle to prevent stale connections from lingering in the pool.

Executing Queries with fetch_all

The fetch_all() function (lines 61-70 of db.py) serves as the primary interface for executing read-only SQL queries against private databases. This async function accepts a SQL string and optional parameter dictionary, executes the query using SQLAlchemy 2.x text() constructs, and returns results as a list of plain Python dictionaries via result.mappings().all().

import asyncio
from InsightEngine.utils.db import fetch_all

async def query_private_data():
    """Execute a read-only query against the private database."""
    sql = """
        SELECT * FROM bilibili_video 
        WHERE create_time > :start_date 
        ORDER BY view_count DESC 
        LIMIT :limit
    """
    params = {"start_date": "2025-01-01", "limit": 10}
    
    rows = await fetch_all(sql, params)
    return rows

# Usage in async context

results = asyncio.run(query_private_data())
print(results)  # [{'id': 1, 'title': '...', 'view_count': 5000}, ...]

Important: The fetch_all() function is designed specifically for read-only operations. It uses engine.connect() rather than engine.begin(), indicating it does not manage transactions for write operations.

High-Level Database Tools for AI Agents

The MediaCrawlerDB Class

Located in InsightEngine/tools/search.py, the MediaCrawlerDB class provides semantic wrappers around raw SQL queries, enabling AI agents to query private databases using SQLAlchemy without writing SQL directly. This class implements methods like search_hot_content(), search_topic_globally(), and search_topic_on_platform() that construct appropriate SQL strings internally and delegate execution to the lower-level database utilities.

Bridging Sync and Async with _execute_query

Since AI agent frameworks often operate in synchronous contexts, MediaCrawlerDB implements _execute_query() (lines 78-92 of search.py) to bridge the gap. This method checks for an existing asyncio event loop, creates one if necessary using asyncio.new_event_loop(), and runs the async fetch_all() function via run_until_complete().

from InsightEngine.tools.search import MediaCrawlerDB

# Initialize the database tool

db = MediaCrawlerDB()

# Search for hot content from the past week

hot_content = db.search_hot_content(
    time_period="week",
    limit=5
)
print(f"Found {hot_content.results_count} trending items")

# Global topic search across all platforms

global_results = db.search_topic_globally(
    topic="人工智能",
    limit_per_table=20
)

# Platform-specific search with date filtering

weibo_data = db.search_topic_on_platform(
    platform="weibo",
    topic="区块链",
    start_date="2025-08-20",
    end_date="2025-08-20",
    limit=10
)

Key Files and Their Roles

File	Role
InsightEngine/utils/db.py	Async SQLAlchemy engine factory, URL builder, and fetch_all helper for read-only queries.
InsightEngine/utils/config.py	Pydantic settings class exposing database connection environment variables.
InsightEngine/tools/search.py	High-level MediaCrawlerDB class providing semantic search methods for AI agents.
InsightEngine/nodes/*	Reasoning graph nodes that invoke database tools as part of larger insight workflows.

Summary

• InsightEngine queries private databases using SQLAlchemy through a centralized async utility layer in InsightEngine/utils/db.py.
• The architecture separates concerns: config.py handles credentials, db.py manages engine lifecycle and read-only query execution, and search.py provides semantic wrappers for AI agents.
• SQLAlchemy 2.x async patterns power the implementation, with dialect-specific drivers (aiomysql for MySQL, asyncpg for PostgreSQL) selected automatically based on configuration.
• The MediaCrawlerDB class bridges synchronous AI agent frameworks with async database operations by managing event loops internally, enabling seamless private database queries without exposing SQL complexity to end users.

Frequently Asked Questions

How does InsightEngine handle database credentials securely?

InsightEngine uses environment variables exclusively for database credentials, parsed through a Pydantic settings class in InsightEngine/utils/config.py. The system reads DB_HOST, DB_USER, DB_PASSWORD, and other connection parameters at runtime, ensuring no sensitive data is hardcoded in the source repository. Additionally, the _build_database_url() function supports a complete DATABASE_URL environment variable as a fallback, allowing integration with secret management systems that inject full connection strings.

Can InsightEngine write data to private databases or only read?

The current implementation in InsightEngine/utils/db.py is designed specifically for read-only operations. The fetch_all() function uses engine.connect() rather than engine.begin(), and returns results via result.mappings().all() without committing transactions. While the underlying AsyncEngine from SQLAlchemy 2.x supports write operations, the InsightEngine utility layer does not expose methods for INSERT, UPDATE, or DELETE operations, maintaining a strict read-only safety boundary for AI agent interactions.

What database dialects does InsightEngine support?

InsightEngine supports MySQL and PostgreSQL through dialect-specific async drivers. When _build_database_url() processes the DB_DIALECT setting, it automatically selects aiomysql for MySQL connections and asyncpg for PostgreSQL connections, constructing the appropriate SQLAlchemy URL format (e.g., mysql+aiomysql:// or postgresql+asyncpg://). This abstraction allows the same codebase to query private databases using SQLAlchemy regardless of whether the underlying infrastructure uses MySQL or PostgreSQL.

How do AI agents call database methods if they run synchronously?

AI agents interact with the database through the MediaCrawlerDB class in InsightEngine/tools/search.py, which handles the sync-to-async bridge internally. The _execute_query() method (lines 78-92) checks for an existing asyncio event loop; if none exists or if the current loop is closed, it creates a new loop using asyncio.new_event_loop() and runs the async fetch_all() function via run_until_complete(). This pattern allows synchronous agent frameworks to query private databases using SQLAlchemy without requiring the caller to manage async/await syntax or event loops explicitly.

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