echoss_db

MySQL, PostgreSQL, OpenSearch, Qdrant compatible query/vector access package

⚠️ MongoDB support is deprecated (v2.2.0) and will be removed in v3.0.0. pymongo is now an optional dependency: pip install echoss-db[mongo].

Prepare

사용 전 config(인증 정보) 유무를 확인한 뒤 사용해야 합니다. config/config.example.yaml을 config/config.yaml로 복사해 실제 값을 채우세요 (실 config는 gitignore 대상). config/config.yaml 기준의 credential 제거 예시는 아래와 같습니다.

credential 필드(passwd, api_key, read_only_api_key)는 평문 대신 ${VAR_NAME} 형태로 적으면 .env(또는 OS 환경변수)에서 값을 읽어 채웁니다. .env는 git에 추적되지 않으므로 .env.example을 .env로 복사해 실제 값을 채우세요. host/port 등 credential이 아닌 필드는 지금처럼 평문 값을 직접 씁니다.

mysql:
  user: <MYSQL_USER>
  passwd: ${MYSQL_PASSWD}
  host: <MYSQL_HOST>
  port: <MYSQL_PORT>
  db: <MYSQL_DB>
  charset: utf8mb4

mongo:
  host: <MONGO_HOST>
  port: <MONGO_PORT>
  db: <MONGO_DB>

elastic:
  user: <ELASTIC_USER>
  passwd: ${ELASTIC_PASSWD}
  host: <ELASTIC_HOST>
  port: <ELASTIC_PORT>
  scheme: <http|https>
  verify_certs: <true|false>

postgres:
  user: <POSTGRES_USER>
  passwd: ${POSTGRES_PASSWD}
  host: <POSTGRES_HOST>
  port: <POSTGRES_PORT>
  db: <POSTGRES_DB>
  schema: <POSTGRES_SCHEMA>

qdrant:
  host: <QDRANT_HOST>
  port: <QDRANT_PORT>
  scheme: <http|https>
  timeout: <SECONDS>
  collection: <QDRANT_COLLECTION>
  default_limit: <DEFAULT_LIMIT>
  # fastembed_model omitted -> default FastEmbed model automatically applied
  # fastembed_model: null -> external vector mode (no FastEmbed)
  api_key: ${QDRANT_API_KEY}                  # optional, admin/전체 권한
  read_only_api_key: ${QDRANT_READ_ONLY_API_KEY}  # optional, 조회 전용

.env 예시(.env.example 참고, 실값은 커밋하지 않음):

MYSQL_PASSWD=
ELASTIC_PASSWD=
POSTGRES_PASSWD=
QDRANT_API_KEY=
QDRANT_READ_ONLY_API_KEY=

Qdrant는 role로 admin/조회 인스턴스를 나눠 생성할 수 있습니다:

from echoss_db.qdrant_vector import QdrantVector

admin = QdrantVector("config/config.yaml")                    # role 기본값 "admin" -> qdrant.api_key 사용
readonly = QdrantVector("config/config.yaml", role="readonly")  # qdrant.read_only_api_key 사용

api_key/read_only_api_key를 둘 다 설정하지 않으면 기존과 동일하게 무인증으로 연결됩니다.

Installation

Version Line Policy

• echoss-db 1.x: legacy line for Python 3.9 / 3.10
• echoss-db 2.x: current line for Python >=3.12 (last release supporting 3.11: 2.1.0)
• company standard runtime for 2.x: Python 3.12

Recommended local environments:

• 1.x development / maintenance: echoss_dev
• 2.x standard development / integration: use a Python 3.12 environment

This package currently documents the 2.x line. Python 3.12 is the company standard target runtime. Python 3.11 or lower is not supported in echoss-db>=2.2.0; use the 1.x line for legacy Python 3.9 / 3.10 environments.

    pip install -U echoss-db

If your environment previously used legacy PostgreSQL drivers, clean them first:

pip uninstall -y psycopg2 psycopg2-binary
pip install -U "psycopg[binary]>=3.2,<4.0.0"

If you already have an older Pillow in a shared environment, upgrade it before installing:

pip install -U "Pillow>=10.3.0,<12.0"

