chDB Python API 참조

chdb.query(sql, output_format='CSV', path='', udf_path='')

>>> # 기본 CSV 쿼리
>>> result = chdb.query("SELECT 1, 'hello'")
>>> print(result)
"1,hello"

>>> # DataFrame 형식으로 쿼리 실행
>>> df = chdb.query("SELECT 1 as id, 'hello' as msg", "DataFrame")
>>> print(df)
   id    msg
0   1  hello

>>> # 파일 기반 데이터베이스를 사용한 쿼리
>>> result = chdb.query("CREATE TABLE test (id INT) ENGINE = Memory", path="mydb.chdb")

>>> # UDF를 사용한 쿼리
>>> result = chdb.query("SELECT my_udf('test')", udf_path="/path/to/udfs")

chdb.sql(sql, output_format='CSV', path='', udf_path='')

>>> # 기본 CSV 쿼리
>>> result = chdb.query("SELECT 1, 'hello'")
>>> print(result)
"1,hello"

>>> # DataFrame 출력으로 쿼리
>>> df = chdb.query("SELECT 1 as id, 'hello' as msg", "DataFrame")
>>> print(df)
   id    msg
0   1  hello

>>> # 파일 기반 데이터베이스로 쿼리
>>> result = chdb.query("CREATE TABLE test (id INT) ENGINE = Memory", path="mydb.chdb")

>>> # UDF를 사용한 쿼리
>>> result = chdb.query("SELECT my_udf('test')", udf_path="/path/to/udfs")

chdb.to_arrowTable(res)

>>> result = chdb.query("SELECT 1 as id, 'hello' as msg", "Arrow")
>>> table = chdb.to_arrowTable(result)
>>> print(table.to_pandas())
   id    msg
0   1  hello

chdb.to_df(r)

>>> result = chdb.query("SELECT 1 as id, 'hello' as msg", "Arrow")
>>> df = chdb.to_df(result)
>>> print(df)
   id    msg
0   1  hello

chdb.connect(connection_string: str = ':memory:') → Connection

>>> # 인메모리 데이터베이스
>>> conn = connect()
>>> conn = connect(":memory:")
>>>
>>> # 파일 기반 데이터베이스
>>> conn = connect("my_data.db")
>>> conn = connect("/path/to/data.db")
>>>
>>> # 매개변수 사용
>>> conn = connect("data.db?mode=ro")  # 읽기 전용 모드
>>> conn = connect(":memory:?verbose&log-level=debug")  # 디버그 로깅
>>>
>>> # 자동 정리를 위한 컨텍스트 매니저 사용
>>> with connect("data.db") as conn:
...     result = conn.query("SELECT 1")
...     print(result)
>>> # 연결 자동 종료

class chdb.session.Session(path=None)

cleanup()

>>> session = Session("test.db")
>>> try:
...     session.query("INVALID SQL")
... finally:
...     session.cleanup()  # 오류 발생 여부와 관계없이 안전하게 정리

close()

>>> session = Session("test.db")
>>> session.query("SELECT 1")
>>> session.close()  # 명시적으로 세션을 닫습니다

query(sql, fmt='CSV', udf_path='')

>>> session = Session("test.db")
>>>
>>> # 기본 CSV 형식으로 쿼리 실행
>>> result = session.query("SELECT 1 as number")
>>> print(result)
number
1

>>> # JSON 포맷으로 쿼리
>>> result = session.query("SELECT 1 as number", fmt="JSON")
>>> print(result)
{"number": "1"}

>>> # Complex query with table creation
>>> session.query("CREATE TABLE test (id INT, name String) ENGINE = Memory")
>>> session.query("INSERT INTO test VALUES (1, 'Alice'), (2, 'Bob')")
>>> result = session.query("SELECT * FROM test ORDER BY id")
>>> print(result)
id,name
1,Alice
2,Bob

send_query(sql, fmt='CSV') → StreamingResult

>>> session = Session("test.db")
>>> session.query("CREATE TABLE big_table (id INT, data String) ENGINE = MergeTree() order by id")
>>>
>>> # 대용량 데이터셋 삽입
>>> for i in range(1000):
...     session.query(f"INSERT INTO big_table VALUES ({i}, 'data_{i}')")
>>>
>>> # 메모리 문제를 방지하기 위해 결과를 스트리밍
>>> streaming_result = session.send_query("SELECT * FROM big_table ORDER BY id")
>>> for chunk in streaming_result:
...     print(f"Processing chunk: {len(chunk)} bytes")
...     # 전체 결과 집합을 로드하지 않고 청크 처리

>>> # 컨텍스트 매니저와 함께 사용
>>> with session.send_query("SELECT COUNT(*) FROM big_table") as stream:
...     for result in stream:
...         print(f"Count result: {result}")

