db_core

db_core is an async-first, multi-backend database abstraction library for Python. It provides a unified, fluent query-builder API across PostgreSQL, Elasticsearch, and Apache Doris — making it the foundation for data access in the Dream ecosystem. It is designed to be imported by other libraries such as repo_core.

Architecture

                  ┌─────────────────────────────┐
                  │       db_core (this)        │
                  │  Unified query builder API  │
                  └────────────┬────────────────┘
                               │
          ┌────────────────────┼────────────────────┐
          │                    │                    │
   ┌──────▼──────┐    ┌───────▼───────┐    ┌───────▼──────┐
   │ PostgresClient│   │ElasticsearchClient│ │ DorisClient │
   │ (SQLAlchemy  │   │ (async           │ │ (SQLAlchemy │
   │  + asyncpg)  │   │  elasticsearch)  │ │  + aiomysql) │
   └──────────────┘   └─────────────────┘ └──────────────┘

Class hierarchy

DBClient (Abstract Base)          ← find_one, find_many, count, exists, table()
│                                   ← create, update, delete, bulk_update, bulk_delete
├── PostgresClient                ← PostgreSQL (pool_size=25, max_overflow=15)
├── DorisClient                   ← Apache Doris via MySQL protocol
├── ElasticsearchClient           ← ES 8.x (async, bool queries, nested paths)
├── DocumentClient                ← Marker for document stores (future)
├── VectorClient                  ← Marker for vector DBs (Pinecone, Milvus — future)
└── OLAPClient                    ← Marker for OLAP systems (ClickHouse — future)

• Async-first: All I/O methods are async def. Built on SQLAlchemy 2.0+ async engine and asyncio.
• Connection pooling: PostgreSQL pools at 25 connections + 15 overflow with pool_pre_ping=True. Doris pools at 10 + 5 overflow.
• Fluent builder: Every query is chainable — table().select().where().order_by().limit().all().
• Backend-agnostic query model: Query, WhereCondition, and WhereGroup are backend-neutral dataclasses. Each client translates them to native query syntax (SQL, Elasticsearch DSL).
• All methods snake_case: Every public method uses snake_case naming (find_one, find_many, find_by_id, delete_by_id, bulk_update, etc.).

Install

As a standalone package

pip install .

As a dependency from another service

db_core is consumed by repo_core in the Dream workspace via local path dependencies. Add this to your consuming project's requirements.txt or pyproject.toml:

# requirements.txt — relative path to db_core
../db_core[elasticsearch]

# pyproject.toml
dependencies = [
    "db_core",
]

Then install with:

pip install -e /path/to/core/backend/db_core

To install with optional backends:

pip install -e "/path/to/core/backend/db_core[elasticsearch,doris]"

For local development and testing

pip install .
pip install -r requirements-dev.txt

Services consuming db_core

Service	Backend(s) used	How it integrates
repo_core	PostgreSQL, Elasticsearch, Doris	Wraps PostgresClient, ElasticsearchClient, and DorisClient in QueryBuilder. Provides auto-wired BaseRepository with lazy client singletons. Uses db_core.schema for migration DDL generation (TableDef → SQLAlchemy MetaData).
sam-backend	PostgreSQL	Uses db_core factories indirectly via repo_core auto-wired repositories.
persistence	PostgreSQL, Elasticsearch	Uses db_core factories indirectly via repo_core auto-wired repositories.

Importing

from db_core import (
    PostgresClient,
    create_postgres_client,
    get_postgres_client,
    create_doris_client,
    get_doris_client,
    QueryBuilder,
    QuerySet,
)

Available public exports

• PostgresClient — PostgreSQL client (SQLAlchemy + asyncpg)
• create_postgres_client / get_postgres_client — Factory functions (resolve URL from env vars)
• create_elasticsearch_client / get_elasticsearch_client — Factory functions for Elasticsearch
• create_doris_client / get_doris_client — Factory functions for Apache Doris
• QueryBuilder — Fluent query builder
• QuerySet — Rich result wrapper with chainable operations
• SchemaBuilder — Build TableDef instances from dict/config
• TableDef, ColumnDef, IndexDef, ForeignKeyDef — SQL schema definition dataclasses
• table_to_sqlalchemy, table_to_sqlalchemy_doris — Convert TableDef → SQLAlchemy Table (used by repo_core migrations)
• table_defs_to_metadata — Aggregate TableDefs into SQLAlchemy MetaData
• ESMappingBuilder, ESMappingGenerator — Elasticsearch mapping/index generation
• ESFieldDef, ESIndexDef, ESIndexTemplateDef — ES schema definition dataclasses
• ESField — Annotation metadata for per-field ES overrides on DomainModels
• Eq, Gt, Gte, Lt, Lte, In, Like, ILike, NotEq — Query operators