Recommended fresh environment examples:

conda create -n echoss_dev_py312 python=3.12 -y
conda activate echoss_dev_py312
pip install -U echoss-db

Repository-local conda env convention:

# 1.x legacy line
CONDA_NO_PLUGINS=true conda run -n echoss_dev python --version

# 2.x standard runtime
python3.12 --version

For deployment, prefer a dedicated service environment instead of reusing a mixed notebook/UI environment. Shared environments can keep stale transitive packages such as Pillow or urllib3 and hide resolver problems until runtime.

Optional mapping extras (pydantic):

    pip install -U "echoss-db[mapping]"

Quick Start

Import package and class

from echoss_db import MysqlQuery, PostgresQuery, MongoQuery, ElasticSearch, QdrantVector

with MysqlQuery('CONFIG_FILE_PATH' or dict) as mysql:
    rows = mysql.select_list("SELECT * FROM users WHERE status=%s", ("active",))

with PostgresQuery('CONFIG_FILE_PATH' or dict) as postgres:
    docs = postgres.select_list("SELECT doc_id, content FROM kb_document LIMIT %s", (10,))

with MongoQuery('CONFIG_FILE_PATH' or dict) as mongo:
    items = mongo.select_list("sample_collection", {"status": "active"})

with ElasticSearch('CONFIG_FILE_PATH' or dict) as elastic:
    hits = elastic.search_list(body={"query": {"match_all": {}}}, fetch_all=False)

with QdrantVector('CONFIG_FILE_PATH' or dict) as qdrant:
    hits = qdrant.search_text(query_text="RAG", limit=5)

Lifecycle And Transactions

Client-level context manager is recommended for routine usage. It closes the client on block exit, but it does not define a transaction boundary.

Direct instantiation also remains valid:

mysql = MysqlQuery('CONFIG_FILE_PATH' or dict)
rows = mysql.select_list("SELECT * FROM users")
mysql.close()

For batch jobs or delayed commit flows, use the explicit SQL transaction API:

with MysqlQuery('CONFIG_FILE_PATH' or dict) as mysql:
    with mysql.transaction() as tx:
        tx.insert("INSERT INTO users(name) VALUES (%s)", ("kim",))
        tx.update("UPDATE users SET status=%s WHERE name=%s", ("active", "kim"))

with PostgresQuery('CONFIG_FILE_PATH' or dict) as postgres:
    with postgres.transaction() as tx:
        tx.insert(
            "INSERT INTO kb_document(source_type) VALUES (%s) RETURNING doc_id",
            ("tutorial",),
            return_lastrowid=True,
        )

Transaction API semantics:

• transaction() is available only on MysqlQuery and PostgresQuery
• transaction-scoped methods are fail-fast and propagate exceptions
• if an exception leaves the with ... transaction() block, rollback is executed automatically
• if the block exits normally, commit is executed automatically
• if tx.commit() is called manually, the current transaction unit is committed and a new transaction unit starts on the same connection
• if tx.rollback() is called manually, the current transaction unit is rolled back and a new transaction unit starts on the same connection

This behavior is intentionally stricter than the legacy convenience methods such as insert() / update() / delete(), which often log and return fallback values instead of raising.

MySQL

# CREATE
mysql.create('QUERY_STRING')

# DROP
mysql.drop('QUERY_STRING')

# TRUNCATE
mysql.truncate('QUERY_STRING')

# ALTER
mysql.alter('QUERY_STRING')

# SELECT
mysql.select('QUERY_STRING', params=None) 
mysql.select_one('QUERY_STRING', params=None)
mysql.select_list('QUERY_STRING', params=None)
mysql.faster_select('QUERY_STRING', params=None)

# INSERT without params
mysql.insert('QUERY_STRING', params=None)
# INSERT with tuple
mysql.insert('QUERY_STRING', params) 
# INSERT with list[tuple]
mysql.insert('QUERY_STRING', params_list) 

# UPDATE
mysql.update('QUERY_STRING', params=None) 

# DELETE
mysql.delete('QUERY_STRING', params=None)

# show Database
mysql.databases()

# show Tables
mysql.tables()

# Ping
mysql.ping()

