Skip to content
Machbase DBMS
개발도구 및 라이브러리
Python

Python

개요

이 문서는 2.3 패키지를 기준으로 정리했습니다. PyPI 패키지명은 machbaseapi(소문자)이고 구현은 순수 Python입니다(네이티브 .so/.dll/.dylib 불필요).

기존 machbase 사용 흐름은 유지됩니다.

  • 패키지 설치명: machbaseapi
  • 기존과 동일하게 import machbaseAPI 사용
  • DB-API 방식 connect(), cursor() 지원
  • append*는 on_ack 콜백을 추가할 수 있어 ACK 관찰 가능
  • append(), appendByTime(), appendData(), appendDataByTime()는 타입 리스트를 생략해도 동작합니다. 서버 메타데이터 기반으로 타입을 자동 추론합니다.
  • 2.3부터 append row의 마지막 일부 컬럼을 생략하면 append null-bit를 통해 NULL로 저장합니다.
  • TAG 테이블은 value 컬럼까지 필수이며, 이후 추가 컬럼과 metadata 컬럼은 생략 시 NULL로 저장할 수 있습니다.
  • 커넥션 풀 옵션(pool_name, pool_size, pool_reset_session) 미지원

아래 예제는 기존 machbase 클래스 기반 레거시 스크립트를 대상으로 합니다.

설치

요구 사항

  • pip을 사용할 수 있는 Python 3.8 이상
  • 접속 가능한 Machbase 서버와 계정 정보(기본 계정 SYS/MANAGER, 포트 5656)
  • 2.3은 네이티브 라이브러리 의존성이 없습니다.

PyPI에서 설치 

pip3 install machbaseapi

pip3가 PATH에 없다면 python3 -m pip install machbaseapi 명령을 사용합니다.

모듈 확인 

python3 - <<'PY'
from machbaseAPI import machbase, connect
print('machbase 클래스 import:', bool(machbase))
print('connect 함수 존재:', callable(connect))
print('module import:', __import__('machbaseAPI'))
PY

위 명령이 성공하면 패키지를 정상적으로 import하고 인스턴스를 생성할 수 있음을 의미합니다.

DB-API 방식 샘플도 바로 사용할 수 있습니다.

from machbaseAPI import connect

conn = connect(host='127.0.0.1', port=5656, user='SYS', password='MANAGER')
cur = conn.cursor()
cur.execute('SELECT * FROM m$tables LIMIT 1')
print(cur.fetchall())
conn.close()

빠르게 시작하기

아래 스니펫은 로컬 서버에 접속해 샘플 테이블을 만들고, 데이터를 삽입한 뒤 조회하고 세션을 종료하는 흐름을 보여줍니다.

#!/usr/bin/env python3
import json
from machbaseAPI import machbase

def main():
    db = machbase()
    if db.open('127.0.0.1', 'SYS', 'MANAGER', 5656) == 0:
        raise SystemExit(db.result())

    try:
        rc = db.execute('drop table py_sample')
        print('drop table rc:', rc)
        print('drop table result:', db.result())

        ddl = (
            "create table py_sample ("
            "ts datetime,"
            "device varchar(40),"
            "value double"
            ")"
        )
        if db.execute(ddl) == 0:
            raise SystemExit(db.result())
        print('create table result:', db.result())

        for seq in range(3):
            sql = (
                "insert into py_sample values ("
                f"to_date('2024-01-0{seq+1}','YYYY-MM-DD'),"
                f"'sensor-{seq}',"
                f"{20.5 + seq}"
                ")"
            )
            if db.execute(sql) == 0:
                raise SystemExit(db.result())
            print('insert result:', db.result())

        if db.select('select * from py_sample order by ts') == 0:
            raise SystemExit(db.result())

        while True:
            rc, payload = db.fetch()
            if rc == 0:
                break
            row = json.loads(payload)
            print('row:', row)

        db.selectClose()
    finally:
        if db.close() == 0:
            raise SystemExit(db.result())

