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
• ensure_pg_database_exists — Create the target PostgreSQL database if missing (connects to the postgres maintenance DB)
• ensure_doris_database_exists — Create the target Doris database if missing (requires the [doris] extra)
• QueryBuilder — Fluent query builder
• QuerySet — Rich result wrapper with chainable operations
• SchemaBuilder — Pure SQL DDL generation (CREATE TABLE, CREATE INDEX, ALTER TABLE) from schema dataclasses
• 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
• IsNull, IsNotNull, RawExpr — Additional query operators (SQL backends only)

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, PG_PORT, PG_DATABASE (or PG_DB) plus PG_DRIVER — all must be set for URL assembly (no defaults are applied)

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 PG_DRIVER=asyncpg

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 — must be combined with ES_PORT and ES_SCHEME (all three are required together)

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_HOST, DORIS_PORT, DORIS_DATABASE (or DORIS_DB) plus DORIS_DRIVER — all required; DORIS_USER and DORIS_PASSWORD are optional

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")

# Tuple form — (operator, value)
.where("age", (">=", 18))
.where("role", ("IN", ["admin", "moderator"]))

# 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%")

# NULL checks
.where_null("deleted_at")                   # WHERE deleted_at IS NULL
.where_not_null("verified_at")              # WHERE verified_at IS NOT NULL

# Raw SQL expression on the value side (SQL backends only)
.where_raw("created_at", ">= UNIX_TIMESTAMP(DATE_SUB(NOW(), INTERVAL :days DAY))",
           params={"days": 7})

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")
)

# OR-connected group
db.table("users").where("age", 25).or_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)
)

# OR-connected nested group (Elasticsearch only)
es.table("articles").or_where_nested("comments", lambda q:
    q.where("comments.author", "Bob").or_where("comments.likes", ">", 20)
)

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

# JOINs and GROUP BY
.join("posts p", "p.user_id = users.id", join_type="LEFT")  # INNER, LEFT, RIGHT, FULL
.group_by("status")                                          # GROUP BY status

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)
has    = await query.exists()       # → bool

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)

Builder-level mutations

The QueryBuilder itself exposes mutation shortcuts bound to the table:

await db.table("users").create({"name": "Ann"}, returning=["id"])   # insert one row
await db.table("users").create_many([{"name": "A"}, {"name": "B"}]) # insert many rows
await db.table("users").save({"id": 1, "name": "Ann Updated"})      # upsert by id
await db.table("users").where("status", "inactive").update(status="active")  # update matches
await db.table("users").where("status", "banned").delete()          # delete matches
await db.table("users").where("status", "banned").remove()          # alias for delete()

update() / delete() flatten only top-level, non-OR Eq and In conditions (_extract_simple_conditions). OR conditions, groups, and other operators are silently skipped — use the client's update() / delete() directly for complex predicates.

Raw SQL execution (execute_raw)

DBClient declares execute_raw(sql, params=None, *, multi_statement=False):

• PostgreSQL — returns List[Dict] (SELECT), int (INSERT/UPDATE/DELETE rowcount), or None (DDL). multi_statement=True runs ;-delimited SQL over a raw asyncpg connection (bound params are only supported in single-statement mode).
• Doris — same semantics; multi_statement=True splits on ; and executes each statement sequentially over a raw aiomysql cursor.
• Elasticsearch — raises NotImplementedError (use the query builder, or client.client for native ES API calls).

rows = await db.execute_raw("SELECT * FROM users WHERE age > :age", {"age": 21})
await db.execute_raw(
    "CREATE TABLE IF NOT EXISTS t (id INT); INSERT INTO t VALUES (1)",
    multi_statement=True,
)

Gap detection (find_sequential_gaps)

Finds missing values in a sequential integer column using a LAG window function. Declared on DBClient (raises NotImplementedError by default) and implemented by DorisClient. Returns dicts containing the partition column plus gap_start / gap_end.

gaps = await doris.find_sequential_gaps(
    "telegram_messages",
    sequence_column="message_id",
    partition_column="peer_id",
    partition_value=123456789,
    min_sequence=1,
    max_sequence=500,
)
# → [{"peer_id": 123456789, "gap_start": 12, "gap_end": 17}, ...]

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

