chpy-orm 0.1.2

A Python wrapper for ClickHouse database operations

chpy-orm - ClickHouse Python Wrapper

A comprehensive Python library for programmatically querying and managing ClickHouse databases. chpy-orm provides a fluent, ORM-style interface with type safety, autocomplete support, and extensive ClickHouse function coverage.

Acknowledgments

This project is developed mainly using Cursor, an AI-powered code editor, but all code is thoroughly tested and verified by the maintainer to ensure quality and reliability.

Features

• 🚀 Fluent Query Builder - Method chaining for building complex queries
• 🔒 Type-Safe Columns - ORM-style column objects with autocomplete support
• 📊 Multiple Output Formats - DataFrame, NumPy, JSON, CSV, Parquet, and more
• 🛠️ DDL Operations - High-level table and database management
• 🎯 Comprehensive Functions - Full coverage of ClickHouse functions
• 🔄 Window Functions - Advanced window function support with OVER clauses
• 🔗 JOIN Support - INNER, LEFT, RIGHT, FULL, and CROSS joins
• 📦 Type System - Support for all ClickHouse data types (Arrays, Maps, Tuples, Nested, etc.)
• 🎨 Generic Table Wrapper - Works with any ClickHouse table, not just crypto_quotes

Table of Contents

• Acknowledgments
• Features
• Installation
  • Requirements
• Quick Start
  • Basic Connection
  • Using Context Manager
• API Reference
  • ClickHouseClient
  • TableWrapper
  • CryptoQuotesTable
  • QueryBuilder
  • ORM Classes
  • DDL
  • Functions
  • WindowSpec
• Best Practices
• Table Schema
• License
• Contributing
• Support
• Basic Examples
• Intermediate Examples
• Advanced Examples

Installation

pip install chpy-orm

Or install in development mode:

pip install -e .

Requirements

• Python 3.9+
• clickhouse-connect
• pandas (optional, for DataFrame support)
• numpy (optional, for NumPy array support)

Quick Start

Basic Connection

from typing import Optional, List
from chpy import ClickHouseClient, Array, LowCardinality, String, Float64, UInt64
from chpy.tables import TableWrapper
from chpy.orm import Table, Column

# Define crypto_quotes table schema
crypto_quotes_columns = [
    Column("pair", String),
    Column("best_bid_price", Float64),
    Column("best_bid_size", Float64),
    Column("best_ask_price", Float64),
    Column("best_ask_size", Float64),
    Column("bid_prices", Array(Float64)),
    Column("bid_sizes", Array(Float64)),
    Column("ask_prices", Array(Float64)),
    Column("ask_sizes", Array(Float64)),
    Column("timestamp_ms", UInt64),
    Column("exchange", LowCardinality(String)),
    Column("sequence_number", UInt64),
    Column("inserted_at", UInt64),
]

# Create table instance
crypto_quotes = Table("crypto_quotes", "stockhouse", crypto_quotes_columns)
crypto_quotes_schema = crypto_quotes  # Alias for clarity when used as schema parameter

# CryptoQuotesTable definition
class CryptoQuotesTable(TableWrapper):
    """
    Programmatic wrapper for the crypto_quotes table.
    
    Provides a high-level interface to query crypto quotes data
    without writing SQL directly. Inherits from TableWrapper for
    generic table functionality.
    """
    
    def __init__(self, client: ClickHouseClient, database: str = "stockhouse"):
        """
        Initialize CryptoQuotesTable wrapper.
        
        Args:
            client: ClickHouseClient instance
            database: Database name (default: 'stockhouse')
        """
        # Initialize base class with crypto_quotes table and schema
        super().__init__(
            client=client,
            table_name="crypto_quotes",
            database=database,
            schema=crypto_quotes_schema
        )

# Initialize the client
client = ClickHouseClient(
    host="localhost",
    port=8123,
    username="default",
    password="",
    database="stockhouse"
)

# Create a table wrapper
table = CryptoQuotesTable(client)

# Simple query
df = (table.query()
    .where(table.c.pair == "BTC-USDT")
    .limit(10)
    .to_dataframe())

print(df)

Using Context Manager

# CryptoQuotesTable is defined above (see Basic Connection section)

with ClickHouseClient(host="localhost", database="stockhouse") as client:
    table = CryptoQuotesTable(client)
    results = table.query().where(table.c.pair == "BTC-USDT").to_list()
    # Connection automatically closed

---

API Reference

ClickHouseClient

Base client for ClickHouse operations.

Initialization

client = ClickHouseClient(
    host="localhost",
    port=8123,
    username="default",
    password="",
    database="default",
    **kwargs  # Additional connection parameters
)

Methods