sql(sql, fmt='CSV', udf_path='')

>>> session = Session("test.db")
>>>
>>> # 기본 CSV 형식으로 쿼리 실행
>>> result = session.query("SELECT 1 as number")
>>> print(result)
number
1

>>> # JSON 포맷으로 쿼리
>>> result = session.query("SELECT 1 as number", fmt="JSON")
>>> print(result)
{"number": "1"}

>>> # 테이블 생성을 포함한 복합 쿼리
>>> session.query("CREATE TABLE test (id INT, name String) ENGINE = MergeTree() order by id")
>>> session.query("INSERT INTO test VALUES (1, 'Alice'), (2, 'Bob')")
>>> result = session.query("SELECT * FROM test ORDER BY id")
>>> print(result)
id,name
1,Alice
2,Bob

chdb.state.connect(connection_string: str = ':memory:') → Connection

>>> # 인메모리 데이터베이스
>>> conn = connect()
>>> conn = connect(":memory:")
>>>
>>> # 파일 기반 데이터베이스
>>> conn = connect("my_data.db")
>>> conn = connect("/path/to/data.db")
>>>
>>> # 매개변수 사용
>>> conn = connect("data.db?mode=ro")  # 읽기 전용 모드
>>> conn = connect(":memory:?verbose&log-level=debug")  # 디버그 로깅
>>>
>>> # 자동 정리를 위한 컨텍스트 매니저 사용
>>> with connect("data.db") as conn:
...     result = conn.query("SELECT 1")
...     print(result)
>>> # 연결 자동 종료

class chdb.state.sqlitelike.Connection(connection_string: str)

close() → None

>>> conn = connect("test.db")
>>> # 쿼리에 연결 사용
>>> conn.query("CREATE TABLE test (id INT) ENGINE = Memory")
>>> # 완료 후 닫기
>>> conn.close()

>>> # 컨텍스트 매니저 사용 (자동 정리)
>>> with connect("test.db") as conn:
...     conn.query("SELECT 1")
...     # 연결이 자동으로 닫힘

cursor() → Cursor

>>> conn = connect(":memory:")
>>> cursor = conn.cursor()
>>> cursor.execute("CREATE TABLE test (id INT, name String) ENGINE = Memory")
>>> cursor.execute("INSERT INTO test VALUES (1, 'Alice')")
>>> cursor.execute("SELECT * FROM test")
>>> rows = cursor.fetchall()
>>> print(rows)
((1, 'Alice'),)

query(query: str, format: str = 'CSV') → Any

>>> conn = connect(":memory:")
>>>
>>> # 기본 CSV 쿼리
>>> result = conn.query("SELECT 1 as num, 'hello' as text")
>>> print(result)
num,text
1,hello

>>> # DataFrame 포맷
>>> df = conn.query("SELECT number FROM numbers(5)", "dataframe")
>>> print(df)
   number
0       0
1       1
2       2
3       3
4       4

send_query(query: str, format: str = 'CSV') → StreamingResult

>>> conn = connect(":memory:")
>>>
>>> # 기본 스트리밍
>>> stream = conn.send_query("SELECT number FROM numbers(1000)")
>>> for chunk in stream:
...     print(f"청크 처리 중: {len(chunk)} 바이트")

>>> # 정리를 위한 컨텍스트 매니저 사용
>>> with conn.send_query("SELECT * FROM large_table") as stream:
...     chunk = stream.fetch()
...     while chunk:
...         process_data(chunk)
...         chunk = stream.fetch()

>>> # RecordBatch 스트리밍을 위한 Arrow 형식
>>> stream = conn.send_query("SELECT * FROM data", "Arrow")
>>> reader = stream.record_batch(rows_per_batch=10000)
>>> for batch in reader:
...     print(f"Batch shape: {batch.num_rows} x {batch.num_columns}")

class chdb.state.sqlitelike.StreamingResult

fetch() → Any

>>> stream = conn.send_query("SELECT * FROM large_table")
>>> chunk = stream.fetch()
>>> while chunk is not None:
...     process_data(chunk)
...     chunk = stream.fetch()

cancel() → None

>>> stream = conn.send_query("SELECT * FROM very_large_table")
>>> for i, chunk in enumerate(stream):
...     if i >= 10:  # 처음 10개의 청크만 처리
...         stream.cancel()
...         break
...     process_data(chunk)

close() → None

record_batch(rows_per_batch: int = 1000000) → pa.RecordBatchReader

>>> stream = conn.send_query("SELECT * FROM data", format="Arrow")
>>> reader = stream.record_batch(rows_per_batch=10000)
>>> for batch in reader:
...     print(f"Processing batch: {batch.num_rows} rows")
...     df = batch.to_pandas()
...     process_dataframe(df)

