chDB Python API 참조

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

>>> # Basic CSV query
>>> result = chdb.query("SELECT 1, 'hello'")
>>> print(result)
"1,hello"

>>> # Query with DataFrame output
>>> df = chdb.query("SELECT 1 as id, 'hello' as msg", "DataFrame")
>>> print(df)
   id    msg
0   1  hello

>>> # Query with file-based database
>>> result = chdb.query("CREATE TABLE test (id INT) ENGINE = Memory", path="mydb.chdb")

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

>>> # Basic CSV query
>>> result = chdb.query("SELECT 1, 'hello'")
>>> print(result)
"1,hello"

>>> # Query with DataFrame output
>>> df = chdb.query("SELECT 1 as id, 'hello' as msg", "DataFrame")
>>> print(df)
   id    msg
0   1  hello

>>> # Query with file-based database
>>> result = chdb.query("CREATE TABLE test (id INT) ENGINE = Memory", path="mydb.chdb")

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

>>> # Temporary database
>>> conn = connect()
>>> conn = connect(":memory:")
>>>
>>> # File-based database
>>> conn = connect("my_data.db")
>>> conn = connect("/path/to/data.db")
>>>
>>> # With parameters
>>> conn = connect("data.db?mode=ro")  # Read-only mode
>>> conn = connect(":memory:?verbose&log-level=debug")  # Debug logging
>>>
>>> # Using context manager for automatic cleanup
>>> with connect("data.db") as conn:
...     result = conn.query("SELECT 1")
...     print(result)
>>> # Connection automatically closed

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")
>>>
>>> # Basic query with default CSV format
>>> result = session.query("SELECT 1 as number")
>>> print(result)
number
1

>>> # Query with JSON format
>>> 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")
>>>
>>> # Basic query with default CSV format
>>> result = session.query("SELECT 1 as number")
>>> print(result)
number
1

>>> # Query with JSON format
>>> 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 = 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

>>> # Temporary database
>>> conn = connect()
>>> conn = connect(":memory:")
>>>
>>> # File-based database
>>> conn = connect("my_data.db")
>>> conn = connect("/path/to/data.db")
>>>
>>> # With parameters
>>> conn = connect("data.db?mode=ro")  # Read-only mode
>>> conn = connect(":memory:?verbose&log-level=debug")  # Debug logging
>>>
>>> # Using context manager for automatic cleanup
>>> with connect("data.db") as conn:
...     result = conn.query("SELECT 1")
...     print(result)
>>> # Connection automatically closed

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)

>>> # Temporary database
>>> conn = Connection()
>>> cursor = conn.cursor()
>>> cursor.execute("SELECT 1")
>>> result = cursor.fetchall()
>>> conn.close()

>>> # File-based database
>>> 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()

>>> # Context manager usage
>>> 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

>>> # 잘못된 매개변수 개수
>>> cursor.execute("INSERT INTO users (name, age) VALUES (%s)", ('John',))
ProgrammingError: Column count doesn't match value count

str(object=’’) -> str
str(bytes_or_buffer[, encoding[, errors]]) -> str

int([x]) -> integer
int(x, base=10) -> integer

>>> int(‘0b100’, base=0)
4

str(object=’’) -> str
str(bytes_or_buffer[, encoding[, errors]]) -> str

>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types  # True 반환
>>> FIELD_TYPE.INT != string_types     # True 반환
>>> FIELD_TYPE.BLOB in string_types    # False 반환

>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types  # True 반환
>>> FIELD_TYPE.INT != string_types     # True 반환
>>> FIELD_TYPE.BLOB in string_types    # False 반환

>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types  # True 반환
>>> FIELD_TYPE.INT != string_types     # True 반환
>>> FIELD_TYPE.BLOB in string_types    # False 반환

>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types  # True 반환
>>> FIELD_TYPE.INT != string_types     # True 반환
>>> FIELD_TYPE.BLOB in string_types    # False 반환

>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types  # True 반환
>>> FIELD_TYPE.INT != string_types     # True 반환
>>> FIELD_TYPE.BLOB in string_types    # False 반환