DBClient provides a full set of mutation methods. Elasticsearch implements bulk_update and bulk_delete (via update_by_query / delete_by_query on the id field), but row-level create / update / delete raise NotImplementedError.

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])

Pure-SQL DDL generation (SchemaBuilder)

SchemaBuilder turns the same dataclasses into ready-to-execute DDL strings. All methods are static, pure functions (no I/O) — run the result through db.execute_raw(sql):

from db_core.schema import SchemaBuilder, IndexDef, ColumnDef

SchemaBuilder.create_table(users)                     # CREATE TABLE IF NOT EXISTS ...
SchemaBuilder.create_index("users", IndexDef(["email"], unique=True))
SchemaBuilder.add_column("users", ColumnDef("bio", "TEXT"))
SchemaBuilder.drop_column("users", "bio")
SchemaBuilder.rename_table("users", "accounts")
SchemaBuilder.drop_table("users")                     # DROP TABLE IF EXISTS ...

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 (12 files, ~241 tests) includes:

• Query builder — 68 tests: select, where, OR, grouping, operators, pagination, builder mutations
• QuerySet — 7 tests for result manipulation
• PostgreSQL — 13 tests for connection, CRUD, EXPLAIN, pooling
• SQL preview — 12 tests for to_sql(), debug_query(), raw_query()
• Elasticsearch — 18 tests for search body generation, nested queries, explain
• Apache Doris — 15 tests for MySQL-protocol execution, bulk update/delete
• Schema builder — 25 tests for TableDef, ColumnDef, IndexDef, FK defs, DDL generation
• ES schema builder — 28 tests for ESFieldDef, ESMappingBuilder, index templates
• ES mapping gen — 30 tests for ESMappingGenerator, ESField overrides, nested models
• Execute raw — 8 tests for execute_raw(), including multi-statement
• Timestamps — 14 tests for auto-timestamp injection
• Factory — 3 tests for env var resolution, URL construction

./run_tests.sh and ./run_tests.fish accept backend filters — --pg, --elastic (or --es), --doris — mapped to the pytest markers pg, elastic, and doris defined in pytest.ini.

Project structure

db_core/
├── __init__.py              # Package exports (clients, factories, QueryBuilder, QuerySet, schema builders)
├── pyproject.toml           # Build config with optional deps (elasticsearch, doris)
├── requirements.txt         # Flat dev install (all backends, unpinned)
├── requirements-dev.txt     # Dev deps: pytest, pytest-asyncio
├── pytest.ini               # Pytest config (asyncio_mode = auto, markers pg/elastic/doris)
├── 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
│   ├── README.md            # Docs index
│   ├── 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 (~241 tests across 12 files)
│   ├── conftest.py          # Async fixtures for PG, ES, Doris (+ autouse per-test PG reseed)
│   ├── init.sql             # Test schema (users, posts) + sample data
│   ├── test_query_builder.py    # 68 tests
│   ├── test_queryset.py         # 7 tests
│   ├── test_postgresql.py       # 13 tests
│   ├── test_elasticsearch.py    # 18 tests
│   ├── test_doris.py            # 15 tests
│   ├── test_factory.py          # 3 tests
│   ├── test_sql_preview.py      # 12 tests
│   ├── test_schema_builder.py   # 25 tests
│   ├── test_es_builder.py       # 28 tests
│   ├── test_es_mapping_gen.py   # 30 tests
│   ├── test_execute_raw.py      # 8 tests
│   └── test_timestamps.py       # 14 tests
└── tests_unit/              # Legacy duplicate test_schema_builder.py (not run by the runners)

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

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 (PG) / LIKE (Doris)	{"wildcard": {col: {"value": ..., "case_insensitive": true}}}
IsNull()	IS NULL	col IS NULL	not supported
IsNotNull()	IS NOT NULL	col IS NOT NULL	not supported
RawExpr(sql, params)	raw expression	col <sql> (with :param bindings)	not supported