echoss-db 1.2.3

echoss AI Bigdata Solution - Database Query Package

echoss_db

---

MySQL, MongoDB, Opensearch compatible query access package

Version History:

• 1.1.0 : Add Scroll, Bulk functions in ElasticSearch
• 1.2.0 : change mysql base package to sqlalchemy (previous version use pymysql)

Prepare

사용 전 config(인증 정보) 의 유무를 확인한 뒤 사용해야한다. config file example:

mysql:
    user : <user_id>
    passwd : <pass code>
    host : <IP addrress or domain>
    port : :<network port number>
    db : <schema name>
    charset : utf8mb4

mongo:
    host : <IP addrress or domain>
    port :<network port number>
    db : <database name>

elastic:
    user : <user_id>
    passwd : <pass code>
    host : <IP addrress or domain>
    port : :<network port number>
    scheme : <http or https>

Installaion

---

To install this package, please use Python 3.8 or higher.

    pip install -U echoss-db

Quick Start

Import package and class

    from echoss_db import MysqlQuery, MongoQuery, ElasticSearch

    mysql = MysqlQuery('CONFIG_FILE_PATH' or dict)
    mongo = MongoQuery('CONFIG_FILE_PATH' or dict)
    elastic = ElasticSearch('CONFIG_FILE_PATH' or dict)

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) -> dataframe
    mysql.select_one('QUERY_STRING', params=None) -> dict
    mysql.select_list('QUERY_STRING', params=None) -> list(dict)
    mysql.faster_select('QUERY_STRING', params=None) -> dataframe

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

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

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

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

MongoDB

---

    # show Database
    mongo.databases()

    # show Collections
    mongo.collections()

    # Ping
    mongo.ping()

    # SELECT
    mongo.select('COLLECTION_NAME','QUERY_STRING or DICTIONARY') -> pd.Dataframe

    # 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) -> any
    elastic.search_list(body=query, fetch_all=True) -> list
    elastic.search_dataframe(body=query, fetch_all=True) -> dataframe
    elastic.search_field(field='FIELD_NAME',value='VALUE') -> list

    # 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("farmers_index", query={"query": {"match_all": {}}}, source_filters=["name", "code"])
    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 모델을 사용합니다.

Tutorial 실행 순서:

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

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

3. fastembed_model 미설정

• 기본값 sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2가 자동 적용됩니다.
• 단, 설치된 qdrant-client/fastembed 조합에서 미지원일 수 있으므로 아래 메서드로 확인하세요.
  • 모델 상세(설명 포함) 목록: 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: <IP addrress or domain>
  port: 6333
  scheme: http
  collection: ai_rag_chunks
  timeout: 5
  default_limit: 10
  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"}},
]
qv.upsert_texts(points, collection="ai_rag_chunks")

hits = qv.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):

qv = 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"}}]
qv.create_collection(dim=len(points[0]["vector"]))
qv.upsert_vectors(points)
hits = qv.search_vector(vector=your_embed_fn("질의"))

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