>>> stream = conn.send_query("SELECT number FROM numbers(1000000)")
>>> for chunk in stream:
...     print(f"Chunk size: {len(chunk)} bytes")

>>> with conn.send_query("SELECT * FROM data") as stream:
...     for chunk in stream:
...         process(chunk)
>>> # 스트림이 자동으로 닫힘

class chdb.state.sqlitelike.Cursor(connection)

close() → None

>>> cursor = conn.cursor()
>>> cursor.execute("SELECT 1")
>>> result = cursor.fetchone()
>>> cursor.close()  # 커서 리소스 정리

column_names() → list

>>> cursor = conn.cursor()
>>> cursor.execute("SELECT id, name, email FROM users LIMIT 1")
>>> print(cursor.column_names())
['id', 'name', 'email']

column_types() → list

>>> cursor = conn.cursor()
>>> cursor.execute("SELECT toInt32(1), toString('hello')")
>>> print(cursor.column_types())
['Int32', 'String']

commit() → None

>>> cursor = conn.cursor()
>>> cursor.execute("INSERT INTO test VALUES (1, 'data')")
>>> cursor.commit()

>>> cursor = conn.cursor()
>>> cursor.execute("SELECT id, name FROM users LIMIT 1")
>>> for desc in cursor.description:
...     print(f"Column: {desc[0]}, Type: {desc[1]}")
Column: id, Type: Int32
Column: name, Type: String

execute(query: str) → None

>>> cursor = conn.cursor()
>>>
>>> # DDL 실행
>>> cursor.execute("CREATE TABLE test (id INT, name String) ENGINE = Memory")
>>>
>>> # DML 실행
>>> cursor.execute("INSERT INTO test VALUES (1, 'Alice')")
>>>
>>> # SELECT 실행 및 결과 조회
>>> cursor.execute("SELECT * FROM test")
>>> rows = cursor.fetchall()
>>> print(rows)
((1, 'Alice'),)

fetchall() → tuple

>>> cursor = conn.cursor()
>>> cursor.execute("SELECT id, name FROM users")
>>> all_users = cursor.fetchall()
>>> for user_id, user_name in all_users:
...     print(f"User {user_id}: {user_name}")

fetchmany(size: int = 1) → tuple

>>> cursor = conn.cursor()
>>> cursor.execute("SELECT * FROM large_table")
>>>
>>> # 결과를 배치로 처리
>>> while True:
...     batch = cursor.fetchmany(100)  # 한 번에 100행씩 가져오기
...     if not batch:
...         break
...     process_batch(batch)

fetchone() → tuple | None

>>> cursor = conn.cursor()
>>> cursor.execute("SELECT id, name FROM users")
>>> row = cursor.fetchone()
>>> while row is not None:
...     user_id, user_name = row
...     print(f"User {user_id}: {user_name}")
...     row = cursor.fetchone()

chdb.state.sqlitelike.to_arrowTable(res)

>>> import chdb
>>> result = chdb.query("SELECT 1 as num, 'hello' as text", "Arrow")
>>> table = to_arrowTable(result)
>>> print(table.schema)
num: int64
text: string
>>> print(table.to_pandas())
   num   text
0    1  hello

chdb.state.sqlitelike.to_df(r)

>>> import chdb
>>> result = chdb.query("SELECT 1 as num, 'hello' as text", "Arrow")
>>> df = to_df(result)
>>> print(df)
   num   text
0    1  hello
>>> print(df.dtypes)
num      int64
text    object
dtype: object

class chdb.dataframe.Table(*args: Any, **kwargs: Any)

chdb.dbapi.connect(*args, **kwargs)

chdb.dbapi.get_client_info()

chdb.dbapi.Binary(x)

class chdb.dbapi.connections.Connection(path=None)

>>> # 인메모리 데이터베이스
>>> conn = Connection()
>>> cursor = conn.cursor()
>>> cursor.execute("SELECT 1")
>>> result = cursor.fetchall()
>>> conn.close()

>>> # 파일 기반 데이터베이스
>>> conn = Connection('mydata.db')
>>> with conn.cursor() as cur:
...     cur.execute("CREATE TABLE users (id INT, name STRING) ENGINE = MergeTree() order by id")
...     cur.execute("INSERT INTO users VALUES (1, 'Alice')")
>>> conn.close()

>>> # 컨텍스트 매니저 사용
>>> with Connection() as cur:
...     cur.execute("SELECT version()")
...     version = cur.fetchone()

close()

commit()

cursor(cursor=None)

>>> conn = Connection()
>>> cur = conn.cursor()
>>> cur.execute("SELECT 1")
>>> result = cur.fetchone()

escape(obj, mapping=None)

>>> conn = Connection()
>>> safe_value = conn.escape("O'Reilly")
>>> query = f"SELECT * FROM users WHERE name = {safe_value}"