• execute(query, parameters=None) - Execute a SELECT query, returns list of dicts
• query_df(query, parameters=None) - Execute query, returns pandas DataFrame
• query_np(query, parameters=None) - Execute query, returns numpy array
• query_arrow(query, parameters=None) - Execute query, returns PyArrow Table
• execute_command(query, parameters=None) - Execute non-SELECT command (INSERT, CREATE, etc.)
• insert(table, data) - Insert data into a table
• close() - Close the connection

Context Manager

with ClickHouseClient(...) as client:
    # Use client
    pass  # Connection automatically closed

TableWrapper

Generic wrapper for any ClickHouse table.

Initialization

table = TableWrapper(
    client=client,
    table_name="my_table",
    database="my_db",
    schema=optional_table_schema  # Optional Table object for type safety
)

Methods

• query() - Start building a query, returns QueryBuilder
• insert(data) - Insert data into the table
• c or columns - Access to schema columns (if schema provided)

CryptoQuotesTable

Specialized wrapper for the crypto_quotes table, extends TableWrapper.

Initialization

table = CryptoQuotesTable(client, database="stockhouse")

Helper Methods

• get_valid_exchanges() - Get list of valid exchange names
• get_exchange_base_currencies(exchange) - Get base currencies for exchange
• get_exchange_currencies(exchange) - Get supported currencies for exchange
• get_exchange_pairs(exchange) - Get all valid trading pairs for exchange
• is_valid_pair(pair, exchange=None) - Check if trading pair is valid
• get_all_valid_pairs() - Get all valid trading pairs across all exchanges

QueryBuilder

Fluent query builder returned by table.query().

Filtering Methods

• where(condition) - Add WHERE condition (chainable)
  • Accepts: ColumnExpression, CombinedExpression, SubqueryExpression, or raw SQL string
• having(condition) - Add HAVING clause for GROUP BY queries (chainable)

Query Building Methods

• select(*columns) - Specify columns to select (chainable)
  • Accepts: Column objects, Function objects, AggregateFunction objects, Subquery objects, or raw SQL strings
• join(table, condition=None, join_type="INNER", alias=None) - Add JOIN clause (chainable)
  • table: Table object or table name string
  • condition: ColumnExpression, CombinedExpression, or raw SQL string (required except for CROSS JOIN)
  • join_type: "INNER", "LEFT", "RIGHT", "FULL", or "CROSS"
  • alias: Optional table alias
• from_subquery(subquery, alias) - Use subquery as FROM clause (chainable)
• order_by(column, desc=True) - Specify ordering (chainable)
• limit(n) - Limit number of results (chainable)
• group_by(*columns) - Group by columns (chainable)

Execution Methods

• to_list() - Execute and return as list of Row objects (if schema) or dictionaries
• to_dict(key_column, value_column=None) - Execute and return as dictionary
• to_dataframe() - Execute and return as pandas DataFrame
• to_numpy(columns=None, dtype=None) - Execute and return as numpy array
• to_json(indent=None) - Execute and return as JSON string
• to_csv(path=None, **kwargs) - Execute and return/write CSV
• to_parquet(path, **kwargs) - Execute and write Parquet file
• count() - Count rows matching filters
• first() - Get first result (or None)
• exists() - Check if any rows match (returns bool)
• __iter__() - Make query builder iterable

ORM Classes

Table

Represents a database table with columns.

table = Table(name="my_table", database="my_db", columns=[...])

Column

Represents a database column.

column = Column(name="pair", type_="String", table=optional_table)

Column objects support:

• Comparison operators: ==, !=, <, <=, >, >=
• in_(values) - IN operator
• not_in(values) - NOT IN operator
• like(pattern) - LIKE operator
• Can be combined with & (AND) and | (OR)

Row

Represents a row from a query result.

• Attribute access: row.column_name
• Dictionary access: row['column_name']
• get(key, default) - Get with default
• to_dict() - Convert to dictionary

DDL

High-level DDL operations for table and database management.

Methods

• create_table(table, columns=None, database=None, engine="MergeTree", order_by=None, ...)
• drop_table(table, database=None, if_exists=True)
• add_column(table, column, database=None, after=None, if_not_exists=True)
• drop_column(table, column_name, database=None, if_exists=True)
• modify_column(table, column, database=None)
• rename_table(old_table, new_name, database=None)
• create_database(database, if_not_exists=True, engine=None, settings=None)
• drop_database(database, if_exists=True)
• create_materialized_view(view, to_table, select_query, ...)
• drop_materialized_view(view, database=None, if_exists=True)
• create_distributed_table(table, cluster, local_table, ...)

Functions

The library provides comprehensive ClickHouse function coverage. Import from chpy.functions:

