timescaledb 0.0.7

TimescaleDB is a Python Client based on SQLModel and SQLAlchemy for high-performance real-time analytics time-series data.

TimescaleDB for Python

Python client for TimescaleDB — the open-source time-series database built on PostgreSQL. This package is built on SQLModel and SQLAlchemy and is designed to be used with FastAPI, Flask, and any other SQLAlchemy-based project.

It gives you Python helpers for the things you actually do with TimescaleDB: creating hypertables, enabling the Hypercore columnstore (and legacy compression), setting retention policies, building continuous aggregates, and running time_bucket / time_bucket_gapfill queries.

Looking for Django? Check out django-timescaledb.

• Supports: Python 3.11, 3.12, 3.13, and 3.14
• Targets: TimescaleDB 2.x (Hypercore columnstore needs 2.18+; direct-create hypertables need 2.20+; generated aggregate columns need 2.28+)
• License: MIT

Contents

• Requirements
• Installation
• Quickstart
• Creating a hypertable
  • Automatically via TimescaleModel
  • Manually via create_hypertable
  • Direct hypertable creation (2.20+)
• Hypercore columnstore (2.18+)
• Compression (legacy)
• Retention policies
• Continuous aggregates
• Querying with time_bucket
• Sample projects
• FastAPI example
• Limitations & status
• Contributing
• Used by

Requirements

• Python: 3.11, 3.12, 3.13, or 3.14.

• PostgreSQL: a PostgreSQL server with the TimescaleDB extension installed (the official timescale/timescaledb Docker images bundle both). The package targets PostgreSQL 15+ in CI.

• TimescaleDB: 2.x. Some features require newer releases:

  Feature	Minimum TimescaleDB
  Hypertables, compression, retention, continuous aggregates	2.x
  Hypercore columnstore (enable_columnstore, add_columnstore_policy, …)	2.18+
  Direct CREATE TABLE ... WITH (tsdb.hypertable) (create_table_with_hypertable)	2.20+
  Generated aggregate columns on continuous aggregates (add_generated_aggregate_column)	2.28+

• A PostgreSQL driver: any SQLAlchemy-compatible driver — psycopg (psycopg 3), psycopg2, or asyncpg. See Installation.

Installation

pip install timescaledb

You also need a PostgreSQL driver. Any SQLAlchemy-compatible driver works — psycopg2, psycopg (v3), or asyncpg:

pip install "psycopg[binary]"   # recommended

The package registers timescaledb SQLAlchemy dialects, so connection URLs such as timescaledb://, timescaledb+psycopg://, and timescaledb+asyncpg:// are available in addition to the standard postgresql:// URLs.

Optional dependencies

The core install is intentionally lightweight — it only depends on SQLModel (plus the PostgreSQL driver you choose). FastAPI and uvicorn are not required to use the library; they are only needed for the example apps. Install them via the fastapi extra:

pip install "timescaledb[fastapi]"

This pulls in FastAPI + uvicorn so you can run the example FastAPI apps (see samples/fastapi_timeseries_api and sample_project/).

Quickstart

from sqlmodel import Field, Session, SQLModel, select

import timescaledb
from timescaledb import TimescaleModel

DATABASE_URL = "postgresql://user:password@localhost:5432/timescaledb"

# create_engine pins the connection timezone (defaults to "UTC")
engine = timescaledb.create_engine(DATABASE_URL, timezone="UTC")


class Metric(TimescaleModel, table=True):
    # TimescaleModel already provides `id` and a `time` column
    sensor_id: int = Field(index=True)
    value: float


# 1. Create the regular tables
SQLModel.metadata.create_all(engine)
# 2. Convert TimescaleModel tables into hypertables (+ any policies)
timescaledb.metadata.create_all(engine)

with Session(engine) as session:
    session.add(Metric(sensor_id=1, value=42.0))
    session.commit()

    results = timescaledb.time_bucket_query(
        session,
        Metric,
        interval="1 hour",
        metric_field="value",
    )
    print(results)

TimescaleModel supplies the id primary key and a timezone-aware time column for you, so a model only needs its own fields.

Creating a hypertable

There are three ways to turn a table into a hypertable. Pick one:

1. Automatically with TimescaleModel + timescaledb.metadata.create_all — least code, configured with class variables.
2. Manually with create_hypertable on any table that has a time column.
3. Directly with create_table_with_hypertable (TimescaleDB 2.20+), which creates the table as a hypertable in a single statement.

Automatically via TimescaleModel

