ClickHouse¶

from datetime import date
from sqlalchemy import func, Date
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from dbwarden.databases.clickhouse import (
    CHTableMeta, CHViewMeta, ch_table, ch,
    merge_tree, materialized_view, kafka, aggregating_view, agg,
    AggregatingView, MaterializedView,
)

class Base(DeclarativeBase):
    pass

# 1. Source table
class Event(Base):
    __tablename__ = "events"

    id: Mapped[int] = mapped_column(primary_key=True)
    event_date: Mapped[date] = mapped_column()
    amount: Mapped[float] = mapped_column()

    class Meta(CHTableMeta):
        ch = ch_table(
            engine=merge_tree(),
            order_by=["event_date", "id"],
            partition_by=func.toYYYYMM(Event.event_date),
        )

# 2. Materialized view: Mode A (class IS the target, MV is auto-generated)
class EventDaily(MaterializedView):
    __tablename__ = "event_daily"

    date: Mapped[date] = mapped_column(primary_key=True)
    total: Mapped[float] = mapped_column()
    cnt: Mapped[int] = mapped_column()

    class Meta(CHViewMeta):
        ch = materialized_view(
            select="SELECT event_date AS date, sum(amount) AS total, "
                   "count(*) AS cnt FROM events GROUP BY event_date",
            engine=merge_tree(),
            order_by=["date"],
        )

# 3. Aggregating view: sources from EventDaily model
class EventAggregated(AggregatingView):
    __tablename__ = "event_aggregated"

    class Meta(CHViewMeta):
        ch = aggregating_view(
            source=EventDaily,
            group_by=[EventDaily.date],
            aggregates=[
                agg.sum(EventDaily.total, "Float64").as_("state"),
            ],
            order_by=[EventDaily.date],
        )

from dbwarden import database_config
from dbwarden.databases.clickhouse import (
    NamedCollectionSpec, named_collection,
    ChRoleSpec, ChUserSpec, ChRowPolicySpec,
    ChQuotaSpec, ChSettingsProfileSpec, ChGrantSpec,
)

database_config(
    database_name="analytics",
    database_type="clickhouse",
    database_url_sync="clickhouse://...",
    ch_named_collections=[...],       # list[NamedCollectionSpec | dict]
    ch_roles=[...],                   # list[ChRoleSpec | dict]
    ch_users=[...],                   # list[ChUserSpec | dict]
    ch_row_policies=[...],            # list[ChRowPolicySpec | dict]
    ch_quotas=[...],                  # list[ChQuotaSpec | dict]
    ch_settings_profiles=[...],       # list[ChSettingsProfileSpec | dict]
    ch_grants=[...],                  # list[ChGrantSpec | dict]
)

class PageView(Base):
    __tablename__ = "page_views"

    id: Mapped[int] = mapped_column(primary_key=True)
    url: Mapped[str] = mapped_column()
    user_id: Mapped[int] = mapped_column()
    event_time: Mapped[datetime] = mapped_column()
    duration_ms: Mapped[int] = mapped_column()

    class Meta(CHTableMeta):
        ch = ch_table(
            engine=replicated_merge_tree("/zk/pv", "{replica}"),
            order_by=["event_time", "user_id"],
            partition_by="toYYYYMM(event_time)",
            ttl=["event_time + toIntervalMonth(6)"],
            settings={"index_granularity": 4096},
        )

from datetime import datetime
from sqlalchemy.orm import Mapped, mapped_column
from dbwarden.databases.clickhouse import (
    CHColumnMeta, CHTableMeta, ch_table, ch,
    merge_tree,
)

class SensorReading(Base):
    __tablename__ = "sensor_readings"

    sensor_id: Mapped[str] = mapped_column()
    ts: Mapped[datetime] = mapped_column()
    temp: Mapped[float] = mapped_column()
    humidity: Mapped[float] = mapped_column()

    class Meta(CHTableMeta):
        ch = ch_table(
            engine=merge_tree(),
            order_by=["sensor_id", "ts"],
        )

        class temp(CHColumnMeta):
            ch = ch.field(codec="ZSTD(5)")

        class ts(CHColumnMeta):
            ch = ch.field(codec="DoubleDelta", ttl="ts + toIntervalDay(90)")

        class humidity(CHColumnMeta):
            ch = ch.field(ttl="ts + toIntervalDay(30)")

class KafkaEvents(Base):
    __tablename__ = "kafka_events"

    payload: Mapped[str] = mapped_column()

    class Meta(CHTableMeta):
        ch = ch_table(
            engine=kafka(
                named_collection="kafka_prod",
                topic="raw_events",
                format="JSONEachRow",
                group_name="dbwarden",
            ),
        )

class ParsedEvents(MaterializedView):
    __tablename__ = "parsed_events"

    class Meta(CHViewMeta):
        ch = materialized_view(
            select="""
                SELECT
                    JSONExtractString(payload, 'type') AS event_type,
                    JSONExtractFloat(payload, 'value') AS value,
                    JSONExtractDateTime(payload, 'ts') AS ts
                FROM kafka_events
            """,
            to="parsed_events_dest",
        )

database_config(
    database_name="analytics",
    database_type="clickhouse",
    database_url_sync="clickhouse://localhost:9000",
    ch_named_collections=[
        named_collection("ldap_auth", ldap_server="ldap.example.com"),
    ],
    ch_roles=[ChRoleSpec("analyst"), ChRoleSpec("engineer")],
    ch_users=[
        ChUserSpec(
            name="bob",
            named_collection="ldap_auth",
            default_role="analyst",
        ),
    ],
    ch_grants=[
        ChGrantSpec(privileges=["SELECT"], on="analytics.*", to="analyst"),
    ],
)

from datetime import datetime
from sqlalchemy.orm import Mapped, mapped_column
from dbwarden.databases.clickhouse import (
    CHTableMeta, ch_table, s3, projection,
)

class S3Logs(Base):
    __tablename__ = "s3_logs"

    ts: Mapped[datetime] = mapped_column()
    level: Mapped[str] = mapped_column()
    message: Mapped[str] = mapped_column()

    class Meta(CHTableMeta):
        ch = ch_table(
            engine=s3(
                named_collection="s3_logs",
                path="logs/*.parquet",
                format="Parquet",
            ),
            projections=[
                projection(
                    name="by_level",
                    query="SELECT level, count() GROUP BY level",
                ),
            ],
        )

from dbwarden.databases.clickhouse import data_op

# After adding a projection to a table with existing data:
data_op(
    name="materialize_daily_agg",
    forward="ALTER TABLE events MATERIALIZE PROJECTION daily_agg",
)