from chpy.functions import (
    # Aggregate
    count, sum, avg, min, max, quantile, stddevPop, stddevSamp,
    # String
    length, upper, lower, substring, concat, startsWith, endsWith,
    # Date/Time
    toYear, toMonth, toDayOfMonth, toHour, addDays, subtractDays,
    # Math
    abs, sqrt, round, floor, ceil,
    # Type conversion
    toString, toInt64, toFloat64, toDateTime,
    # Conditional
    if_ as if_func, coalesce,
    # Array
    array, arraySum, arrayAvg, arrayMax, arrayMin,
    # And many more...
)

All functions support:

• .alias(name) - Set column alias
• .over(window_spec) - Add OVER clause for window functions

WindowSpec

Window specification for OVER clauses.

from chpy.functions.base import WindowSpec

WindowSpec()
    .partition_by(column1, column2)
    .order_by(column3, desc=True)
    .rows_between("UNBOUNDED PRECEDING", "CURRENT ROW")

Methods

• partition_by(*columns) - Add PARTITION BY clause
• order_by(*columns, desc=False) - Add ORDER BY clause
• rows_between(start, end) - Add ROWS BETWEEN frame
• range_between(start, end) - Add RANGE BETWEEN frame
• to_sql() - Convert to SQL OVER clause

---

Best Practices

1. Use Schema Objects for Type Safety

# Good: Type-safe with autocomplete
df = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .to_dataframe())

# Avoid: Raw strings (no type checking)
df = (table.query()
    .where("pair = 'BTC-USDT'")
    .to_dataframe())

2. Use Method Chaining

# Good: Fluent interface
results = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.exchange == "BINANCE")
    .order_by(crypto_quotes.timestamp_ms, desc=True)
    .limit(100)
    .to_list())

# Avoid: Multiple variables
builder = table.query()
builder = builder.where(crypto_quotes.pair == "BTC-USDT")
builder = builder.where(crypto_quotes.exchange == "BINANCE")
results = builder.to_list()

3. Use Context Managers

# Good: Automatic connection management
with ClickHouseClient(...) as client:
    table = CryptoQuotesTable(client)
    results = table.query().to_list()

# Avoid: Manual connection management
client = ClickHouseClient(...)
try:
    table = CryptoQuotesTable(client)
    results = table.query().to_list()
finally:
    client.close()

4. Select Only Needed Columns

# Good: Select specific columns
df = (table.query()
    .select(crypto_quotes.pair, crypto_quotes.best_bid_price)
    .to_dataframe())

# Avoid: Selecting all columns when you only need a few
df = (table.query()
    .to_dataframe())  # Selects all columns

5. Use Appropriate Output Formats

# For data analysis: DataFrame
df = table.query().to_dataframe()

# For numerical computation: NumPy
arr = table.query().select(...).to_numpy()

# For iteration: List
results = table.query().to_list()

# For export: CSV/Parquet
table.query().to_csv("output.csv")
table.query().to_parquet("output.parquet")

6. Use Functions for Transformations

# Good: Use library functions
results = (table.query()
    .select(
        upper(crypto_quotes.pair).alias("pair_upper"),
        round(crypto_quotes.best_bid_price, 2).alias("rounded_price")
    )
    .to_list())

# Avoid: Raw SQL when functions are available
results = (table.query()
    .select("upper(pair) as pair_upper, round(best_bid_price, 2) as rounded_price")
    .to_list())

7. Use Window Functions for Advanced Analytics

# Good: Window functions for running calculations
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        avg(crypto_quotes.best_bid_price).over(
            WindowSpec().partition_by(crypto_quotes.pair)
        ).alias("avg_price")
    )
    .to_dataframe())

---

Table Schema

The crypto_quotes table has the following columns:

• pair (String): Trading pair (e.g., "BTC-USDT")
• best_bid_price (Float64): Best bid price
• best_bid_size (Float64): Best bid size
• best_ask_price (Float64): Best ask price
• best_ask_size (Float64): Best ask size
• bid_prices (Array(Float64)): Array of bid prices
• bid_sizes (Array(Float64)): Array of bid sizes
• ask_prices (Array(Float64)): Array of ask prices
• ask_sizes (Array(Float64)): Array of ask sizes
• timestamp_ms (UInt64): Timestamp in milliseconds
• exchange (LowCardinality(String)): Exchange name
• sequence_number (UInt64): Sequence number
• inserted_at (UInt64): Insert timestamp

---

License

MIT

---

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

---

Support

For issues, questions, or contributions, please open an issue on GitHub.

Basic Examples

Example 1: Simple Query with Column Objects

from chpy import ClickHouseClient
# crypto_quotes schema and CryptoQuotesTable are defined above (see Quick Start section)
from datetime import datetime, timedelta

client = ClickHouseClient(host="localhost", database="stockhouse")
table = CryptoQuotesTable(client)