>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types  # True 반환
>>> FIELD_TYPE.INT != string_types     # True 반환
>>> FIELD_TYPE.BLOB in string_types    # False 반환

>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types  # True 반환
>>> FIELD_TYPE.INT != string_types     # True 반환
>>> FIELD_TYPE.BLOB in string_types    # False 반환

>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types  # Returns True
>>> FIELD_TYPE.INT != string_types     # Returns True
>>> FIELD_TYPE.BLOB in string_types    # Returns False

import chdb.dbapi as dbapi

print("chdb driver version: {0}".format(dbapi.get_client_info()))

# Create connection and cursor
conn = dbapi.connect()
cur = conn.cursor()

# Execute query
cur.execute('SELECT version()')
print("description:", cur.description)
print("data:", cur.fetchone())

# Clean up
cur.close()
conn.close()

import chdb.dbapi as dbapi

conn = dbapi.connect()
cur = conn.cursor()

# Create table
cur.execute("""
    CREATE TABLE employees (
        id UInt32,
        name String,
        department String,
        salary Decimal(10,2)
    ) ENGINE = Memory
""")

# Insert data
cur.execute("""
    INSERT INTO employees VALUES
    (1, 'Alice', 'Engineering', 75000.00),
    (2, 'Bob', 'Marketing', 65000.00),
    (3, 'Charlie', 'Engineering', 80000.00)
""")

# Query data
cur.execute("SELECT * FROM employees WHERE department = 'Engineering'")

# Fetch results
print("Column names:", [desc[0] for desc in cur.description])
for row in cur.fetchall():
    print(row)

conn.close()

import chdb.dbapi as dbapi

# Temporary database (default)
conn1 = dbapi.connect()

# Persistent database file
conn2 = dbapi.connect("./my_database.chdb")

# Connection with parameters
conn3 = dbapi.connect("./my_database.chdb?log-level=debug&verbose")

# Read-only connection
conn4 = dbapi.connect("./my_database.chdb?mode=ro")

# Automatic connection cleanup
with dbapi.connect("test.chdb") as conn:
    cur = conn.cursor()
    cur.execute("SELECT count() FROM numbers(1000)")
    result = cur.fetchone()
    print(f"Count: {result[0]}")
    cur.close()

chdb.create_function(name, func, arg_types=None, return_type=None, *, on_null=None, on_error=None)

from chdb import create_function, drop_function, query
from chdb.sqltypes import INT64, STRING

create_function("strlen", len, arg_types=[STRING], return_type=INT64)
print(query("SELECT strlen('hello')"))  # 5

drop_function("strlen")

chdb.drop_function(name)

from chdb import func

@func(arg_types=None, return_type=None, *, on_null=None, on_error=None)
def my_function(...):
    ...

from chdb import func, query
from chdb.sqltypes import INT64, STRING

# Explicit types
@func([INT64, INT64], INT64)
def add(a, b):
    return a + b

# Types inferred from annotations
@func()
def multiply(a: int, b: int) -> int:
    return a * b

# Explicit return_type, arg_types inferred from annotations
@func(return_type=STRING)
def greet(name: str):
    return f"Hello, {name}!"

print(query("SELECT add(12, 22)"))       # 34
print(query("SELECT multiply(3, 7)"))    # 21
print(query("SELECT greet('world')"))    # Hello, world!

from chdb.sqltypes import (
    BOOL,
    INT8, INT16, INT32, INT64, INT128, INT256,
    UINT8, UINT16, UINT32, UINT64, UINT128, UINT256,
    FLOAT32, FLOAT64,
    STRING,
    DATE, DATE32, DATETIME, DATETIME64,
)

from chdb import create_function, func
from chdb.sqltypes import INT64

# 1. ChdbType constants
create_function("f1", lambda x: x, arg_types=[INT64], return_type=INT64)

# 2. ClickHouse type strings
create_function("f2", lambda x: x, arg_types=["Int64"], return_type="Int64")

# 3. Parameterized type strings
create_function("f3", lambda x: x, arg_types=["DateTime('UTC')"], return_type="DateTime('UTC')")