Elasticsearch and Doris client classes (ElasticsearchClient, DorisClient) require optional extras. Their factory functions are always available from the root __init__.py.

Quick start

PostgreSQL

from db_core import create_postgres_client

# Create a client from environment variables or explicit URL
db = create_postgres_client()
await db.connect()

# Query with the fluent builder
users = await (
    db.table("users")
      .select("id", "name", "email")
      .where("status", "=", "active")
      .where_gte("age", 18)
      .order_by("-created_at")
      .limit(10)
      .all()
)

await db.disconnect()

Elasticsearch

from db_core.clients.elasticsearch import ElasticsearchClient

es = ElasticsearchClient(hosts="http://localhost:9200")
await es.connect()

# Same fluent builder API — translated to Elasticsearch DSL
results = await (
    es.table("users")
      .where("age", ">=", 18)
      .where("status", "active")
      .order_by("-age")
      .limit(10)
      .all()
)

await es.disconnect()

Apache Doris

from db_core.clients.doris import DorisClient

doris = DorisClient("mysql+aiomysql://root@localhost:9030/analytics")
await doris.connect()

logs = await doris.table("event_logs").where("event_type", "click").limit(100).all()

await doris.disconnect()

Environment configuration

PostgreSQL (create_postgres_client())

Resolution order:

1. Explicit url parameter
2. DATABASE_URL env var
3. POSTGRES_URL env var
4. PG_URL / PG_DSN env var
5. Individual vars: PG_USER, PG_PASSWORD, PG_HOST (default: localhost), PG_PORT (default: 5432), PG_DATABASE

export DATABASE_URL="postgresql+asyncpg://user:pass@localhost:5432/dbname"
# or
export PG_USER=myuser PG_PASSWORD=mypass PG_HOST=db.example.com PG_PORT=5432 PG_DATABASE=mydb

Elasticsearch (create_elasticsearch_client())

Resolution order:

1. Explicit hosts parameter
2. ELASTICSEARCH_URL env var
3. ES_URL env var
4. ES_HOST env var (with ES_PORT default 9200, ES_SCHEME default http)

export ELASTICSEARCH_URL="http://localhost:9200"
# or
export ES_HOST=es.example.com ES_PORT=9200 ES_SCHEME=https

Apache Doris (create_doris_client())

Resolution order:

1. Explicit url parameter
2. DORIS_URL / DORIS_DSN env var
3. Individual vars: DORIS_USER (default: root), DORIS_PASSWORD, DORIS_HOST (default: localhost), DORIS_PORT (default: 9030), DORIS_DATABASE

export DORIS_URL="mysql+aiomysql://root:password@doris-fe:9030/analytics"

Query builder — complete reference

Column selection

db.table("users").select("id", "name", "email")   # explicit columns
db.table("users")                                   # defaults to SELECT *

WHERE conditions

# Basic equality (two-argument form)
.where("status", "active")

# Explicit operator (three-argument form)
.where("age", ">=", 18)
.where("email", "!=", "spam@example.com")
.where("role", "IN", ["admin", "moderator"])
.where("name", "LIKE", "John%")
.where("name", "ILIKE", "john%")            # case-insensitive (PostgreSQL only)

# Convenience helpers
.where_eq("status", "active")               # same as .where("status", "active")
.where_gt("age", 21)
.where_gte("age", 18)
.where_lt("age", 65)
.where_lte("score", 100)
.where_not_eq("deleted", True)
.where_in("role", ["admin", "mod"])
.where_like("email", "%@gmail.com")
.where_ilike("city", "%york%")

OR conditions

# Simple OR
db.table("users").where("age", 25).or_where("age", 30)

# Grouped AND conditions combined with OR
db.table("users").where("active", True).where_group(lambda q:
    q.where("role", "admin").or_where("role", "moderator")
)