# Query using column objects (type-safe with autocomplete)
df = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        crypto_quotes.best_ask_price,
        crypto_quotes.timestamp_ms
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.exchange == "BINANCE")
    .where(crypto_quotes.timestamp_ms >= datetime.now() - timedelta(days=1))
    .order_by(crypto_quotes.timestamp_ms, desc=True)
    .limit(100)
    .to_dataframe())

print(df.head())

Example 2: Using Table Shortcuts

# Use table.c or table.columns for column access
results = (table.query()
    .where(table.c.pair == "BTC-USDT")
    .where(table.c.best_bid_price > 50000)
    .order_by(table.c.timestamp_ms, desc=True)
    .limit(5)
    .to_list())

for row in results:
    print(f"Pair: {row['pair']}, Bid: {row['best_bid_price']}")

Example 3: Comparison Operators

# All comparison operators are supported
results = (table.query()
    .where(crypto_quotes.best_bid_price > 50000)
    .where(crypto_quotes.best_bid_price < 60000)
    .where(crypto_quotes.best_bid_price >= 51000)
    .where(crypto_quotes.best_bid_price <= 59000)
    .where(crypto_quotes.pair != "ETH-USDT")
    .limit(10)
    .to_list())

Example 4: IN and NOT IN Operators

# IN operator for multiple values
results = (table.query()
    .where(crypto_quotes.pair.in_(["BTC-USDT", "ETH-USDT", "BNB-USDT"]))
    .where(crypto_quotes.exchange == "BINANCE")
    .to_list())

# NOT IN operator
results = (table.query()
    .where(crypto_quotes.exchange.not_in(["BINANCE", "KUCOIN"]))
    .limit(10)
    .to_list())

Example 5: LIKE Operator

# Pattern matching with LIKE
results = (table.query()
    .where(crypto_quotes.pair.like("BTC-%"))
    .limit(10)
    .to_list())

Example 6: Complex Expressions with AND/OR

# Combine conditions with & (AND) and | (OR)
results = (table.query()
    .where(
        (crypto_quotes.pair == "BTC-USDT") & 
        (crypto_quotes.exchange.in_(["BINANCE", "KUCOIN"])) &
        (crypto_quotes.timestamp_ms >= datetime.now() - timedelta(days=1))
    )
    .limit(10)
    .to_list())

# OR expression
results = (table.query()
    .where(
        (crypto_quotes.pair == "BTC-USDT") | 
        (crypto_quotes.pair == "ETH-USDT")
    )
    .where(crypto_quotes.exchange == "BINANCE")
    .limit(10)
    .to_list())

Example 7: Multiple WHERE Clauses

# Multiple where() calls are combined with AND
results = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.exchange == "BINANCE")
    .where(crypto_quotes.best_bid_price > 50000)
    .where(crypto_quotes.best_ask_price < 60000)
    .limit(5)
    .to_list())

Example 8: Selecting Specific Columns

# Select only the columns you need
df = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        crypto_quotes.best_ask_price,
        crypto_quotes.exchange,
        crypto_quotes.timestamp_ms
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(10)
    .to_dataframe())

print(df.columns)  # Only selected columns

Example 9: Ordering Results

# Order by one or more columns
results = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .order_by(crypto_quotes.exchange, desc=False)  # ASC
    .order_by(crypto_quotes.timestamp_ms, desc=True)  # DESC
    .limit(10)
    .to_list())

Example 10: Counting Rows

# Count rows matching conditions
count = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.exchange == "BINANCE")
    .where(crypto_quotes.timestamp_ms >= datetime.now() - timedelta(days=1))
    .count())

print(f"Found {count} rows")

Example 11: Getting First Result

# Get the first matching row
first = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .order_by(crypto_quotes.timestamp_ms, desc=True)
    .first())

if first:
    print(f"Latest quote: {first['pair']} @ {first['best_bid_price']}")

Example 12: Checking Existence

# Check if any rows match
exists = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.exchange == "BINANCE")
    .exists())

print(f"Data exists: {exists}")

Example 13: Iterating Over Results

# Iterate over results (lazy evaluation)
for row in table.query().where(crypto_quotes.pair == "BTC-USDT").limit(10):
    print(f"Pair: {row['pair']}, Bid: {row['best_bid_price']}")

Example 14: Output Formats

# List of dictionaries
results = table.query().where(crypto_quotes.pair == "BTC-USDT").to_list()

# Pandas DataFrame
df = table.query().where(crypto_quotes.pair == "BTC-USDT").to_dataframe()

# NumPy array
arr = table.query().select(
    crypto_quotes.best_bid_price,
    crypto_quotes.best_ask_price
).to_numpy()

# JSON string
json_str = table.query().limit(10).to_json(indent=2)

# CSV string
csv_str = table.query().limit(100).to_csv()

# Write to file
table.query().limit(10000).to_csv("quotes.csv")
table.query().limit(10000).to_parquet("quotes.parquet")