from sqlmodel import Field, Session, SQLModel

import timescaledb
from timescaledb import TimescaleModel

DATABASE_URL = "postgresql://user:password@localhost:5432/timescaledb"
engine = timescaledb.create_engine(DATABASE_URL, timezone="UTC")


class SensorReading(TimescaleModel, table=True):
    sensor_id: int = Field(index=True)
    value: float

    # __time_column__ = "time"  # already set by TimescaleModel
    __chunk_time_interval__ = "INTERVAL 7 days"
    __drop_after__ = "INTERVAL 1 year"
    __enable_compression__ = True
    __compress_orderby__ = "time DESC"
    __compress_segmentby__ = "sensor_id"
    __migrate_data__ = True
    __if_not_exists__ = True


# Create the tables, then the hypertables + compression + retention policies
SQLModel.metadata.create_all(engine)
timescaledb.metadata.create_all(engine)

timescaledb.metadata.create_all(engine) walks every TimescaleModel subclass, creates the hypertable, and applies whatever compression, columnstore, and retention settings the model opts into.

Database drivers (SQLAlchemy dialects)

timescaledb registers timescaledb-scheme SQLAlchemy dialects so you can make the TimescaleDB backend explicit in your connection URL. Each one is a thin subclass of the matching PostgreSQL driver, so behavior is identical to PostgreSQL apart from the URL scheme:

URL scheme	Driver	Dialect
timescaledb://	psycopg2 (default)	TimescaledbPsycopg2Dialect
timescaledb+psycopg2://	psycopg2	TimescaledbPsycopg2Dialect
timescaledb+psycopg://	psycopg (psycopg 3)	TimescaledbPsycopgDialect
timescaledb+asyncpg://	asyncpg	TimescaledbAsyncpgDialect

import timescaledb

# psycopg (psycopg 3)
engine = timescaledb.create_engine(
    "timescaledb+psycopg://user:password@localhost:5432/timescaledb"
)

Install the driver you intend to use, e.g. pip install "psycopg[binary]" for psycopg 3, pip install psycopg2-binary for psycopg2, or pip install asyncpg for asyncpg. Plain postgresql:// URLs continue to work unchanged.

Manually via create_hypertable

Use this on a plain SQLModel table (or any existing table) that has a time column. It gives you the most direct control over each step:

from sqlmodel import Field, Session, SQLModel
from datetime import datetime

import timescaledb

DATABASE_URL = "postgresql://user:password@localhost:5432/timescaledb"
engine = timescaledb.create_engine(DATABASE_URL)


class Sensor(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True)
    time: datetime = Field(default=None, primary_key=True)
    sensor_id: int = Field(index=True)
    value: float

    __tablename__ = "my_time_series_table"


hypertable_options = {
    "time_column": "time",
    "compress_orderby": "time DESC",
    "compress_segmentby": "sensor_id",
    "chunk_time_interval": "7 days",
    "drop_after": "1 year",
    "migrate_data": True,
    "if_not_exists": True,
}

table_name = "my_time_series_table"

with Session(engine) as session:
    # Create the table in the database
    SQLModel.metadata.create_all(engine)

    # Create the hypertable
    timescaledb.create_hypertable(
        session,
        commit=True,
        table_name=table_name,
        hypertable_options=hypertable_options,
    )

    # Enable compression
    timescaledb.enable_table_compression(
        session,
        commit=True,
        table_name=table_name,
        compress_orderby=hypertable_options["compress_orderby"],
        compress_segmentby=hypertable_options["compress_segmentby"],
    )
    # Compress chunks once they age past the chunk interval
    timescaledb.add_compression_policy(
        session,
        commit=True,
        table_name=table_name,
        compress_after=hypertable_options["chunk_time_interval"],
    )
    # Drop chunks after the retention window
    timescaledb.add_retention_policy(
        session,
        table_name=table_name,
        drop_after=hypertable_options["drop_after"],
    )

Direct hypertable creation (2.20+)

TimescaleDB 2.20+ can create a table as a hypertable in one statement with CREATE TABLE ... WITH (tsdb.hypertable). For brand-new tables, compile and run that SQL straight from a model:

from sqlmodel import Session

import timescaledb

with Session(engine) as session:
    timescaledb.create_table_with_hypertable(
        session,
        SensorReading,
        chunk_interval="7 days",
    )

Use timescaledb.format_create_table_with_hypertable_sql(...) if you just want the SQL string without executing it.