# Close
# crash process close
mysql.close()

# debug query : default True
mysql.query_debug(False)

PostgreSQL

# CREATE
postgres.create('QUERY_STRING')

# DROP
postgres.drop('QUERY_STRING')

# TRUNCATE
postgres.truncate('QUERY_STRING')

# ALTER
postgres.alter('QUERY_STRING')

# SELECT
postgres.select('QUERY_STRING', params=None) 
postgres.select_one('QUERY_STRING', params=None) 
postgres.select_list('QUERY_STRING', params=None) 
postgres.faster_select('QUERY_STRING', params=None, fetch_size=1000) 
postgres.faster_select_generator('QUERY_STRING', params=None, fetch_size=1000) 

# SELECT with JSONB normalization
# JSONB 컬럼은 드라이버/환경에 따라 str 또는 dict로 반환됩니다(반환 타입 비결정적).
# parse_json으로 항상 dict/list로 정규화하여 호출부의 방어적 파싱을 제거합니다.
postgres.select('SELECT id, detail_json FROM t', parse_json=['detail_json'])  # 지정 컬럼만
postgres.select('SELECT * FROM t', parse_json=True)   # 모든 컬럼의 JSON-like str 시도
# 기본 parse_json=False는 기존 동작과 동일(하위호환)
# transaction 경로도 동일 지원: tx.select(..., parse_json=[...])

# INSERT
postgres.insert('QUERY_STRING', params=None) 
postgres.insert('QUERY_STRING', params=None, return_lastrowid=True)  

# UPSERT (INSERT ... ON CONFLICT) — ON CONFLICT 절 자동 생성
# 단일 dict 또는 list[dict](배치, executemany). 식별자는 자동 quoting, 값은 파라미터 바인딩.
postgres.upsert(
    'enrichment.product_detail_records',
    {"mall_product_no": 123, "status": "ok", "updated_at": None},
    conflict_keys=['mall_product_no'],     # 복합키도 가능: ['a','b','c','d']
    update_columns=None,                   # None이면 conflict_keys/auto_now 제외 전체
    auto_now=['updated_at'],               # NOW()로 채움(INSERT/UPDATE 양쪽)
)  # 반환: 영향 rowcount

# UPDATE
postgres.update('QUERY_STRING', params=None)  

# DELETE
postgres.delete('QUERY_STRING', params=None) 

# show Databases
postgres.databases()

# show Tables
postgres.tables()

# Current schema
postgres.current_schema()

# Ping
postgres.ping()

# Close
postgres.close()

# debug query : default True
postgres.query_debug(False)

MongoDB (Deprecated)

⚠️ Deprecated since v2.2.0, removal in v3.0.0. Requires pip install echoss-db[mongo].

# show Database
mongo.databases()

# show Collections
mongo.collections()

# Ping
mongo.ping()

# SELECT
mongo.select('COLLECTION_NAME','QUERY_STRING or DICTIONARY') 

# INSERT
mongo.insert('COLLECTION_NAME','QUERY_STRING or DICTIONARY')
mongo.insert_many('COLLECTION_NAME','QUERY_STRING or DICTIONARY')

# UPDATE
mongo.update('COLLECTION_NAME','FILTER_STRING or DICTIONARY', 'UPDATE_STRING or DICTIONARY')
mongo.update_many('COLLECTION_NAME','FILTER_STRING or DICTIONARY', 'UPDATE_STRING or DICTIONARY')

# DELETE
mongo.delete('COLLECTION_NAME','QUERY_STRING or DICTIONARY')
mongo.delete_many('COLLECTION_NAME','QUERY_STRING or DICTIONARY')

ElasticSearch

# CREATE
elastic.index(index='INDEX_NAME')

# DROP
elastic.delete_index(index='INDEX_NAME')

# SELECT
elastic.search(body=query) 
elastic.search_list(body=query, fetch_all=True)
elastic.search_dataframe(body=query, fetch_all=True)
elastic.search_field(field='FIELD_NAME',value='VALUE')

# INSERT
elastic.index(index='INDEX_NAME', body='JSON_BODY', id='ID')

# UPDATE
elastic.update(id='ID', body='JSON_BODY')