# Dictionary (key-value mapping)
pair_dict = (table.query()
    .select(crypto_quotes.pair, crypto_quotes.best_bid_price)
    .to_dict(crypto_quotes.pair, crypto_quotes.best_bid_price))

---

Intermediate Examples

Example 15: Grouping and Aggregation

from chpy.functions import avg, count, min, max, sum

# Group by columns and aggregate
results = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.exchange,
        avg(crypto_quotes.best_bid_price).alias("avg_bid"),
        min(crypto_quotes.best_bid_price).alias("min_bid"),
        max(crypto_quotes.best_bid_price).alias("max_bid"),
        count().alias("cnt")
    )
    .where(crypto_quotes.timestamp_ms >= datetime.now() - timedelta(days=7))
    .group_by(crypto_quotes.pair, crypto_quotes.exchange)
    .having("avg_bid > 0")
    .order_by("avg_bid", desc=True)
    .limit(10)
    .to_list())

for row in results:
    print(f"{row['pair']} on {row['exchange']}: "
          f"avg={row['avg_bid']:.2f}, count={row['cnt']}")

Example 16: String Functions

from chpy.functions import length, upper, lower, substring, concat, startsWith, endsWith

# Use string functions in SELECT
results = (table.query()
    .select(
        crypto_quotes.pair,
        length(crypto_quotes.pair).alias("pair_length"),
        upper(crypto_quotes.pair).alias("pair_upper"),
        lower(crypto_quotes.exchange).alias("exchange_lower"),
        substring(crypto_quotes.pair, 1, 3).alias("pair_prefix"),
        concat(crypto_quotes.pair, " on ", crypto_quotes.exchange).alias("description")
    )
    .where(crypto_quotes.pair.in_(["BTC-USDT", "ETH-USDT"]))
    .limit(5)
    .to_list())

Example 17: Date and Time Functions

from chpy.functions import toYear, toMonth, toDayOfMonth, toHour, toDateTime, divide

# Extract date/time components
results = (table.query()
    .select(
        crypto_quotes.timestamp_ms,
        toYear(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))).alias("year"),
        toMonth(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))).alias("month"),
        toDayOfMonth(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))).alias("day"),
        toHour(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))).alias("hour")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(5)
    .to_list())

Example 18: Mathematical Functions

from chpy.functions import abs, sqrt, round, floor, ceil

# Mathematical transformations
results = (table.query()
    .select(
        crypto_quotes.best_bid_price,
        crypto_quotes.best_ask_price,
        abs(crypto_quotes.best_bid_price).alias("abs_bid"),
        sqrt(crypto_quotes.best_bid_price).alias("sqrt_bid"),
        round(crypto_quotes.best_bid_price, 2).alias("rounded_bid"),
        floor(crypto_quotes.best_bid_price).alias("floor_bid"),
        ceil(crypto_quotes.best_ask_price).alias("ceil_ask")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.best_bid_price > 0)
    .limit(5)
    .to_list())

Example 19: Conditional Functions

from chpy.functions import if_ as if_func, coalesce

# Conditional logic
results = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        crypto_quotes.best_ask_price,
        if_func(
            crypto_quotes.best_bid_price > 50000,
            "high",
            "normal"
        ).alias("price_category"),
        coalesce(crypto_quotes.best_bid_price, 0).alias("safe_bid")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(5)
    .to_list())

Example 20: Type Conversion Functions

from chpy.functions import toString, toInt64, toFloat64

# Convert types
results = (table.query()
    .select(
        crypto_quotes.pair,
        toString(crypto_quotes.best_bid_price).alias("bid_as_string"),
        toInt64(crypto_quotes.best_bid_price).alias("bid_as_int"),
        toFloat64(crypto_quotes.best_bid_price).alias("bid_as_float")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(3)
    .to_list())

Example 21: Combining Multiple Functions

# Combine different function types in one query
results = (table.query()
    .select(
        upper(crypto_quotes.pair).alias("pair_upper"),
        length(crypto_quotes.pair).alias("pair_length"),
        round(crypto_quotes.best_bid_price, 2).alias("rounded_price"),
        toYear(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))).alias("year"),
        if_func(
            crypto_quotes.best_bid_price > 50000,
            "premium",
            "standard"
        ).alias("tier")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(3)
    .to_list())

Example 22: Raw SQL Conditions

# Use raw SQL for complex conditions
results = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .where("best_bid_price > 50000 AND best_ask_price < 60000")
    .where("timestamp_ms >= toUnixTimestamp(now()) * 1000 - 86400000")
    .limit(10)
    .to_list())

Example 23: Working with Generic Tables

from chpy import ClickHouseClient, TableWrapper
from chpy.orm import Table, Column