Hypercore columnstore (2.18+)

TimescaleDB 2.18 introduced the Hypercore columnstore API. This package supports the modern columnstore path (enable_columnstore, add_columnstore_policy, convert_to_columnstore / convert_to_rowstore) while keeping the older compression helpers available.

from sqlmodel import Session

import timescaledb

with Session(engine) as session:
    timescaledb.enable_columnstore(
        session,
        table_name="my_time_series_table",
        orderby="time DESC",
        segmentby="sensor_id",
    )
    timescaledb.add_columnstore_policy(
        session,
        table_name="my_time_series_table",
        after="60 days",
        if_not_exists=True,
    )

You can opt in from a TimescaleModel instead. timescaledb.metadata.create_all then enables columnstore and adds the policy automatically:

from sqlmodel import Field

from timescaledb import TimescaleModel


class SensorReading(TimescaleModel, table=True):
    sensor_id: int = Field(index=True)
    value: float

    __enable_columnstore__ = True
    __columnstore_orderby__ = "time DESC"
    __columnstore_segmentby__ = "sensor_id"
    __columnstore_after__ = "60 days"

Available columnstore class variables: __enable_columnstore__, __columnstore_orderby__, __columnstore_segmentby__, __columnstore_after__, __columnstore_created_before__, __columnstore_if_not_exists__, __columnstore_schedule_interval__, and __columnstore_timezone__.

Manual chunk conversion and policy inspection are also available via convert_to_columnstore, convert_to_rowstore, list_columnstore_policies, remove_columnstore_policy, and sync_columnstore_policies.

Compression (legacy)

The pre-Hypercore compression helpers (enable_table_compression, add_compression_policy, sync_compression_policies) remain fully supported for existing code and older TimescaleDB versions. On TimescaleDB 2.18+, prefer the Hypercore columnstore API for new work. See the manual hypertable example above for usage.

Retention policies

Drop chunks automatically once they age past a window:

timescaledb.add_retention_policy(
    session,
    table_name="my_time_series_table",
    drop_after="1 year",
)

Or opt in from a model with __drop_after__ and let timescaledb.metadata.create_all apply it. Use sync_retention_policies to reconcile policies across all opted-in models.

Continuous aggregates

Continuous aggregates can be created, scheduled, refreshed, and extended from a SQLModel session:

from datetime import datetime, timezone
from sqlmodel import Session

import timescaledb

with Session(engine) as session:
    timescaledb.create_continuous_aggregate(
        session,
        "conditions_summary_hourly",
        """
        SELECT time_bucket('1 hour', time) AS bucket, avg(temp) AS avg_temp
        FROM conditions
        GROUP BY bucket
        """,
        with_data=False,
    )
    timescaledb.add_continuous_aggregate_policy(
        session,
        "conditions_summary_hourly",
        start_offset="1 month",
        end_offset="1 hour",
        schedule_interval="1 hour",
        buckets_per_batch=10,
        refresh_newest_first=True,
    )
    timescaledb.refresh_continuous_aggregate(
        session,
        "conditions_summary_hourly",
        window_start=datetime(2026, 1, 1, tzinfo=timezone.utc),
        window_end=datetime(2026, 2, 1, tzinfo=timezone.utc),
        force=True,
    )
    # TimescaleDB 2.28+: add a generated aggregate column without a full rebuild
    timescaledb.add_generated_aggregate_column(
        session,
        "conditions_summary_hourly",
        "max_temp",
        "DOUBLE PRECISION",
        "max(temp)",
    )

Remove a refresh policy with remove_continuous_aggregate_policy. The newer policy options — buckets_per_batch, max_batches_per_execution, refresh_newest_first, and include_tiered_data — are all supported.

Querying with time_bucket

Two helpers wrap the most common time-series read patterns and return a list of {"bucket": ..., "avg": ...} mappings.

time_bucket_query buckets rows by an interval and aggregates a metric field:

rows = timescaledb.time_bucket_query(
    session,
    Metric,
    interval="1 hour",
    time_field="time",
    metric_field="value",
)

time_bucket_gapfill_query fills gaps in a bounded time range, with optional LOCF (last observation carried forward) or interpolation:

from datetime import datetime, timezone

rows = timescaledb.time_bucket_gapfill_query(
    session,
    Metric,
    interval="1 hour",
    metric_field="value",
    start=datetime(2026, 1, 1, tzinfo=timezone.utc),
    finish=datetime(2026, 1, 2, tzinfo=timezone.utc),
    use_locf=True,        # or use_interpolate=True
)