if __name__ == '__main__':
    main()

결과 처리

대다수 machbase 메서드는 성공 시 1, 실패 시 0을 반환합니다. 호출 직후 db.result()를 사용하면 서버에서 반환한 JSON 형식의 페이로드를 확인할 수 있습니다. select() 결과를 순회할 때는 (0, None)가 반환될 때까지 db.fetch()를 반복 호출하고, 마지막에 db.selectClose()로 리소스를 해제합니다.

지원 API 매트릭스

클래스API설명반환
machbaseopen(host, user, password, port)기본 계정과 포트로 Machbase 서버에 연결합니다.성공 시 1, 실패 시 0
machbaseopenEx(host, user, password, port, conn_str)추가 연결 문자열 속성을 사용해 확장 연결을 수행합니다.1 또는 0
machbaseclose()현재 세션을 종료합니다.1 또는 0
machbaseisOpened()핸들이 열려 있는지 확인합니다.1 또는 0
machbaseisConnected()서버와의 연결 상태를 확인합니다.1 또는 0
machbaseexecute(sql, type=0)UPDATE를 제외한 SQL 문을 실행합니다.1 또는 0
machbaseschema(sql)스키마 관련 명령을 실행합니다.1 또는 0
machbasetables()모든 테이블의 메타데이터를 조회합니다.1 또는 0
machbasecolumns(table_name)특정 테이블의 컬럼 메타데이터를 조회합니다.1 또는 0
machbasecolumn(table_name)저수준 카탈로그 호출로 컬럼 레이아웃을 가져옵니다.1 또는 0
machbasestatistics(table_name, user='SYS')CLI를 통해 테이블 통계를 요청합니다.1 또는 0
machbaseselect(sql)스트리밍 SELECT 또는 DESC를 실행합니다.1 또는 0
machbasefetch()select() 호출 이후 다음 행을 가져옵니다.(rc, json_str)
machbaseselectClose()열린 결과 집합 커서를 닫습니다.1 또는 0
machbaseresult()최신 JSON 페이로드를 반환합니다.JSON 문자열
machbaseappendOpen(table_name, types=None)컬럼 타입 코드를 지정하여 Append 프로토콜을 시작합니다. 생략 시 서버 메타데이터로 타입을 사용할 수 있습니다.1 또는 0
machbaseappendData(table_name, aTypes=None, values=None, format='YYYY-MM-DD HH24:MI:SS', on_ack=None)활성 Append 세션으로 행을 추가합니다. 2.1 이후에는 aTypes를 생략하고 두 번째 인자로 행을 바로 전달할 수 있습니다. 호출 시 데이터 패킷을 즉시 전송합니다.1 또는 0
machbaseappendDataByTime(table_name, aTypes=None, values=None, format='YYYY-MM-DD HH24:MI:SS', times=None, on_ack=None)명시적 타임스탬프로 행을 추가합니다. 2.1 이후에는 aTypes를 생략하고 두 번째 인자로 행을 바로 전달할 수 있습니다. 호출 시 데이터 패킷을 즉시 전송합니다.1 또는 0
machbaseappendFlush()이미 전송된 Append 데이터의 pending response를 확인하는 동기화 지점입니다. 전송 지연 버퍼를 비우는 API가 아닙니다.1 또는 0
machbaseappendClose()Append 세션을 종료합니다.1 또는 0
machbaseappend(table_name, aTypes=None, aValues=None, format='YYYY-MM-DD HH24:MI:SS')열기·추가·닫기를 한 번에 처리하는 편의 함수입니다. 2.1 이후에는 aTypes를 생략하고 aValues를 두 번째 인자로 바로 전달할 수 있습니다.1 또는 0
machbaseappendByTime(table_name, aTypes=None, aValues=None, format='YYYY-MM-DD HH24:MI:SS', times=None)타임스탬프 인지 Append를 위한 편의 함수입니다. 2.1 이후에는 aTypes를 생략하고 aValues를 두 번째 인자로 바로 전달할 수 있습니다.1 또는 0