Nested path queries (Elasticsearch only)

es.table("articles").where_nested("comments", lambda q:
    q.where("comments.author", "Alice").where("comments.likes", ">", 10)
)

Ordering, pagination

.order_by("name")           # ASC
.order_by("-created_at")    # DESC (prefix with '-')
.order_by("status", "-age") # multiple fields
.limit(50)
.offset(100)                # skip first 100 rows

Execution methods

query = db.table("users").where("status", "active")

result = await query.get()          # → QuerySet (lazy wrapper)
rows   = await query.all()          # → List[Dict] (all rows)
row    = await query.first()        # → Optional[Dict] (first row)
count  = await query.count()        # → int (row count)

SQL/query preview (debugging)

# Preview without executing
sql = db.table("users").where("age", ">", 21).to_sql()
print(sql)  # SELECT * FROM users WHERE age > 21

# Detailed debug output (SQL + params)
debug = db.debug_query(query_builder.query)
print(debug)  # SQL: SELECT * FROM users WHERE age > 21\nParams: {}

# PostgreSQL EXPLAIN
plan = await db.explain(query_builder.query, analyze=True)
print(plan)

Convenience methods (DBClient base class)

All clients inherit these methods from DBClient:

Method	Returns	Description
find_one(table, conditions)	Optional[Dict]	Find a single record
find_many(table, conditions, limit, offset, order_by)	List[Dict]	Find multiple records
find_by_id(table, id, id_field="id")	Optional[Dict]	Find by primary key
find_all(table, limit, offset, order_by)	List[Dict]	Find all records
count(table, conditions)	int	Count matching records
exists(table, conditions)	bool	Check if records exist

user = await db.find_one("users", {"email": "alice@example.com"})
active_users = await db.find_many("users", {"status": "active"}, limit=10, order_by=["-created_at"])
user_123 = await db.find_by_id("users", 123)
all_users = await db.find_all("users", limit=100)
user_count = await db.count("users", {"status": "active"})
has_admin = await db.exists("users", {"role": "admin"})

CRUD operations (PostgreSQL, Doris)

DBClient provides a full set of mutation methods. Clients that don't support mutations (e.g., Elasticsearch) raise NotImplementedError for unsupported operations.

Method	Returns	Description
create(table, data, returning=None)	Optional[Dict]	Insert one row
create_many(table, data, returning=None)	List[Dict]	Insert multiple rows
save(table, data, id_field="id", returning=None)	Optional[Dict]	Insert or update (upsert by id)
update(table, conditions, values, limit=None)	int	Update matching rows
update_by_id(table, id, values, id_field="id")	int	Update by primary key
delete(table, conditions, limit=None)	int	Delete matching rows
delete_by_id(table, id, id_field="id")	int	Delete by primary key
bulk_update(table, ids, values)	int	Update many rows by ID list
bulk_delete(table, ids)	int	Delete many rows by ID list

# Create
new_user = await db.create("users", {"name": "John", "email": "john@example.com", "age": 27},
                            returning=["id", "created_at"])

# Save (upsert — inserts if no id, updates if id present)
saved = await db.save("users", {"id": new_user["id"], "name": "John D.", "age": 28})

# Update
updated_count = await db.update("users", {"status": "inactive"}, {"status": "active"})
await db.update_by_id("users", 42, {"verified": True})

# Delete
await db.delete("users", {"status": "banned"})
await db.delete_by_id("users", 1337)

# Bulk operations
await db.bulk_update("users", [1, 2, 3], {"status": "verified"})
await db.bulk_delete("users", [99, 100, 101])

QuerySet — result manipulation

After executing a query, you get a QuerySet with chainable operations:

qs = await db.table("users").where("status", "inactive").get()

rows = qs.all()                 # List[Dict]
first = qs.first()              # Optional[Dict]
exactly_one = qs.one()          # Dict (raises ValueError if ≠ 1)
count = qs.count()              # int

# Bulk update/delete on result set
updated = await qs.update(status="active")      # updates all rows in the result set
deleted = await qs.delete()                     # deletes all rows in the result set
# .remove() is an alias for .delete()

Schema module — TableDef & SQLAlchemy generation