# Create a schema for any table
columns = [
    Column("id", "UInt64"),
    Column("name", "String"),
    Column("value", "Float64"),
    Column("created_at", "DateTime")
]
schema = Table("my_table", "my_db", columns)

# Create wrapper with schema
client = ClickHouseClient(host="localhost", database="my_db")
table = TableWrapper(client, "my_table", "my_db", schema=schema)

# Query with type-safe columns
df = (table.query()
    .where(schema.id > 100)
    .where(schema.name == "test")
    .to_dataframe())

# Or query without schema (using raw strings)
df = (table.query()
    .where("id > 100")
    .where("name = 'test'")
    .to_dataframe())

# Insert data
table.insert([
    {"id": 1, "name": "test", "value": 1.5, "created_at": datetime.now()}
])

---

Advanced Examples

Example 24: JOIN Operations

from chpy.orm import Table, Column

# Define another table schema
other_table_columns = [
    Column("symbol", "String"),
    Column("name", "String"),
    Column("market_cap", "Float64"),
]
other_table = Table("market_data", "stockhouse", other_table_columns)

# INNER JOIN with column expressions
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        other_table.name,
        other_table.market_cap
    )
    .join(
        other_table,
        condition=(crypto_quotes.pair == other_table.symbol),
        join_type="INNER"
    )
    .where(crypto_quotes.exchange == "BINANCE")
    .to_dataframe())

# LEFT JOIN with alias
result = (table.query()
    .select(
        crypto_quotes.pair,
        other_table.market_cap
    )
    .join(
        other_table,
        condition=(crypto_quotes.pair == other_table.symbol),
        join_type="LEFT",
        alias="md"
    )
    .to_list())

# Multiple JOINs
result = (table.query()
    .join(other_table, condition=(crypto_quotes.pair == other_table.symbol))
    .join(
        "stockhouse.exchanges",
        condition="crypto_quotes.exchange = exchanges.code",
        join_type="LEFT"
    )
    .to_dataframe())

# CROSS JOIN (no condition needed)
result = (table.query()
    .join("stockhouse.reference_table", join_type="CROSS")
    .limit(100)
    .to_list())

Example 25: Window Functions

from chpy.functions.base import WindowSpec
from chpy.functions.window import rowNumber, rank, denseRank
from chpy.functions.aggregate import avg, sum

# Basic window function with PARTITION BY
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        avg(crypto_quotes.best_bid_price).over(
            WindowSpec().partition_by(crypto_quotes.pair)
        ).alias("avg_by_pair")
    )
    .to_dataframe())

# Window function with ORDER BY
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        rank().over(
            WindowSpec()
                .partition_by(crypto_quotes.pair)
                .order_by(crypto_quotes.best_bid_price, desc=True)
        ).alias("price_rank")
    )
    .to_dataframe())

# Running average with frame specification
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.timestamp_ms,
        crypto_quotes.best_bid_price,
        avg(crypto_quotes.best_bid_price).over(
            WindowSpec()
                .partition_by(crypto_quotes.pair)
                .order_by(crypto_quotes.timestamp_ms)
                .rows_between("UNBOUNDED PRECEDING", "CURRENT ROW")
        ).alias("running_avg")
    )
    .where(crypto_quotes.exchange == "BINANCE")
    .to_dataframe())

# Multiple window functions in same query
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.exchange,
        avg(crypto_quotes.best_bid_price).over(
            WindowSpec().partition_by(crypto_quotes.pair)
        ).alias("avg_by_pair"),
        rowNumber().over(
            WindowSpec()
                .partition_by(crypto_quotes.exchange)
                .order_by(crypto_quotes.timestamp_ms)
        ).alias("row_num")
    )
    .to_dataframe())

# Window functions with GROUP BY
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.exchange,
        avg(crypto_quotes.best_bid_price).over(
            WindowSpec().partition_by(crypto_quotes.pair)
        ).alias("window_avg"),
        avg(crypto_quotes.best_bid_price).alias("group_avg")
    )
    .where(crypto_quotes.exchange == "BINANCE")
    .group_by(crypto_quotes.pair, crypto_quotes.exchange)
    .to_dataframe())

Example 26: Subqueries

from chpy.orm import Subquery

# Scalar subquery in SELECT
subquery_builder = (table.query()
    .select(avg(crypto_quotes.best_bid_price))
    .where(crypto_quotes.pair == "BTC-USDT"))

result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        Subquery(subquery_builder).alias("avg_btc_price")
    )
    .where(crypto_quotes.pair == "ETH-USDT")
    .to_list())

# Subquery in WHERE with IN
subquery_builder = (table.query()
    .select(crypto_quotes.pair)
    .where(crypto_quotes.exchange == "BINANCE")
    .group_by(crypto_quotes.pair)
    .having("count() > 100"))