escape_string(s)

query(sql, fmt='CSV')

>>> conn = Connection()
>>> result = conn.query("SELECT 1, 'hello'", "CSV")
>>> print(result)
"1,hello\n"

rollback()

class chdb.dbapi.cursors.Cursor(connection)

>>> conn = Connection()
>>> cur = conn.cursor()
>>> cur.execute("SELECT 1 as id, 'test' as name")
>>> result = cur.fetchone()
>>> print(result)  # (1, 'test') 출력
>>> cur.close()

callproc(procname, args=())

close()

execute(query, args=None)

>>> # 물음표 매개변수
>>> cur.execute("SELECT * FROM users WHERE id = ? AND age > ?", (123, 18))
>>>
>>> # 이름 있는 매개변수
>>> cur.execute("SELECT * FROM users WHERE name = %(name)s", {'name': 'Alice'})
>>>
>>> # 매개변수 없음
>>> cur.execute("SELECT COUNT(*) FROM users")

executemany(query, args)

>>> # 물음표 매개변수를 사용한 일괄 삽입
>>> users_data = [(1, 'Alice'), (2, 'Bob'), (3, 'Charlie')]
>>> cur.executemany("INSERT INTO users VALUES (?, ?)", users_data)
>>>
>>> # 이름 있는 매개변수를 사용한 일괄 삽입
>>> users_data = [
...     {'id': 1, 'name': 'Alice'},
...     {'id': 2, 'name': 'Bob'}
... ]
>>> cur.executemany(
...     "INSERT INTO users VALUES (%(id)s, %(name)s)",
...     users_data
... )

fetchall()

>>> cursor.execute("SELECT id, name FROM users")
>>> all_rows = cursor.fetchall()
>>> print(len(all_rows))  # 전체 행 수

fetchmany(size=1)

>>> cursor.execute("SELECT id, name FROM users")
>>> rows = cursor.fetchmany(3)
>>> print(rows)  # [(1, 'Alice'), (2, 'Bob'), (3, 'Charlie')]

fetchone()

>>> cursor.execute("SELECT id, name FROM users LIMIT 3")
>>> row = cursor.fetchone()
>>> print(row)  # (1, 'Alice') - 첫 번째 행
>>> row = cursor.fetchone()
>>> print(row)  # (2, 'Bob')

mogrify(query, args=None)

>>> cur.mogrify("SELECT * FROM users WHERE id = ?", (123,))
"SELECT * FROM users WHERE id = 123"

nextset()

setinputsizes(*args)

setoutputsizes(*args)

StandardError
├── Warning
└── Error
    ├── InterfaceError
    └── DatabaseError
        ├── DataError
        ├── OperationalError
        ├── IntegrityError
        ├── InternalError
        ├── ProgrammingError
        └── NotSupportedError

>>> try:
...     cursor.execute("SELECT * FROM nonexistent_table")
... except ProgrammingError as e:
...     print(f"SQL Error: {e}")
...
SQL Error: Table 'nonexistent_table' doesn't exist

>>> try:
...     cursor.execute("INSERT INTO users (id) VALUES (1), (1)")
... except IntegrityError as e:
...     print(f"Constraint violation: {e}")
...
Constraint violation: Duplicate entry '1' for key 'PRIMARY'

>>> # SQL에서 0으로 나누기
>>> cursor.execute("SELECT 1/0")
DataError: Division by zero

>>> # 잘못된 날짜 포맷
>>> cursor.execute("INSERT INTO table VALUES ('invalid-date')")
DataError: Invalid date format

>>> # 기본 키(primary key) 중복
>>> cursor.execute("INSERT INTO users (id, name) VALUES (1, 'John')")
>>> cursor.execute("INSERT INTO users (id, name) VALUES (1, 'Jane')")
IntegrityError: Duplicate entry '1' for key 'PRIMARY'

>>> # 외래 키 위반
>>> cursor.execute("INSERT INTO orders (user_id) VALUES (999)")
IntegrityError: Cannot add or update a child row: foreign key constraint fails

>>> # 트랜잭션을 지원하지 않는 연결에서 트랜잭션 롤백 시도
>>> connection.rollback()
NotSupportedError: Transactions are not supported

>>> # 지원되지 않는 SQL 구문 사용
>>> cursor.execute("SELECT * FROM table WITH (NOLOCK)")
NotSupportedError: WITH clause not supported in this database version

>>> # 테이블을 찾을 수 없음
>>> cursor.execute("SELECT * FROM nonexistent_table")
ProgrammingError: Table 'nonexistent_table' doesn't exist

>>> # SQL 구문 오류
>>> cursor.execute("SELCT * FROM users")
ProgrammingError: You have an error in your SQL syntax