The db_core.schema package provides dataclasses for describing database tables and utilities for converting them into SQLAlchemy metadata. This is used by repo_core's migration system to auto-generate DDL from Pydantic domain models.

from db_core.schema import (
    SchemaBuilder,
    TableDef,
    ColumnDef,
    IndexDef,
    ForeignKeyDef,
    table_to_sqlalchemy,
    table_to_sqlalchemy_doris,
    table_defs_to_metadata,
)

# Define a table
users = TableDef(
    name="users",
    columns=[
        ColumnDef("id", "UUID", primary_key=True),
        ColumnDef("name", "TEXT", nullable=False),
        ColumnDef("email", "TEXT"),
        ColumnDef("created_at", "TIMESTAMPTZ", default="NOW()"),
    ],
    indexes=[IndexDef(["email"])],
)

# Convert to SQLAlchemy Table (PostgreSQL dialect)
sa_table = table_to_sqlalchemy(users)

# Convert to SQLAlchemy Table (Doris/MySQL dialect)
sa_table_doris = table_to_sqlalchemy_doris(users)

# Aggregate multiple TableDefs into SQLAlchemy MetaData
metadata = table_defs_to_metadata([users, posts])

Schema module — Elasticsearch mappings

The db_core.schema package also provides dataclasses and builders for Elasticsearch index mappings — mirroring the SQL TableDef/SchemaBuilder pattern for the search-engine backend.

Defining mappings with ESFieldDef / ESIndexDef

from db_core.schema import (
    ESMappingBuilder,
    ESFieldDef,
    ESIndexDef,
)

# Define a flat index
users_index = ESIndexDef(
    name="users",
    properties=[
        ESFieldDef("id", "keyword"),
        ESFieldDef("username", "text",
                   fields={"keyword": {"type": "keyword"}}),
        ESFieldDef("created_at", "date", format_="epoch_millis"),
    ],
    settings={"number_of_shards": 1, "number_of_replicas": 0},
)

body = ESMappingBuilder.build_create_body(users_index)
# → {"settings": {...}, "mappings": {"properties": {...}}}

await es_client.create_index("users", body)

Nested fields (recursive properties)

# ESFieldDef.properties supports recursion for nested/object types
messages_index = ESIndexDef(
    name="messages",
    properties=[
        ESFieldDef("id", "keyword"),
        ESFieldDef("message", "text",
                   fields={"keyword": {"type": "keyword"}}),
        ESFieldDef("replies", "nested", properties=[
            ESFieldDef("user_id", "keyword"),
            ESFieldDef("text", "text"),
        ]),
    ],
)

Auto-generating mappings from DomainModels

from db_core.schema import ESMappingGenerator, ESMappingBuilder

# One-liner: DomainModel → ESIndexDef
index_def = ESMappingGenerator.from_model(
    TelegramMessage, "telegram_messages",
    number_of_shards=1, number_of_replicas=0,
)

body = ESMappingBuilder.build_create_body(index_def)

if not await es_client.index_exists("telegram_messages"):
    await es_client.create_index("telegram_messages", body)

ESMappingGenerator.from_model() inspects:

• model._search_include — which fields to index
• model.model_fields — Pydantic field types (auto-maps str→text, int→long, etc.)
• list[DomainModel] fields — converted to "nested" with recursive sub-properties
• Annotated[str, ESField(analyzer="english")] — per-field overrides

Index templates

from db_core.schema import ESIndexTemplateDef

tpl = ESIndexTemplateDef(
    name="logs-template",
    index_patterns=["logs-*"],
    template=users_index,
    priority=100,
    composed_of=["common-settings"],
)
body = ESMappingBuilder.build_index_template(tpl)

Index lifecycle methods on ElasticsearchClient

The client now includes full index management:

Method	Purpose
index_exists(index) → bool	Check if index exists
create_index(index, body)	Create index with settings/mappings
delete_index(index)	Delete an index
put_mapping(index, body)	Update mapping on existing index
get_mapping(index) → dict	Retrieve current mapping
index_document(index, id, doc)	Index (create/replace) a document
delete_document(index, id)	Delete a document by ID

Migrations

db_core powers migrations across all three backends. PostgreSQL and Doris use repo_core's MigrationBuilder fluent DSL (built on db_core.schema). Elasticsearch uses db_core.schema directly via ESMappingGenerator + ESMappingBuilder.

