Skip to main content

echoss AI Bigdata Solution - Database Query Package

Project description

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.yamlconfig/config.yaml로 복사해 실제 값을 채우세요 (실 config는 gitignore 대상). config/config.yaml 기준의 credential 제거 예시는 아래와 같습니다.

mysql:
  user: <MYSQL_USER>
  passwd: <MYSQL_PASSWORD>
  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_PASSWORD>
  host: <ELASTIC_HOST>
  port: <ELASTIC_PORT>
  scheme: <http|https>
  verify_certs: <true|false>

postgres:
  user: <POSTGRES_USER>
  passwd: <POSTGRES_PASSWORD>
  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)

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()의 차원은 모델에서 자동 추론합니다.
  1. fastembed_model 설정
  • 지정한 모델로 동작합니다.
  • create_collection()의 차원은 설정 모델 기준으로 자동 추론합니다.
  1. 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

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

echoss_db-2.2.0.tar.gz (66.8 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

echoss_db-2.2.0-py3-none-any.whl (45.0 kB view details)

Uploaded Python 3

File details

Details for the file echoss_db-2.2.0.tar.gz.

File metadata

  • Download URL: echoss_db-2.2.0.tar.gz
  • Upload date:
  • Size: 66.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.13

File hashes

Hashes for echoss_db-2.2.0.tar.gz
Algorithm Hash digest
SHA256 3f46a8fbef3caf2aaac3236643312c3f25c78b3b0d14ab4c495cf9e8dd78486a
MD5 06148358c2080d08a228d1f1484db33c
BLAKE2b-256 3aafdd463888f6fe2bde2f25d9a862e463de675e9441ffd9dcfba43b847f8c3e

See more details on using hashes here.

File details

Details for the file echoss_db-2.2.0-py3-none-any.whl.

File metadata

  • Download URL: echoss_db-2.2.0-py3-none-any.whl
  • Upload date:
  • Size: 45.0 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.13

File hashes

Hashes for echoss_db-2.2.0-py3-none-any.whl
Algorithm Hash digest
SHA256 3a794d87f3cce47b5e251ae2573959f7137aeec91d41f0278694c62e6600f8de
MD5 bcd5ce587426502cce28839f01ccd7c7
BLAKE2b-256 ae71c65c998d85aa798a11f7347bfcec8a5cd91bf8f01a8e169a08c211bc3379

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page