DB-API 스타일 API (2.3)

API설명반환
connect(host, port, user, password, ...)DB-API 연결 생성MachbaseConnection
cursor(dictionary=True)커서 생성 (True: dict, False: tuple)MachbaseCursor
cursor.execute(sql, params=None)SQL 실행cursor
cursor.fetchone()한 건 조회`tuple
cursor.fetchmany(size)최대 size건 조회list
cursor.fetchall()전체 조회list
cursor.close()커서 종료None
cursor.rowcount영향 행 수int
connection.append(table, rows, types=None, times=None, strict=False)Append 프로토콜로 row를 추가합니다. 2.3부터 trailing 컬럼 생략 시 NULL padding을 적용합니다.입력 row 수

2.3 append 타입 생략과 trailing NULL padding (권장)

append()와 appendByTime()는 타입 리스트를 생략하고 호출할 수 있습니다. 두 번째 인자로 행 집합을 그대로 전달하면 서버 메타데이터 기반으로 처리합니다. 2.3부터는 입력 row의 마지막 일부 컬럼을 생략할 수 있고, 생략된 컬럼은 append null-bit를 통해 NULL로 저장됩니다.

#!/usr/bin/env python3
from machbaseAPI.machbaseAPI import machbase

def main():
    db = machbase()
    if db.open('127.0.0.1', 'SYS', 'MANAGER', 5656) == 0:
        raise SystemExit(db.result())

    try:
        db.execute('drop table py_append_auto')
        db.result()
        ddl = 'create table py_append_auto(ts datetime, tag varchar(16), reading double)'
        if db.execute(ddl) == 0:
            raise SystemExit(db.result())
        db.result()

        rows = [
            ['2024-01-01 10:00:00', 'node-1', 30.0],
            ['2024-01-01 10:01:00', 'node-1', 30.5],
        ]
        if db.append('PY_APPEND_AUTO', rows) == 0:
            raise SystemExit(db.result())
        print('append without types result:', db.result())
    finally:
        if db.close() == 0:
            raise SystemExit(db.result())

if __name__ == '__main__':
    main()

DB-API append trailing NULL 예제

connect().append()도 같은 trailing NULL padding 규칙을 사용합니다. 중간 컬럼을 건너뛰는 positional 입력은 지원하지 않으므로, 중간 값을 NULL로 입력하려면 해당 위치에 None을 명시합니다.

from machbaseAPI import connect

conn = connect(host='127.0.0.1', port=5656, user='SYS', password='MANAGER')
cur = conn.cursor()

cur.execute('drop table py_append_null')
cur.execute('create table py_append_null(ts datetime, name varchar(20), value double, note varchar(40))')

conn.append('PY_APPEND_NULL', [
    ['2024-01-01 10:00:00', 'sensor-1', 12.3],
    ['2024-01-01 10:00:01', 'sensor-2', None, 'manual null'],
])

cur.execute('select ts, name, value, note from py_append_null order by ts')
print(cur.fetchall())
conn.close()

첫 번째 row는 note 컬럼을 생략했으므로 NULL로 저장됩니다. 두 번째 row는 value 위치에 None을 명시했으므로 value가 NULL로 저장됩니다.

TAG 테이블 append와 metadata NULL 예제

TAG 테이블은 name, time, value에 해당하는 값까지는 반드시 입력해야 합니다. value 뒤에 정의한 추가 컬럼 또는 metadata 컬럼은 생략할 수 있으며, 생략된 컬럼은 NULL로 저장됩니다.

from machbaseAPI import connect

conn = connect(host='127.0.0.1', port=5656, user='SYS', password='MANAGER')
cur = conn.cursor()

cur.execute('drop table py_tag_append_null')
cur.execute('''
    create tag table py_tag_append_null (
        name varchar(40) primary key,
        time datetime basetime,
        value double summarized,
        status varchar(20)
    ) metadata (
        site varchar(20),
        line integer
    )
''')

conn.append('PY_TAG_APPEND_NULL', [
    ['tag-1', '2024-01-01 10:00:00', 12.3],
])

cur.execute('select name, time, value, status, site, line from py_tag_append_null')
print(cur.fetchall())
conn.close()

위 예제에서 status, site, line은 모두 NULL로 저장됩니다. 반대로 value를 생략한 TAG append는 오류로 처리됩니다.

API 참고 및 샘플 (레거시 1.x 기준)

아래 예제는 기존 레거시 버전 기준 정리입니다. 2.3 순수 Python에서는 getSessionId(), count(), checkBit()와 같은 API가 제공되지 않습니다. 필요 시 2.3 DB-API 예제를 참고하세요.

각 스크립트에서 호스트·포트·계정 정보를 환경에 맞게 수정하세요. 모든 예제는 독립 실행이 가능하며 python3 script.py 형태로 실행할 수 있습니다.

연결 관리

machbase.open(), machbase.isOpened(), machbase.isConnected(), machbase.getSessionId(), machbase.close() 

#!/usr/bin/env python3
from machbaseAPI.machbaseAPI import machbase

def main():
    db = machbase()
    print('isOpened before open:', db.isOpened())
    print('isConnected before open:', db.isConnected())

    if db.open('127.0.0.1', 'SYS', 'MANAGER', 5656) == 0:
        raise SystemExit(db.result())

    print('session id:', db.getSessionId())
    print('isOpened after open:', db.isOpened())
    print('isConnected after open:', db.isConnected())

    if db.close() == 0:
        raise SystemExit(db.result())

    print('isOpened after close:', db.isOpened())
    print('isConnected after close:', db.isConnected())

if __name__ == '__main__':
    main()

machbase.openEx() 

#!/usr/bin/env python3
from machbaseAPI.machbaseAPI import machbase

def main():
    db = machbase()
    conn_str = 'APP_NAME=python-demo'
    if db.openEx('127.0.0.1', 'SYS', 'MANAGER', 5656, conn_str) == 0:
        raise SystemExit(db.result())
    print('connected with openEx, session id:', db.getSessionId())
    if db.close() == 0:
        raise SystemExit(db.result())

if __name__ == '__main__':
    main()

DML과 결과 버퍼

machbase.execute(), machbase.result(), machbase.count() 

#!/usr/bin/env python3
import json
from machbaseAPI.machbaseAPI import machbase

def main():
    db = machbase()
    if db.open('127.0.0.1', 'SYS', 'MANAGER', 5656) == 0:
        raise SystemExit(db.result())

    try:
        rc = db.execute('drop table py_exec_demo')
        print('drop table rc:', rc)
        print('drop table result:', db.result())

        ddl = 'create table py_exec_demo(id integer, note varchar(32))'
        if db.execute(ddl) == 0:
            raise SystemExit(db.result())
        print('create table result:', db.result())

        for idx in range(2):
            sql = f"insert into py_exec_demo values ({idx}, 'row-{idx}')"
            if db.execute(sql) == 0:
                raise SystemExit(db.result())
            print('insert result:', db.result())

        if db.execute('select * from py_exec_demo order by id') == 0:
            raise SystemExit(db.result())
        payload = db.result()
        print('select payload:', payload)
        rows = json.loads(payload)
        print('decoded rows:', rows)
        print('row count via count():', db.count())
    finally:
        if db.close() == 0:
            raise SystemExit(db.result())

if __name__ == '__main__':
    main()

스트리밍 SELECT 도우미

machbase.select(), machbase.fetch(), machbase.selectClose() 

#!/usr/bin/env python3
import json
from machbaseAPI.machbaseAPI import machbase

def main():
    db = machbase()
    if db.open('127.0.0.1', 'SYS', 'MANAGER', 5656) == 0:
        raise SystemExit(db.result())

    try:
        rc = db.execute('drop table py_select_demo')
        print('drop table rc:', rc)
        print('drop table result:', db.result())

        ddl = 'create table py_select_demo(id integer, value double)'
        if db.execute(ddl) == 0:
            raise SystemExit(db.result())
        print('create table result:', db.result())

        for idx in range(5):
            sql = f"insert into py_select_demo values ({idx}, {idx * 1.5})"
            if db.execute(sql) == 0:
                raise SystemExit(db.result())
            print('insert result:', db.result())

        if db.select('select id, value from py_select_demo order by id') == 0:
            raise SystemExit(db.result())

        print('buffered rows:', db.count())
        while True:
            rc, payload = db.fetch()
            if rc == 0:
                break
            print('fetched row:', json.loads(payload))

        db.selectClose()
    finally:
        if db.close() == 0:
            raise SystemExit(db.result())

if __name__ == '__main__':
    main()

스키마 도우미

machbase.schema() 

#!/usr/bin/env python3
from machbaseAPI.machbaseAPI import machbase

def main():
    db = machbase()
    if db.open('127.0.0.1', 'SYS', 'MANAGER', 5656) == 0:
        raise SystemExit(db.result())

    try:
        rc = db.schema('drop table py_schema_demo')
        print('schema drop rc:', rc)
        print('schema drop result:', db.result())

        ddl = 'create table py_schema_demo(name varchar(20), created datetime)'
        if db.schema(ddl) == 0:
            raise SystemExit(db.result())
        print('schema create result:', db.result())
    finally:
        if db.close() == 0:
            raise SystemExit(db.result())

if __name__ == '__main__':
    main()

메타데이터와 통계

machbase.tables(), machbase.columns(), machbase.column(), machbase.statistics() 

#!/usr/bin/env python3
from machbaseAPI.machbaseAPI import machbase

def main():
    db = machbase()
    if db.open('127.0.0.1', 'SYS', 'MANAGER', 5656) == 0:
        raise SystemExit(db.result())

    try:
        if db.tables() == 0:
            raise SystemExit(db.result())
        print('tables metadata:', db.result())

        if db.columns('PY_EXEC_DEMO') == 0:
            raise SystemExit(db.result())
        print('columns metadata:', db.result())

        if db.column('PY_EXEC_DEMO') == 0:
            raise SystemExit(db.result())
        print('column metadata:', db.result())

        if db.statistics('PY_EXEC_DEMO') == 0:
            raise SystemExit(db.result())
        print('statistics output:', db.result())
    finally:
        if db.close() == 0:
            raise SystemExit(db.result())

if __name__ == '__main__':
    main()

Append 프로토콜 기본기

appendOpen(), appendData(), appendFlush(), appendClose()를 조합하면 행을 효율적으로 스트리밍할 수 있습니다. 2.1 이후에는 타입을 생략하고 appendOpen()으로 시작할 수 있습니다. appendData()와 appendDataByTime()는 호출 시 데이터 패킷을 즉시 전송합니다. appendFlush()는 이미 전송된 append 데이터의 pending response를 확인하는 동기화 지점입니다.

#!/usr/bin/env python3
from machbaseAPI.machbaseAPI import machbase

def main():
    db = machbase()
    if db.open('127.0.0.1', 'SYS', 'MANAGER', 5656) == 0:
        raise SystemExit(db.result())

    try:
        rc = db.execute('drop table py_append_demo')
        print('drop table rc:', rc)
        print('drop table result:', db.result())

        ddl = 'create table py_append_demo(ts datetime, device varchar(32), value double)'
        if db.execute(ddl) == 0:
            raise SystemExit(db.result())
        print('create table result:', db.result())

        if db.appendOpen('PY_APPEND_DEMO') == 0:
            raise SystemExit(db.result())

        rows = [
            ['2024-01-01 09:00:00', 'sensor-a', 21.5],
            ['2024-01-01 09:05:00', 'sensor-b', 22.1],
        ]
        if db.appendData('PY_APPEND_DEMO', rows) == 0:
            raise SystemExit(db.result())
        print('appendData result:', db.result())

        if db.appendFlush() == 0:
            raise SystemExit(db.result())
        print('appendFlush result:', db.result())

        if db.appendClose() == 0:
            raise SystemExit(db.result())
        print('appendClose result:', db.result())
    finally:
        if db.close() == 0:
            raise SystemExit(db.result())

if __name__ == '__main__':
    main()

Append 편의 함수

machbase.append() 

#!/usr/bin/env python3
from machbaseAPI.machbaseAPI import machbase

def main():
    db = machbase()
    if db.open('127.0.0.1', 'SYS', 'MANAGER', 5656) == 0:
        raise SystemExit(db.result())

    try:
        db.execute('drop table py_append_auto')
        db.result()
        ddl = 'create table py_append_auto(ts datetime, tag varchar(16), reading double)'
        if db.execute(ddl) == 0:
            raise SystemExit(db.result())
        db.result()

        values = [
            ['2024-01-01 10:00:00', 'node-1', 30.0],
            ['2024-01-01 10:01:00', 'node-1', 30.5],
        ]
        if db.append('PY_APPEND_AUTO', values) == 0:
            raise SystemExit(db.result())
        print('append() result:', db.result())
    finally:
        if db.close() == 0:
            raise SystemExit(db.result())

if __name__ == '__main__':
    main()

machbase.appendDataByTime(), machbase.appendByTime() 

#!/usr/bin/env python3
from machbaseAPI.machbaseAPI import machbase

def main():
    db = machbase()
    if db.open('127.0.0.1', 'SYS', 'MANAGER', 5656) == 0:
        raise SystemExit(db.result())

    try:
        db.execute('drop table py_append_time')
        db.result()
        ddl = 'create table py_append_time(ts datetime, tag varchar(16), reading double)'
        if db.execute(ddl) == 0:
            raise SystemExit(db.result())
        db.result()

        rows = [
            ['2024-01-01 11:00:00', 'node-2', 40.1],
            ['2024-01-01 11:01:00', 'node-2', 40.7],
        ]
        epoch_times = [1704106800, 1704106860]

        if db.appendOpen('PY_APPEND_TIME') == 0:
            raise SystemExit(db.result())
        if db.appendDataByTime('PY_APPEND_TIME', rows, 'YYYY-MM-DD HH24:MI:SS', epoch_times) == 0:
            raise SystemExit(db.result())
        print('appendDataByTime result:', db.result())
        db.appendClose()

        if db.appendByTime('PY_APPEND_TIME', rows, 'YYYY-MM-DD HH24:MI:SS', epoch_times) == 0:
            raise SystemExit(db.result())
        print('appendByTime result:', db.result())
    finally:
        if db.close() == 0:
            raise SystemExit(db.result())

if __name__ == '__main__':
    main()

진단 도우미

machbase.checkBit()

checkBit()는 기존 native 기반 버전에 있던 포인터 폭 확인용 API로, 2.3 순수 Python 패키지에서는 더 이상 제공되지 않습니다.

저수준 바인딩

2.3 순수 Python 패키지에서는 get_library_path(), openDB(), execAppend*() 또는 포인터 유틸리티 API(예: getlAddr, getrAddr)와 같은 저수준 ctypes 인터페이스를 제공하지 않습니다. 기존 C 레이어 직접 접근이 필요한 경우에는 2.0 이전 버전(네이티브 기반 패키지)을 사용하세요.

최근 업데이트
JDBCNode.js / TypeScript