# How InsightEngine Queries Private Databases Using SQLAlchemy: Async Architecture Explained

> Discover how InsightEngine queries private databases with SQLAlchemy. Learn about its async architecture and raw SQL execution for AI agents.

- Repository: [BaiFu/bettafish](https://github.com/666ghj/bettafish)
- Tags: deep-dive
- Published: 2026-02-23

---

**InsightEngine uses an async SQLAlchemy helper in [`InsightEngine/utils/db.py`](https://github.com/666ghj/bettafish/blob/main/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`](https://github.com/666ghj/bettafish/blob/main/InsightEngine/utils/config.py), engine initialization in [`InsightEngine/utils/db.py`](https://github.com/666ghj/bettafish/blob/main/InsightEngine/utils/db.py), low-level query execution via `fetch_all()`, and high-level semantic wrappers in [`InsightEngine/tools/search.py`](https://github.com/666ghj/bettafish/blob/main/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`](https://github.com/666ghj/bettafish/blob/main/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`](https://github.com/666ghj/bettafish/blob/main/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.

```python

# 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`](https://github.com/666ghj/bettafish/blob/main/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`](https://github.com/666ghj/bettafish/blob/main/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()`.

```python
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`](https://github.com/666ghj/bettafish/blob/main/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`](https://github.com/666ghj/bettafish/blob/main/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()`.

```python
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`](https://github.com/666ghj/bettafish/blob/main/InsightEngine/utils/db.py) | Async SQLAlchemy engine factory, URL builder, and `fetch_all` helper for read-only queries. |
| [`InsightEngine/utils/config.py`](https://github.com/666ghj/bettafish/blob/main/InsightEngine/utils/config.py) | Pydantic settings class exposing database connection environment variables. |
| [`InsightEngine/tools/search.py`](https://github.com/666ghj/bettafish/blob/main/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`](https://github.com/666ghj/bettafish/blob/main/InsightEngine/utils/db.py).
- The architecture separates concerns: [`config.py`](https://github.com/666ghj/bettafish/blob/main/config.py) handles credentials, [`db.py`](https://github.com/666ghj/bettafish/blob/main/db.py) manages engine lifecycle and read-only query execution, and [`search.py`](https://github.com/666ghj/bettafish/blob/main/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`](https://github.com/666ghj/bettafish/blob/main/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`](https://github.com/666ghj/bettafish/blob/main/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`](https://github.com/666ghj/bettafish/blob/main/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.