result = (table.query()
    .where(crypto_quotes.pair.in_(Subquery(subquery_builder)))
    .to_list())

# EXISTS subquery
subquery_builder = (table.query()
    .where(crypto_quotes.exchange == "BINANCE"))

result = (table.query()
    .where(Subquery.exists(subquery_builder))
    .to_list())

# Derived table (subquery in FROM)
subquery_builder = (table.query()
    .select(
        crypto_quotes.pair,
        avg(crypto_quotes.best_bid_price).alias("avg_price")
    )
    .group_by(crypto_quotes.pair))

result = (table.query()
    .from_subquery(Subquery(subquery_builder), alias="avg_prices")
    .where("avg_price > 50000")
    .to_list())

Example 27: DDL Operations - Creating Tables

from chpy import ClickHouseClient, DDL
from chpy.orm import Table, Column
from chpy import LowCardinality, Nullable, Array, Map, DateTime

client = ClickHouseClient(host="localhost", database="my_db")
ddl = DDL(client)

# Create table from schema
columns = [
    Column("id", "UInt64"),
    Column("name", LowCardinality("String")),
    Column("value", "Float64"),
    Column("tags", Array("String")),
    Column("metadata", Map("String", "String")),
    Column("created_at", DateTime("UTC")),
    Column("description", Nullable("String"))
]
schema = Table("my_table", "my_db", columns)

ddl.create_table(
    schema,
    engine="MergeTree",
    order_by="id",
    partition_by="toYYYYMM(created_at)",
    primary_key="id"
)

# Or create table with string name
ddl.create_table(
    "users",
    columns=columns,
    database="my_db",
    order_by="id"
)

# Create table with advanced options
ddl.create_table(
    schema,
    engine="MergeTree",
    order_by=["created_at", "id"],
    partition_by="toYYYYMM(created_at)",
    primary_key=["id"],
    settings={"index_granularity": 8192}
)

Example 28: DDL Operations - Altering Tables

# Add column
ddl.add_column(
    "my_db.my_table",
    Column("new_col", "String"),
    after="id"
)

# Drop column
ddl.drop_column("my_db.my_table", "old_col")

# Modify column
ddl.modify_column(
    "my_db.my_table",
    Column("name", "FixedString(100)")
)

# Rename table
ddl.rename_table("my_db.old_table", "new_table")

Example 29: DDL Operations - Databases and Views

# Create database
ddl.create_database("my_database", engine="Atomic")

# Drop database
ddl.drop_database("my_database")

# Create materialized view
target_columns = [
    Column("pair", "String"),
    Column("avg_price", "Float64"),
]
target_table = Table("mv_target", "my_db", target_columns)

ddl.create_materialized_view(
    "my_view",
    target_table,
    "SELECT pair, avg(price) as avg_price FROM source_table GROUP BY pair",
    database="my_db",
    order_by="pair"
)

# Drop materialized view
ddl.drop_materialized_view("my_db.my_view")

# Create distributed table
columns = [Column("id", "UInt64"), Column("name", "String")]
schema = Table("dist_table", "my_db", columns)

ddl.create_distributed_table(
    schema,
    cluster="my_cluster",
    local_table="my_db.local_table",
    sharding_key="rand()"
)

Example 30: ClickHouse Type System

from chpy import (
    # Type modifiers
    LowCardinality, Nullable, Array, Tuple, Map, Nested,
    FixedString, Enum,
    # Special types
    IPv4, IPv6, UUID, Date, DateTime, DateTime64,
    # Primitive types
    String, Bool, UInt8, UInt16, UInt32, UInt64, UInt128, UInt256,
    Int8, Int16, Int32, Int64, Int128, Int256,
    Float32, Float64,
    Decimal32, Decimal64, Decimal128, Decimal256,
    # Convenience functions
    LowCardinalityNullable, NullableArray, ArrayNullable
)
from chpy.orm import Column, Table

# Primitive types
Column("name", String)
Column("age", Int64)
Column("price", Float64)
Column("count", UInt64)
Column("is_active", Bool)

# LowCardinality for string optimization
Column("exchange", LowCardinality(String))
Column("name", LowCardinality(Nullable(String)))  # Nested types

# Nullable types
Column("description", Nullable(String))
Column("tags", Nullable(Array(String)))

# Array types
Column("tags", Array(String))
Column("prices", Array(Float64))
Column("nested_tags", Array(Nullable(String)))

# Tuple types
Column("coordinates", Tuple(Float64, Float64))
Column("metadata", Tuple(String, Int64, Float64))

# Map types
Column("settings", Map(String, String))
Column("counts", Map(String, Int64))

# Nested types
Column("user", Nested("name", String, "age", Int64))
# Or with tuples:
Column("user", Nested(("name", String), ("age", Int64)))

# FixedString
Column("code", FixedString(10))