# 4. Python types — passed directly or used as annotations
create_function("f4", lambda x: x, arg_types=[int], return_type=int)

@func()
def f5(x: int) -> int:
    return x

from chdb import func, query, NullHandling

# Default: NULL in → NULL out, function not called
@func(return_type="Int64")
def add_one(x: int) -> int:
    return x + 1

print(query("SELECT add_one(NULL)"))  # NULL

# Pass NULL as None
@func(return_type="Int64", on_null="pass")
def null_safe(x):
    return 0 if x is None else x + 1

print(query("SELECT null_safe(NULL)"))  # 0

from chdb import func, query

# Default: exception propagates
@func(arg_types=["Int64", "Int64"], return_type="Int64")
def divide(a, b):
    return a // b

# print(query("SELECT divide(1, 0)"))  # Error: division by zero

# Ignore: exception → NULL
@func(arg_types=["Int64", "Int64"], return_type="Int64", on_error="ignore")
def safe_divide(a, b):
    return a // b

print(query("SELECT safe_divide(1, 0)"))   # NULL
print(query("SELECT safe_divide(10, 2)"))  # 5

from chdb import func, query
from datetime import datetime, timedelta, date

@func(arg_types=["DateTime('UTC')"], return_type="DateTime('UTC')")
def add_one_hour(dt):
    return dt + timedelta(hours=1)

@func()
def get_year(d: date) -> int:
    return d.year

print(query("SELECT add_one_hour(toDateTime('2024-01-01 12:00:00', 'UTC'))"))  # 2024-01-01 13:00:00
print(query("SELECT get_year(toDate('2024-06-15'))"))  # 2024

chdb.udf.chdb_udf(return_type='String')

@chdb_udf()
def sum_udf(lhs, rhs):
    return int(lhs) + int(rhs)

@chdb_udf()
def func_use_json(arg):
    import json
    # ... use json module

chdb.udf.generate_udf(func_name, args, return_type, udf_body)

chdb.utils.convert_to_columnar(items: List[Dict[str, Any]]) → Dict[str, List[Any]]

>>> items = [
...     {"name": "Alice", "age": 30, "city": "New York"},
...     {"name": "Bob", "age": 25},
...     {"name": "Charlie", "city": "San Francisco"}
... ]
>>> convert_to_columnar(items)
{
    'name': ['Alice', 'Bob', 'Charlie'],
    'age': [30, 25, None],
    'city': ['New York', None, 'San Francisco']
}

chdb.utils.flatten_dict(d: Dict[str, Any], parent_key: str = '', sep: str = '_') → Dict[str, Any]

>>> nested_dict = {
...     "a": 1,
...     "b": {
...         "c": 2,
...         "d": {
...             "e": 3
...         }
...     },
...     "f": [4, 5, {"g": 6}],
...     "h": [{"i": 7}, {"j": 8}]
... }
>>> flatten_dict(nested_dict)
{
    'a': 1,
    'b_c': 2,
    'b_d_e': 3,
    'f_0': 4,
    'f_1': 5,
    'f_2_g': 6,
    'h': '[{"i": 7}, {"j": 8}]'
}

chdb.utils.infer_data_type(values: List[Any]) → str

chdb.utils.infer_data_types`(column_data: Dict[str, List[Any]], n_rows: int = 10000) → List[tuple]

class chdb.rwabc.PyReader(data: Any)

abstractmethod (col_names: List[str], count: int) → List[Any]

class chdb.rwabc.PyWriter(col_names: List[str], types: List[type], data: Any)

abstractmethod finalize() → bytes

abstractmethod write(col_names: List[str], columns: List[List[Any]]) → None

>>> try:
...     result = chdb.query("SELECT * FROM non_existent_table")
... except chdb.ChdbError as e:
...     print(f"Query failed: {e}")
Query failed: Table 'non_existent_table' doesn't exist

>>> try:
...     result = chdb.query("SELECT invalid_syntax FROM")
... except chdb.ChdbError as e:
...     print(f"Syntax error: {e}")
Syntax error: Syntax error near 'FROM'

str(object=’’) -> str
str(bytes_or_buffer[, encoding[, errors]]) -> str