PostgreSQL — fluent MigrationBuilder

from repo_core.migrations.builder import MigrationBuilder
from repo_core.migrations.runner import Migration

def get_postgres_migrations() -> list[Migration]:
    return [
        # ── Create a table ──────────────────────────────────
        Migration(
            version="enriched_message_0001",
            description="Create enriched_messages table",
            up=(
                MigrationBuilder("enriched_messages")
                .uuid("id", primary_key=True)
                .bigint("peer_id", nullable=False)
                .bigint("message_id", nullable=False)
                .text("message_text", nullable=True)
                .jsonb("entities", nullable=True)
                .jsonb("media", nullable=True)
                .boolean("is_post", nullable=True, default=False)
                .bigint("crawled_at", nullable=True)
                .timestamp("created_at", nullable=True, default="now()")
                .timestamp("updated_at", nullable=True, default="now()")
                .index(["peer_id", "message_id"])
                .create_table()
            ),
            down=MigrationBuilder("enriched_messages").drop_table(),
        ),

        # ── Alter: add/drop columns, add/drop indexes ──────
        Migration(
            version="enriched_message_0002",
            description="Add channel_id column and unique index",
            up=(
                MigrationBuilder("enriched_messages")
                .add_column("channel_id", "BIGINT", nullable=True)
                .index(["channel_id"], unique=True)
                .alter_table()
            ),
            down=(
                MigrationBuilder("enriched_messages")
                .drop_index("uq_enriched_messages_channel_id")
                .drop_column("channel_id")
                .alter_table()
            ),
        ),
    ]

Doris (warehouse) — same DSL, dialect-aware

def get_doris_migrations() -> list[Migration]:
    return [
        Migration(
            version="tg_message_doris_0001",
            description="Create telegram_messages warehouse table in Doris",
            dialect="doris",
            up=(
                MigrationBuilder("telegram_messages", dialect="doris")
                .uuid("id")
                .bigint("telegram_id", nullable=False)
                .bigint("message_sent_at", nullable=True)
                .text("message", nullable=True)
                .jsonb("peer_id", nullable=True)
                .date("partition_date", nullable=True)
                # Doris-specific distribution & key clauses
                .doris_unique_key(["id"])
                .doris_distributed_by("id", buckets=10)
                .doris_partition_by_range("partition_date")
                .create_table()
            ),
            down=MigrationBuilder("telegram_messages", dialect="doris").drop_table(),
        ),
    ]

The Doris builder handles:

• doris_unique_key([cols]) / doris_duplicate_key([cols]) — data model
• doris_distributed_by(col, buckets=N) — hash distribution
• doris_partition_by_range(col) — range partitioning
• doris_properties({"replication_num": "1"}) — table properties

Elasticsearch — model-driven index creation

ES "migrations" follow a different pattern — there is no MigrationBuilder. Instead, define your DomainModel with _search_include and use ESMappingGenerator + ESMappingBuilder to produce the index body:

from typing import Annotated
from repo_core.base.models import DomainModel
from db_core.schema import (
    ESMappingBuilder,
    ESMappingGenerator,
    ESField,
)


# ── Step 1: Define the model with ES annotations ──────────
class Article(DomainModel):
    _search_include = {"id", "title", "body", "tags", "published_at"}

    id: str
    title: Annotated[str, ESField(analyzer="english")]
    body: str
    tags: list[str]
    published_at: int


# ── Step 2: Generate the index definition ─────────────────
index_def = ESMappingGenerator.from_model(
    Article, "articles",
    number_of_shards=1,
    number_of_replicas=0,
)

body = ESMappingBuilder.build_create_body(index_def)
# → {
#     "settings": {"number_of_shards": 1, "number_of_replicas": 0},
#     "mappings": {
#         "properties": {
#             "id": {"type": "keyword"},
#             "title": {"type": "text", "analyzer": "english",
#                       "fields": {"keyword": {"type": "keyword"}}},
#             "body": {"type": "text",
#                      "fields": {"keyword": {"type": "keyword"}}},
#             "tags": {"type": "text",
#                      "fields": {"keyword": {"type": "keyword"}}},
#             "published_at": {"type": "long"}
#         }
#     }
# }