# DELETE
elastic.delete(id='ID')

# SCROLL
elastic.prepare_scroll(
    index_name='farmers_index',
    query={'match_all': {}},
    source_fields=['name', 'code'],
    size=10000,
    scroll_ttl='2m'
)
chunk_list = elastic.next_scroll_chunk()

# BULK
success, error_list = elastic.bulk_insert('farm_index', doc_list, id_field='farm_id')
success, error_list = elastic.bulk_upsert('farm_index', doc_list, id_field='farm_id')

# Ping
elastic.ping()

# Connection Information
elastic.info()

Qdrant

QdrantVector는 기본적으로 FastEmbed 모델을 사용합니다. 기본 설치는 qdrant-client[fastembed]를 포함하며, Python 3.11+ 환경을 전제로 합니다.

Manual test 실행 순서:

1. tests/manual/example_postgres_qdrant_ingest.py (테스트용 충분한 chunk 데이터 생성)

2. tests/manual/example_qdrant.py (Qdrant 업서트/검색 확인)

3. fastembed_model 미설정

• 기본값 sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2(384차원)가 자동 적용됩니다.
• 이때 초기화 시 경고 로그가 출력됩니다. 외부에서 계산한 벡터(예: 1024차원)를 직접 넣으려면 fastembed_model: null로 설정해야 하며, 미설정 상태로 upsert_vectors()를 호출하면 차원 불일치 오류가 납니다.
• 설치/runtime 문제가 있으면 fallback 없이 초기화 단계에서 예외가 발생합니다.
• 모델 상세(설명 포함) 목록: list_supported_models_by_class("TextEmbedding")
• vector 입력 없이 원문 텍스트(document/text/content)를 전달합니다.
• 명시 메서드 upsert_texts() / search_text() 사용을 권장합니다.
• create_collection()의 차원은 모델에서 자동 추론합니다.

2. fastembed_model 설정

• 지정한 모델로 동작합니다.
• create_collection()의 차원은 설정 모델 기준으로 자동 추론합니다.

3. fastembed_model: null

• FastEmbed를 명시적으로 비활성화하고 벡터 직접 주입 모드로 동작합니다.
• 이 경우 upsert_vectors() / search_vector() 사용을 권장합니다.
• create_collection(dim=...)에서 dim은 필수입니다.

Config example:

qdrant:
  host: <QDRANT_HOST>
  port: <QDRANT_PORT>
  scheme: <http|https>
  collection: <QDRANT_COLLECTION>
  timeout: <SECONDS>
  default_limit: <DEFAULT_LIMIT>
  fastembed_model: sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2

Text-input upsert example (fastembed_model enabled):

points = [
    {"id": 1, "document": "RAG 시스템 설계 문서", "payload": {"source": "wiki"}},
    {"id": 2, "text": "Qdrant 검색 예제", "payload": {"source": "blog"}},
]
qdrant.upsert_texts(points, collection='ai_rag_chunks')

hits = qdrant.search_text(query_text='RAG 아키텍처', limit=5, collection='ai_rag_chunks')

External vector-input mode is documented as a code pattern only (for lightweight tutorial runtime):

qdrant = QdrantVector({
    'qdrant': {
        'host': '<host>',
        'port': 6333,
        'scheme': 'http',
        'collection': 'ai_rag_chunks_external',
        'fastembed_model': None,
    }
})

points = [{"id": 1, "vector": your_embed_fn("문서"), "payload": {"source": "external"}}]
qdrant.create_collection(dim=len(points[0]['vector']))
qdrant.upsert_vectors(points)
hits = qdrant.search_vector(vector=your_embed_fn('질의'))

Payload filter on search (외부 벡터 + 조건 검색):

# dict 필터: 값이 list면 MatchAny, scalar면 MatchValue로 자동 변환
hits = qdrant.search_vector(
    vector=your_embed_fn('질의'),
    filter_={"doc_type": ["summary_long", "faq_question"],  # MatchAny
             "mall_product_no": 123},                         # MatchValue
    limit=10,
)