# Enum types
Column("status", Enum("active", 1, "inactive", 0))
# Or with dict:
Column("status", Enum({"active": 1, "inactive": 0}))

# Special types
Column("ip_address", IPv4)
Column("ipv6_address", IPv6)
Column("user_id", UUID)
Column("created", Date)
Column("timestamp", DateTime("UTC"))
Column("precise_time", DateTime64(3, "UTC"))  # 3 decimal places

# Decimal types
Column("amount", Decimal64(2))  # Decimal64 with 2 decimal places
Column("precise_amount", Decimal128(4))  # Decimal128 with 4 decimal places

# Convenience functions
Column("name", LowCardinalityNullable(String))  # LowCardinality(Nullable(String))
Column("tags", NullableArray(String))  # Nullable(Array(String))
Column("items", ArrayNullable(String))  # Array(Nullable(String))

# Use in table definitions
columns = [
    Column("id", UInt64),
    Column("name", LowCardinality(String)),
    Column("tags", Array(String)),
    Column("metadata", Map(String, String)),
    Column("created_at", DateTime("UTC")),
]
table = Table("my_table", "my_db", columns)

Example 31: Row Objects with Schema

# When using a schema, results are returned as Row objects
# Row objects support both attribute and dictionary access

results = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(5)
    .to_list())

for row in results:
    # Attribute-style access (when schema is available)
    print(row.pair)  # or row['pair']
    print(row.best_bid_price)  # or row['best_bid_price']
    
    # Dictionary-style access
    print(row['pair'])
    print(row.get('pair', 'default'))
    
    # Convert to dict
    row_dict = row.to_dict()

Example 32: Complex Aggregations

from chpy.functions import (
    quantile, quantileExact, stddevPop, stddevSamp,
    argMin, argMax, topK, uniq
)

# Advanced aggregations
results = (table.query()
    .select(
        crypto_quotes.pair,
        avg(crypto_quotes.best_bid_price).alias("avg_price"),
        quantile(0.5)(crypto_quotes.best_bid_price).alias("median_price"),
        quantile(0.95)(crypto_quotes.best_bid_price).alias("p95_price"),
        stddevPop(crypto_quotes.best_bid_price).alias("stddev"),
        argMin(crypto_quotes.timestamp_ms, crypto_quotes.best_bid_price).alias("min_price_time"),
        topK(5)(crypto_quotes.exchange).alias("top_exchanges"),
        uniq(crypto_quotes.exchange).alias("unique_exchanges")
    )
    .where(crypto_quotes.timestamp_ms >= datetime.now() - timedelta(days=7))
    .group_by(crypto_quotes.pair)
    .to_list())

Example 33: Array Functions

from chpy.functions import (
    array, arraySum, arrayAvg, arrayMax, arrayMin,
    arrayElement, has, hasAll, hasAny, indexOf
)

# Working with array columns
results = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.bid_prices,
        arraySum(crypto_quotes.bid_prices).alias("total_bids"),
        arrayAvg(crypto_quotes.bid_prices).alias("avg_bid"),
        arrayMax(crypto_quotes.bid_prices).alias("max_bid"),
        arrayMin(crypto_quotes.bid_prices).alias("min_bid"),
        arrayElement(crypto_quotes.bid_prices, 1).alias("first_bid"),
        has(crypto_quotes.bid_prices, 50000).alias("has_50k")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(5)
    .to_list())

Example 34: Time Series Analysis

from chpy.functions import (
    toStartOfHour, toStartOfDay, toStartOfWeek,
    addDays, subtractDays, dateDiff
)

# Time-based aggregations
results = (table.query()
    .select(
        toStartOfHour(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))).alias("hour"),
        crypto_quotes.pair,
        avg(crypto_quotes.best_bid_price).alias("avg_price"),
        min(crypto_quotes.best_bid_price).alias("min_price"),
        max(crypto_quotes.best_bid_price).alias("max_price"),
        count().alias("quote_count")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.timestamp_ms >= datetime.now() - timedelta(days=7))
    .group_by(
        toStartOfHour(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))),
        crypto_quotes.pair
    )
    .order_by("hour", desc=True)
    .to_dataframe())

Example 35: Inserting Data

# Insert data into table
data = [
    {
        "pair": "BTC-USDT",
        "best_bid_price": 50000.0,
        "best_bid_size": 1.5,
        "best_ask_price": 50001.0,
        "best_ask_size": 2.0,
        "timestamp_ms": int(datetime.now().timestamp() * 1000),
        "exchange": "BINANCE",
        "sequence_number": 1,
        "inserted_at": int(datetime.now().timestamp() * 1000)
    },
    # ... more rows
]

table.insert(data)

# Or use the client directly
client.insert("stockhouse.crypto_quotes", data)

---