# ── Step 3: Create the index (idempotent) ─────────────────
async def ensure_es_index(es_client, index_name: str, model_cls: type) -> None:
    """Ensure an ES index exists with the correct mapping."""
    if not await es_client.index_exists(index_name):
        idx = ESMappingGenerator.from_model(model_cls, index_name,
                                            number_of_shards=1)
        body = ESMappingBuilder.build_create_body(idx)
        await es_client.create_index(index_name, body)

# Run at startup:
await ensure_es_index(es_client, "articles", Article)

For updating mappings on an existing index (add fields only — ES is append-only for mappings):

# Re-generate from the updated model
new_index = ESMappingGenerator.from_model(Article, "articles")
mapping_body = ESMappingBuilder.build_mapping(new_index)
await es_client.put_mapping("articles", mapping_body)

For nested documents (list[DomainModel] fields), the generator handles recursion automatically:

class Comment(DomainModel):
    _search_include = {"user_id", "text"}

    user_id: str
    text: str

class Post(DomainModel):
    _search_include = {"id", "title", "comments"}

    id: str
    title: str
    comments: list[Comment] | None = None  # → ES "nested" with sub-properties


# comments becomes:
# {"comments": {"type": "nested", "properties": {
#     "user_id": {"type": "text", "fields": {"keyword": {"type": "keyword"}}},
#     "text": {"type": "text", "fields": {"keyword": {"type": "keyword"}}}
# }}}

Connection lifecycle

# Create → Connect → Use → Disconnect
db = create_postgres_client()
await db.connect()          # verifies engine is ready (SELECT 1)

# ... do work ...

await db.disconnect()       # disposes engine, closes all pooled connections

Important: Always call disconnect() when done, especially in long-running services. The connection pool holds connections until disposed. In FastAPI lifetime handlers or similar, connect on startup and disconnect on shutdown.

FastAPI integration pattern

from contextlib import asynccontextmanager
from fastapi import FastAPI
from db_core import create_postgres_client

db = create_postgres_client()

@asynccontextmanager
async def lifespan(app: FastAPI):
    await db.connect()
    yield
    await db.disconnect()

app = FastAPI(lifespan=lifespan)

Consuming db_core from other services

Pattern 1: Auto-wired via repo_core (recommended)

# repo_core handles db_core client creation automatically
from repo_core.domains.article.model import Article
from repo_core.domains.article.repository import ArticleRepository

repo = ArticleRepository(model=Article)  # auto-wires create_postgres_client()
async with repo:
    articles = await repo.all("articles", limit=10)

Pattern 2: Direct usage

# In any service that needs database access
from db_core import create_postgres_client
from db_core.clients.elasticsearch import ElasticsearchClient
from db_core.clients.doris import create_doris_client

db = create_postgres_client()
es = ElasticsearchClient(hosts="http://localhost:9200")
doris = create_doris_client()

await db.connect()
await es.connect()
await doris.connect()

Pattern 3: Dependency injection (FastAPI style)

from fastapi import Depends
from db_core import PostgresClient, create_postgres_client

_db: PostgresClient | None = None

async def get_db() -> PostgresClient:
    global _db
    if _db is None:
        _db = create_postgres_client()
        await _db.connect()
    return _db

@app.get("/users")
async def list_users(db: PostgresClient = Depends(get_db)):
    return await db.find_all("users", limit=50)

Testing

# Start the shared Dream infrastructure (from workspace root)
cd infrastructure/development && docker compose up -d

# Run tests via the project scripts (recommended)
./run_tests.fish      # Fish shell
./run_tests.sh        # Bash/Zsh

# Or run pytest directly (after venv setup and infrastructure is running)
pytest tests/ -v

The test suite includes:

• Query builder — 28 tests covering select, where, OR, grouping, operators, pagination
• QuerySet — 6 tests for result manipulation
• PostgreSQL — 12 tests for connection, CRUD, EXPLAIN, pooling, timestamps
• SQL preview — 11 tests for to_sql(), debug_query(), raw_query()
• Elasticsearch — tests for search body generation, nested queries
• Apache Doris — tests for MySQL-protocol query execution
• Schema builder — tests for TableDef, ColumnDef, IndexDef, FK def construction
• ES schema builder — 28 tests for ESFieldDef, ESMappingBuilder, index templates
• ES mapping gen — 30 tests for ESMappingGenerator, ESField overrides, nested models
• Execute raw — tests for execute_raw() with and without multi-statement
• Factory — tests for env var resolution, URL construction