# 고급 조건은 qdrant_client Filter 객체를 그대로 전달 (should/must_not/range 등)
import qdrant_client.http.models as qm
hits = qdrant.search_vector(vector=v, filter_=qm.Filter(must_not=[...]))

# 빌더만 단독 사용도 가능
qfilter = qdrant.build_filter({"doc_type": ["faq_question"]})

주의: 구버전(1.2.x)은 access_tags_any 외의 dict 키를 무시했습니다. 2.2.0부터 일반 dict 키가 실제 필터로 적용되므로, 기존에 무시되던 dict를 넘기던 호출은 결과가 달라질 수 있습니다(의도된 동작 변경).

Collection metadata / management:

info = qdrant.get_collection_info('ai_rag_chunks')
# {"name": ..., "vectors": {"default": {"size": 1024, "distance": "Cosine"}},
#  "points_count": 3300, "status": "green"}

# 차원 불일치 시 재생성 마이그레이션을 wrapper 안에서 완결
if info["vectors"]["default"]["size"] != 1024:
    qdrant.delete_collection('ai_rag_chunks')
    qdrant.create_collection(dim=1024, collection='ai_rag_chunks')

Large-dimension upsert robustness (retry / batching):

# 대형 차원(예: 1024d) 대량 적재 시 timeout 권장값: 30초 이상
# (config의 qdrant.timeout). 실패에 대비해 retry/backoff, 대량은 batch_size 사용.
qdrant.upsert_vectors(
    points,                # 수천 건
    retry=3, backoff=0.5,  # 실패 시 0.5, 1.0, 2.0초 백오프로 재시도
    batch_size=256,        # 256개씩 청크 전송
    wait=True,
)

Mapping Lite (Optional)

echoss_db.mapping은 ORM이 아닌 경량 row/model 매핑 기능입니다.

from dataclasses import dataclass
from echoss_db.mapping import map_row, map_rows, to_params

@dataclass
class UserDTO:
    id: int
    name: str
    age: int = 0

row = {"id": 1, "name": "kim"}
dto = map_row(row, UserDTO)
dto_list = map_rows([row], UserDTO)
params = to_params(dto)

• 기본 설치: dataclass 지원
• 선택 설치(echoss-db[mapping]): pydantic 모델 지원
• 튜토리얼: tests/manual/example_mapping_mysql.py

Code Quality

When creating new functions, please follow the Google style Python docstrings. See example below:

def example_function(param1: int, param2: str) -> bool:
    """Example function that does something.

    Args:
        param1: The first parameter.
        param2: The second parameter.

    Returns:
        The return value. True for success, False otherwise.

    """

Version history

v0.1.0 initial version v0.1.1 echoss_logger include v0.1.7 mysql support query with params. return cursor.rowcount for insert/update/delete query v1.0.0 mysql support query with list params v1.0.1 elastic search_list fetch_all option, mongo support insert_many, delete_many, update_many method v1.0.2 mysql reuse cursor v1.0.7 echoss-query last version v1.0.8 change package name to echoss-db v1.0.11 update() check 'doc' or 'script' in body v1.1.0 add Scroll, Bulk functions in ElasticSearch v1.1.5 improve prepare_scroll and dataframe conversion logic, add request_timeout in bulk operation v1.2.0 change mysql base package to sqlalchemy (previous version use pymysql) v1.2.1 add return_lastrowid parameter in insert() method v1.2.2 return dictionary select result v1.2.3 add postgres DB and qdrant vector storage support v2.0.0 add optional echoss_db.mapping (dataclass default, pydantic optional) v2.1.0 raise Python baseline to 3.11, standardize on psycopg3, and keep Qdrant fastembed as default runtime v2.2.0 qdrant generalized dict filter (breaking behavior: previously ignored keys now applied) and get_collection_info, postgres upsert/parse_json options, upsert retry/backoff/batch_size, deprecate MongoQuery (pymongo moved to optional extra [mongo]), raise Python floor to 3.12 v2.2.2 qdrant admin/read-only API key support (role param, api_key/read_only_api_key config fields), config.yaml credential fields (mysql/postgres/elastic passwd, qdrant api_key/read_only_api_key) now accept ${VAR_NAME} placeholders resolved from .env/OS env (new python-dotenv dependency)