Both accept a filters list of SQLAlchemy conditions for narrowing the query.

Sample projects

The samples/ directory has ten self-contained, fully tested example projects, each focused on a different TimescaleDB feature. Every sample runs against TimescaleDB in Docker and ships with a pytest suite that spins up a throwaway container automatically via testcontainers.

#	Project	Highlights
01	iot_sensor_network	TimescaleModel, create_hypertable, time_bucket_query, last-point query
02	devops_metrics_gapfill	time_bucket_gapfill_query with gapfill, LOCF, and interpolation
03	crypto_ohlcv_candles	first()/last() + time_bucket → OHLCV candlesticks
04	energy_metering_compression	native compression + measuring the ratio
05	hypercore_columnstore	Hypercore columnstore (2.18+)
06	ecommerce_clickstream_retention	retention policy + funnel rollups
07	fleet_gps_tracking	manual create_hypertable path + downsampling
08	continuous_aggregates_rollups	hierarchical continuous aggregates (hourly → daily)
09	fastapi_timeseries_api	a FastAPI REST API over a hypertable, tested with TestClient
10	weather_lifecycle_full	capstone: hypertable + columnstore + retention + continuous aggregate + gapfill

See samples/README.md for setup and how to run the suites. There is also a minimal end-to-end FastAPI app in sample_project/.

FastAPI example

A minimal FastAPI app over a hypertable. The pattern mirrors sample_project/.

models.py

from datetime import datetime

from sqlmodel import Field, SQLModel

from timescaledb import TimescaleModel


class Metric(TimescaleModel, table=True):
    temp: float

    __enable_compression__ = True
    __chunk_time_interval__ = "2 weeks"
    __drop_after__ = "1 year"


class MetricCreate(SQLModel):
    temp: float


class MetricRead(SQLModel):
    id: int
    temp: float
    time: datetime

database.py

from sqlmodel import Session, SQLModel

import timescaledb

DATABASE_URL = "postgresql://user:password@localhost:5432/timescaledb"

engine = timescaledb.create_engine(DATABASE_URL, timezone="UTC", echo=False)


def get_session():
    with Session(engine) as session:
        yield session


def init_db():
    # Create all tables that inherit from SQLModel
    SQLModel.metadata.create_all(engine)
    # Create hypertables (+ policies) for all TimescaleModel subclasses
    timescaledb.metadata.create_all(engine)

main.py

from contextlib import asynccontextmanager

from fastapi import Depends, FastAPI, HTTPException
from sqlmodel import Session, select

from .database import get_session, init_db
from .models import Metric, MetricCreate, MetricRead


@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/metrics/", response_model=MetricRead)
def create_metric(metric: MetricCreate, session: Session = Depends(get_session)):
    db_metric = Metric.model_validate(metric)
    session.add(db_metric)
    session.commit()
    session.refresh(db_metric)
    return db_metric


@app.get("/metrics/{metric_id}", response_model=MetricRead)
def read_metric(metric_id: int, session: Session = Depends(get_session)):
    metric = session.get(Metric, metric_id)
    if not metric:
        raise HTTPException(status_code=404, detail="Metric not found")
    return metric


@app.get("/metrics/", response_model=list[MetricRead])
def list_metrics(session: Session = Depends(get_session)):
    return session.exec(select(Metric)).all()

timescaledb.create_engine wraps sqlmodel.create_engine (itself a wrapper around sqlalchemy.create_engine) and pins the connection timezone for you.

Limitations & status

• Beta. The package is in the 0.0.x series — the public API is still settling and may change between releases. Pin a version if you need stability.
• Helpers are synchronous. The timescaledb.asyncpg dialect is registered so you can use a timescaledb+asyncpg:// URL with raw/async SQLAlchemy, but the helper functions in this package (create_hypertable, time_bucket_query, enable_columnstore, the continuous-aggregate helpers, etc.) are all synchronous and operate on a SQLModel/SQLAlchemy Session. There is no async helper API yet.

Contributing

Contributions are welcome. See CONTRIBUTING.md for how to set up a dev environment, run the test suite (Docker + testcontainers), run lint/mypy, and the release process. For runnable, end-to-end examples of every feature, see the samples/ directory.

Used by

• analytics-api — complete tutorial project for building an Analytics API using FastAPI + TimescaleDB.

---

For a summary of recent upstream TimescaleDB changes and how they map onto this package, see docs/timescale-recent-updates.md.