Project structure

db_core/
├── __init__.py              # Package exports (PostgresClient, QueryBuilder, QuerySet, factories)
├── pyproject.toml           # Build config with optional deps (elasticsearch, doris)
├── requirements.txt         # Core deps: sqlalchemy[asyncio], asyncpg, elasticsearch, aiomysql
├── requirements-dev.txt     # Dev deps: pytest, pytest-asyncio
├── pytest.ini               # Pytest config (asyncio_mode = auto)
├── run_tests.fish           # Test runner (Fish): venv + deps + docker + tests
├── run_tests.sh             # Test runner (Bash): venv + deps + docker + tests
├── clients/                 # Backend client implementations
│   ├── base.py              # DBClient, DocumentClient, VectorClient, OLAPClient
│   ├── postgresql.py        # PostgresClient (SQLAlchemy async engine + raw SQL)
│   ├── elasticsearch.py     # ElasticsearchClient (AsyncElasticsearch + DSL translation)
│   ├── doris.py             # DorisClient (SQLAlchemy + aiomysql, MySQL protocol)
│   └── factory.py           # create_*_client / get_*_client factory functions
├── query/                   # Query building components
│   ├── builder.py           # QueryBuilder, Query, WhereCondition, WhereGroup dataclasses
│   ├── operators.py         # Eq, Gt, Gte, Lt, Lte, In, Like, ILike, NotEq (frozen dataclasses)
│   └── queryset.py          # QuerySet — result wrapper with all(), first(), one(), update(), delete()
├── schema/                  # Schema definition dataclasses + DDL/mapping generation
│   ├── builder.py           # SchemaBuilder, TableDef, ColumnDef, IndexDef, ForeignKeyDef
│   ├── sqlalchemy_gen.py    # table_to_sqlalchemy, table_to_sqlalchemy_doris, table_defs_to_metadata
│   ├── es_builder.py        # ESMappingBuilder, ESFieldDef, ESIndexDef, ESIndexTemplateDef
│   └── es_mapping_gen.py    # ESMappingGenerator, ESField — DomainModel → ESIndexDef bridge
├── docs/                    # Documentation
│   ├── implementation.md    # Implementation summary and changelog
│   └── testing.md           # Testing guide
├── examples/                # Example scripts
│   └── sql_preview_demo.py  # Demonstrates to_sql(), debug_query(), find_one/find_many
└── tests/                   # Test suite (58+ tests)
    ├── conftest.py          # Async fixtures for PG, ES, Doris
    ├── init.sql             # Test schema (users, posts) + sample data
    ├── test_query_builder.py
    ├── test_queryset.py
    ├── test_postgresql.py
    ├── test_elasticsearch.py
    ├── test_doris.py
    ├── test_factory.py
    ├── test_sql_preview.py
    ├── test_schema_builder.py
    ├── test_es_builder.py       # ESFieldDef, ESMappingBuilder — 28 tests
    └── test_es_mapping_gen.py   # ESMappingGenerator, ESField — 30 tests

Optional dependencies

Extra	Install command	Adds
elasticsearch	pip install db_core[elasticsearch]	ElasticsearchClient, create_elasticsearch_client
doris	pip install db_core[doris]	DorisClient, create_doris_client
(none)	pip install db_core	PostgresClient, QueryBuilder, QuerySet only

Query operators (internal)

All operators are frozen dataclasses in db_core.query.operators:

Operator	Signature	SQL equivalent	ES equivalent
Eq(value)	equality	col = value	{"term": {col: value}}
NotEq(value)	inequality	col != value	must_not + term
Gt(value)	greater than	col > value	{"range": {col: {"gt": value}}}
Gte(value)	greater or equal	col >= value	{"range": {col: {"gte": value}}}
Lt(value)	less than	col < value	{"range": {col: {"lt": value}}}
Lte(value)	less or equal	col <= value	{"range": {col: {"lte": value}}}
In(values)	IN list	col IN (v1, v2, ...)	{"terms": {col: values}}
Like(pattern)	LIKE	col LIKE pattern	{"wildcard": ...}
ILike(pattern)	ILIKE	col ILIKE pattern	(not natively supported)