TimescaleDB with Python

https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view
Admin
A comprehensive guide for integrating TimescaleDB with Python applications to efficiently manage and...
Tokens:16,165
Snippets:105
Trust Score:8.5
Update:6 months ago
Show doc for...
Tokens:
Raw
### Run TimescaleDB Container with Docker

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Starts a TimescaleDB container with persistent storage. Ensure Docker is installed and running.

```Bash
# Pull and run TimescaleDB container

# -d: Run in detached mode
# -p: Map port 5432 to host
# --name: Container name for easy reference
# -v: Persist data across container restarts
docker run -d \
  --name timescaledb \
  -p 5432:5432 \
  -e POSTGRES_PASSWORD=password \
  -v timescale_data:/var/lib/postgresql/data \
  timescaledb/timescaledb:latest-pg15
```

--------------------------------

### IoT Monitoring Service Example

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Demonstrates initializing a service, registering sensors, recording readings, and retrieving data summaries and trends from a TimescaleDB database using Python.

```python
import asyncio
from typing import Dict, List, Optional, Tuple

import asyncpg


class SensorConfig:
    def __init__(self, sensor_id: int, name: str, location: str, alert_thresholds: Dict[str, Tuple[float, float]]):
        self.sensor_id = sensor_id
        self.name = name
        self.location = location
        self.alert_thresholds = alert_thresholds


class IoTMonitoringService:
    def __init__(self, db_url: str):
        self.db_url = db_url
        self.pool: Optional[asyncpg.Pool] = None
        self.sensors: Dict[int, SensorConfig] = {}

    async def initialize(self):
        """Initialize database connection pool"""
        self.pool = await asyncpg.create_pool(self.db_url)

    async def register_sensor(self, config: SensorConfig):
        """Register a new sensor with its configuration"""
        self.sensors[config.sensor_id] = config
        # In a real application, you might also store sensor config in the database

    async def record_reading(self, sensor_id: int, **kwargs):
        """Record a sensor reading into the database"""
        if not self.pool:
            raise Exception("Database not initialized")
        if sensor_id not in self.sensors:
            raise ValueError(f"Sensor ID {sensor_id} not registered")

        # Prepare data for insertion
        data = {"sensor_id": sensor_id, **kwargs}
        columns = ", ".join(data.keys())
        values_placeholder = ", ".join([f"${i+1}" for i in range(len(data))])

        query = f"INSERT INTO sensor_readings ({columns}) VALUES ({values_placeholder})"

        async with self.pool.acquire() as conn:
            await conn.execute(query, *data.values())

        # Check for alerts (simplified)
        await self.check_alerts(sensor_id, **kwargs)

    async def check_alerts(self, sensor_id: int, **readings):
        """Check sensor readings against alert thresholds"""
        if sensor_id not in self.sensors:
            return

        config = self.sensors[sensor_id]
        for key, value in readings.items():
            if key in config.alert_thresholds:
                min_thresh, max_thresh = config.alert_thresholds[key]
                if not (min_thresh <= value <= max_thresh):
                    # In a real app, trigger an alert notification
                    print(f"ALERT: Sensor {sensor_id} ({config.name}) - {key} out of bounds: {value} (Threshold: {min_thresh}-{max_thresh})")
                    # Optionally record the alert to a separate table
                    if self.pool:
                        await self.pool.execute(
                            "INSERT INTO sensor_alerts (sensor_id, alert_type, message, value, threshold) VALUES ($1, $2, $3, $4, $5)",
                            sensor_id, key, f"{key} out of bounds", value, f"{min_thresh}-{max_thresh}"
                        )

    async def get_sensor_summary(self, sensor_id: int, hours: int = 24) -> List[Dict]:
        """Get the latest summary data for a sensor"""
        if not self.pool:
            raise Exception("Database not initialized")

        async with self.pool.acquire() as conn:
            rows = await conn.fetch(
                """SELECT time, sensor_id, temperature, humidity, pressure, battery, rssi
                FROM sensor_readings
                WHERE sensor_id = $1 AND time > NOW() - INTERVAL $2 * INTERVAL '1 hour'
                ORDER BY time DESC
                LIMIT 1""",
                sensor_id, hours
            )
            return [dict(row) for row in rows]

    async def get_hourly_trends(self, sensor_id: int, days: int = 7) -> List[Dict]:
        """Get hourly trend data for a sensor"""
        if not self.pool:
            raise Exception("Database not initialized")

        async with self.pool.acquire() as conn:
            # Example using TimescaleDB continuous aggregates or downsampling would be more efficient here
            # For simplicity, this query aggregates hourly data directly
            rows = await conn.fetch(
                """SELECT
                    date_trunc('hour', time) as hour,
                    AVG(temperature) as avg_temperature,
                    AVG(humidity) as avg_humidity,
                    AVG(pressure) as avg_pressure,
                    AVG(battery) as avg_battery
                FROM sensor_readings
                WHERE sensor_id = $1 AND time > NOW() - INTERVAL $2 * INTERVAL '1 day'
                GROUP BY hour
                ORDER BY hour DESC""",
                sensor_id, days
            )
            return [dict(row) for row in rows]

    async def get_recent_alerts(self, sensor_id: Optional[int] = None, hours: int = 24) -> List[Dict]:
        """Get recent alerts, optionally filtered by sensor_id"""
        if not self.pool:
            raise Exception("Database not initialized")

        async with self.pool.acquire() as conn:
            if sensor_id:
                rows = await conn.fetch(
                    """SELECT time, sensor_id, alert_type, message, value, threshold
                    FROM sensor_alerts
                    WHERE sensor_id = $1 AND time > NOW() - INTERVAL '1 hour' * $2
                    ORDER BY time DESC
                    LIMIT 100""", sensor_id, hours
                )
            else:
                rows = await conn.fetch(
                    """SELECT time, sensor_id, alert_type, message, value, threshold
                    FROM sensor_alerts
                    WHERE time > NOW() - INTERVAL '1 hour' * $1
                    ORDER BY time DESC
                    LIMIT 100""", hours
                )

            return [dict(row) for row in rows]

    async def close(self):
        """Close database connections"""
        if self.pool:
            await self.pool.close()

# Example usage
async def main():
    # Initialize service
    service = IoTMonitoringService(
        "postgresql://postgres:password@localhost:5432/timeseries_db"
    )
    await service.initialize()

    # Register sensors with thresholds
    service.register_sensor(SensorConfig(
        sensor_id=1,
        name="Warehouse Sensor A",
        location="warehouse_a",
        alert_thresholds={
            'temperature': (15, 30),  # Alert if outside 15-30C
            'humidity': (30, 70),     # Alert if outside 30-70%
            'battery': (20, 100)      # Alert if battery below 20%
        }
    ))

    # Record a reading
    await service.record_reading(
        sensor_id=1,
        temperature=25.5,
        humidity=55.0,
        pressure=1013.25,
        battery=85.0,
        rssi=-65
    )

    # Get summary
    summary = await service.get_sensor_summary(1, hours=24)
    print(f"Sensor summary: {summary}")

    # Get trends
    trends = await service.get_hourly_trends(1, days=7)
    print(f"Found {len(trends)} hourly data points")

    await service.close()

if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Setup Database Schema with TimescaleDB

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Creates essential tables for sensor readings and alerts, enables TimescaleDB extensions, and converts tables into hypertables for efficient time-series data handling.

```python
    async def _setup_schema(self):
        """Create necessary tables and hypertables"""
        async with self.pool.acquire() as conn:
            # Enable TimescaleDB
            await conn.execute("CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE;")

            # Create sensor readings table
            await conn.execute("""
                CREATE TABLE IF NOT EXISTS sensor_readings (
                    time        TIMESTAMPTZ NOT NULL,
                    sensor_id   INTEGER NOT NULL,
                    temperature DOUBLE PRECISION,
                    humidity    DOUBLE PRECISION,
                    pressure    DOUBLE PRECISION,
                    battery     DOUBLE PRECISION,
                    rssi        INTEGER
                );
            """
            )

            # Convert to hypertable
            await conn.execute("""
                SELECT create_hypertable(
                    'sensor_readings', 'time',
                    chunk_time_interval => INTERVAL '1 day',
                    if_not_exists => TRUE
                );
            """
            )

            # Create alerts table
            await conn.execute("""
                CREATE TABLE IF NOT EXISTS sensor_alerts (
                    time        TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    sensor_id   INTEGER NOT NULL,
                    alert_type  TEXT NOT NULL,
                    message     TEXT,
                    value       DOUBLE PRECISION,
                    threshold   DOUBLE PRECISION
                );
            """
            )

            await conn.execute("""
                SELECT create_hypertable(
                    'sensor_alerts', 'time',
                    if_not_exists => TRUE
                );
            """
            )

```

--------------------------------

### Install Python Packages for TimescaleDB

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Installs essential Python libraries for interacting with TimescaleDB, including database adapters and data manipulation tools.

```Bash
# Install required Python packages
# psycopg2-binary: PostgreSQL adapter (includes libpq)
# asyncpg: Async PostgreSQL driver for high performance
# sqlalchemy: ORM with PostgreSQL support
# pandas: Data manipulation and analysis
pip install psycopg2-binary asyncpg sqlalchemy pandas
```

--------------------------------

### Setup Compression and Retention Policies

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Configures TimescaleDB policies to compress sensor reading data after 7 days and drop data older than 90 days, optimizing storage and performance.

```python
    async def _setup_policies(self):
        """Set up compression and retention policies"""
        async with self.pool.acquire() as conn:
            # Enable compression
            await conn.execute("""
                ALTER TABLE sensor_readings SET (
                    timescaledb.compress,
                    timescaledb.compress_segmentby = 'sensor_id',
                    timescaledb.compress_orderby = 'time DESC'
                );
            """
            )

            # Compression policy - compress after 7 days
            await conn.execute("""
                SELECT add_compression_policy(
                    'sensor_readings',
                    compress_after => INTERVAL '7 days',
                    if_not_exists => true
                );
            """
            )

            # Retention policy - drop after 90 days
            await conn.execute("""
                SELECT add_retention_policy(
                    'sensor_readings',
                    drop_after => INTERVAL '90 days',
                    if_not_exists => true
                );
            """
            )

```

--------------------------------

### SQLAlchemy Integration with TimescaleDB Boilerplate

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Basic setup for using SQLAlchemy with TimescaleDB, including creating an engine, defining a base for declarative models, and setting up a session maker. This code serves as a starting point for ORM-based interaction.

```python
# sqlalchemy_integration.py
# Using SQLAlchemy with TimescaleDB
from sqlalchemy import create_engine, Column, Integer, Float, String, DateTime, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime, timedelta
```

--------------------------------

### Setup Continuous Aggregates for Hourly Statistics

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Creates a materialized view to efficiently store and query hourly statistics for sensor readings, including average temperature, humidity, battery levels, and the count of readings.

```python
    async def _setup_continuous_aggregates(self):
        """Set up continuous aggregates for efficient queries"""
        async with self.pool.acquire() as conn:
            # Hourly statistics aggregate
            await conn.execute("""
                CREATE MATERIALIZED VIEW IF NOT EXISTS sensor_hourly_agg
                WITH (timescaledb.continuous) AS
                SELECT
                    time_bucket('1 hour', time) AS hour,
                    sensor_id,
                    AVG(temperature) AS avg_temp,
                    MIN(temperature) AS min_temp,
                    MAX(temperature) AS max_temp,
                    AVG(humidity) AS avg_humidity,
                    AVG(battery) AS avg_battery,
                    COUNT(*) AS readings
                FROM sensor_readings
                GROUP BY hour, sensor_id
                WITH NO DATA;
            """
            )

```

--------------------------------

### Configure Tiered Data Retention Policies in TimescaleDB

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Example demonstrating how to set up tiered retention policies for different data aggregation levels. This function configures distinct retention periods for raw data, hourly aggregates, and daily aggregates.

```python
def setup_tiered_retention():
    """
    Example of tiered retention with different policies.
    Raw data: 30 days
    Hourly aggregates: 1 year
    Daily aggregates: 5 years
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            # Raw data - shortest retention
            cur.execute("""
                SELECT add_retention_policy(
                    'sensor_data',
                    drop_after => INTERVAL '30 days',
                    if_not_exists => true
                );
            """ )

            # Hourly aggregates - medium retention
            cur.execute("""
                SELECT add_retention_policy(
                    'sensor_hourly_stats',
                    drop_after => INTERVAL '1 year',
                    if_not_exists => true
                );
            """ )

            # Daily aggregates - longest retention
            cur.execute("""
                SELECT add_retention_policy(
                    'sensor_daily_stats',
                    drop_after => INTERVAL '5 years',
                    if_not_exists => true
                );
            """ )

            print("Tiered retention policies configured")
```

--------------------------------

### GET /alerts

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Retrieves recent alerts, optionally filtered by sensor ID.

```APIDOC
## GET /alerts

### Description
Get recent alerts, optionally filtered by sensor.

### Method
GET

### Parameters
#### Query Parameters
- **sensor_id** (int) - Optional
- **hours** (int) - Optional - Time window in hours (default: 24)
```

--------------------------------

### GET /sensor/{sensor_id}/summary

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Retrieves summary statistics for a specific sensor over a defined time period.

```APIDOC
## GET /sensor/{sensor_id}/summary

### Description
Get summary statistics for a sensor including count, averages, and min/max values.

### Method
GET

### Parameters
#### Path Parameters
- **sensor_id** (int) - Required

#### Query Parameters
- **hours** (int) - Optional - Time window in hours (default: 24)
```

--------------------------------

### GET /sensor/{sensor_id}/trends

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Retrieves hourly trends for a sensor from continuous aggregates.

```APIDOC
## GET /sensor/{sensor_id}/trends

### Description
Get hourly trends from continuous aggregate for a specific sensor.

### Method
GET

### Parameters
#### Path Parameters
- **sensor_id** (int) - Required

#### Query Parameters
- **days** (int) - Optional - Time window in days (default: 7)
```

--------------------------------

### Get the most recent reading per sensor

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Uses DISTINCT ON combined with ORDER BY to efficiently retrieve the latest record for each sensor.

```python
def get_last_reading_per_sensor():
    """
    Get the most recent reading for each sensor efficiently.
    Uses DISTINCT ON which TimescaleDB optimizes well.
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            # DISTINCT ON returns only the first row for each sensor_id
            # Combined with ORDER BY time DESC, gets the latest reading
            cur.execute("""
                SELECT DISTINCT ON (sensor_id)
                    sensor_id,
                    time,
                    temperature,
                    humidity,
                    pressure,
                    location
                FROM sensor_data
                WHERE time > NOW() - INTERVAL '1 hour'
                ORDER BY sensor_id, time DESC;
            "")

            return cur.fetchall()
```

--------------------------------

### Get Chunk Information for TimescaleDB Hypertable

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Retrieve detailed information about all chunks belonging to a hypertable. This is helpful for understanding data distribution and planning retention strategies. The output includes chunk name, time range, compression status, and size.

```python
def get_chunk_info(table_name: str):
    """
    Get information about all chunks for a hypertable.
    Useful for understanding data distribution and planning retention.
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT
                    chunk_name,
                    range_start,
                    range_end,
                    is_compressed,
                    pg_size_pretty(total_bytes) AS chunk_size
                FROM timescaledb_information.chunks
                WHERE hypertable_name = %s
                ORDER BY range_start DESC;
            """, (table_name,))

            return cur.fetchall()
```

--------------------------------

### IoT Monitoring Application Entry Point

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Placeholder for the main application file.

```python
# iot_monitoring_app.py
```

--------------------------------

### Initialize Database Engine

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Establishes a connection to the PostgreSQL/TimescaleDB instance using SQLAlchemy.

```python
engine = create_engine("postgresql://postgres:password@localhost:5432/timeseries_db")
```

--------------------------------

### Initialize IoT Monitoring Service

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Initializes the database connection pool and sets up the necessary schema, continuous aggregates, and policies for the IoT monitoring service.

```python
import asyncio
import asyncpg
from datetime import datetime
from typing import List, Dict, Optional
from dataclasses import dataclass
import json

@dataclass
class SensorConfig:
    """Configuration for a sensor"""
    sensor_id: int
    name: str
    location: str
    alert_thresholds: Dict[str, tuple]  # metric: (min, max)

class IoTMonitoringService:
    """
    Complete IoT monitoring service using TimescaleDB.
    Handles data collection, storage, analysis, and alerting.
    """

    def __init__(self, dsn: str):
        self.dsn = dsn
        self.pool: Optional[asyncpg.Pool] = None
        self.sensors: Dict[int, SensorConfig] = {}

    async def initialize(self):
        """Initialize database connection and schema"""
        # Create connection pool
        self.pool = await asyncpg.create_pool(
            self.dsn,
            min_size=5,
            max_size=20
        )

        # Set up database schema
        await self._setup_schema()
        await self._setup_continuous_aggregates()
        await self._setup_policies()

```

--------------------------------

### Initialize batch insert script

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Placeholder file header for batch insertion logic.

```python
# batch_insert.py
```

--------------------------------

### Run Analysis Workflow

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Demonstrates a complete end-to-end workflow including loading, analyzing, and anomaly detection.

```python
# Example analysis workflow
def run_analysis_workflow():
    """Complete analysis workflow example"""
    # Load data
    df = load_sensor_data_to_dataframe(sensor_id=1, days=7)
    print(f"Loaded {len(df)} readings")

    # Analyze
    analysis = analyze_sensor_data(df)
    print("\nSummary Statistics:")
    print(analysis["summary"])

    # Resample to hourly
    hourly = resample_and_analyze(df, freq='1H')
    print(f"\nHourly data: {len(hourly)} rows")

    # Find anomalies
    temp_mean = df['temperature'].mean()
    temp_std = df['temperature'].std()
    anomalies = df[abs(df['temperature'] - temp_mean) > 2 * temp_std]
    print(f"\nFound {len(anomalies)} temperature anomalies")

    return df, analysis

if __name__ == "__main__":
    run_analysis_workflow()
```

--------------------------------

### Initialize Retention Policy Script

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Boilerplate file for implementing data retention policies.

```python
# retention.py
```

--------------------------------

### Query Time-Series Data with Python

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Common patterns for retrieving sensor data, calculating hourly averages, and generating location statistics using time_bucket.

```python
# query_examples.py
# Common time-series query patterns
from timescale_connection import get_connection
from datetime import datetime, timedelta

def get_recent_readings(sensor_id: int, hours: int = 24):
    """
    Retrieve recent readings for a specific sensor.
    Uses time filter to leverage chunk exclusion.
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT time, temperature, humidity, pressure
                FROM sensor_data
                WHERE sensor_id = %s
                  AND time > NOW() - INTERVAL '%s hours'
                ORDER BY time DESC
                LIMIT 1000;
            """, (sensor_id, hours))

            return cur.fetchall()

def get_hourly_averages(sensor_id: int, days: int = 7):
    """
    Calculate hourly averages using time_bucket.
    time_bucket groups timestamps into fixed intervals.
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            # time_bucket creates fixed-size time intervals
            # '1 hour' means aggregate data into hourly buckets
            cur.execute("""
                SELECT
                    time_bucket('1 hour', time) AS hour,
                    AVG(temperature) AS avg_temp,
                    AVG(humidity) AS avg_humidity,
                    MIN(temperature) AS min_temp,
                    MAX(temperature) AS max_temp,
                    COUNT(*) AS reading_count
                FROM sensor_data
                WHERE sensor_id = %s
                  AND time > NOW() - INTERVAL '%s days'
                GROUP BY hour
                ORDER BY hour DESC;
            """, (sensor_id, days))

            return cur.fetchall()

def get_location_statistics(location: str, interval: str = '15 minutes'):
    """
    Get aggregated statistics for a location.
    Demonstrates dynamic interval selection.
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT
                    time_bucket(%s, time) AS bucket,
                    COUNT(DISTINCT sensor_id) AS active_sensors,
                    AVG(temperature) AS avg_temp,
                    STDDEV(temperature) AS temp_stddev,
                    AVG(humidity) AS avg_humidity,
                    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY temperature) AS temp_p95
                FROM sensor_data
                WHERE location = %s
                  AND time > NOW() - INTERVAL '24 hours'
                GROUP BY bucket
                ORDER BY bucket DESC;
            """, (interval, location))

            return cur.fetchall()
```

--------------------------------

### Create sensor and metrics hypertables in Python

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Defines functions to initialize PostgreSQL tables and convert them into TimescaleDB hypertables with specific chunk intervals and indexes.

```python
from timescale_connection import get_connection

def create_sensor_hypertable():
    """
    Create a hypertable for storing sensor readings.
    Hypertables automatically partition data by time for efficient queries.
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            # Step 1: Create regular PostgreSQL table
            # Note: time column should be NOT NULL for hypertables
            cur.execute("""
                CREATE TABLE IF NOT EXISTS sensor_data (
                    time        TIMESTAMPTZ NOT NULL,
                    sensor_id   INTEGER NOT NULL,
                    temperature DOUBLE PRECISION,
                    humidity    DOUBLE PRECISION,
                    pressure    DOUBLE PRECISION,
                    location    TEXT
                );
            """)

            # Step 2: Convert to hypertable
            # chunk_time_interval: Size of each time partition
            # if_not_exists: Prevent error if already a hypertable
            cur.execute("""
                SELECT create_hypertable(
                    'sensor_data',
                    'time',
                    chunk_time_interval => INTERVAL '1 day',
                    if_not_exists => TRUE
                );
            """)

            # Step 3: Create indexes for common query patterns
            # Composite index on sensor_id and time for filtered queries
            cur.execute("""
                CREATE INDEX IF NOT EXISTS idx_sensor_time
                ON sensor_data (sensor_id, time DESC);
            """)

            # Index on location for location-based queries
            cur.execute("""
                CREATE INDEX IF NOT EXISTS idx_sensor_location
                ON sensor_data (location, time DESC);
            """)

            print("Hypertable 'sensor_data' created successfully")

def create_metrics_hypertable():
    """
    Create a hypertable for application metrics.
    Uses shorter chunk interval for high-frequency data.
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS app_metrics (
                    time        TIMESTAMPTZ NOT NULL,
                    host        TEXT NOT NULL,
                    metric_name TEXT NOT NULL,
                    value       DOUBLE PRECISION NOT NULL,
                    tags        JSONB DEFAULT '{}'
                );
            """)

            # Shorter chunk interval for high-frequency metrics
            # 1 hour chunks work well for metrics collected every second
            cur.execute("""
                SELECT create_hypertable(
                    'app_metrics',
                    'time',
                    chunk_time_interval => INTERVAL '1 hour',
                    if_not_exists => TRUE
                );
            """)

            # Composite index for metric queries by host and name
            cur.execute("""
                CREATE INDEX IF NOT EXISTS idx_metrics_host_name
                ON app_metrics (host, metric_name, time DESC);
            """)

            print("Hypertable 'app_metrics' created successfully")

if __name__ == "__main__":
    create_sensor_hypertable()
    create_metrics_hypertable()
```

--------------------------------

### Initialize Pandas Integration

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Imports necessary libraries for loading TimescaleDB data into Pandas DataFrames.

```python
# pandas_integration.py
# Using pandas with TimescaleDB for data analysis
import pandas as pd
from sqlalchemy import create_engine
from datetime import datetime, timedelta
```

--------------------------------

### Async Batch Inserts with asyncpg

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

High-performance asynchronous batch insertion using COPY and prepared statements to minimize database overhead.

```python
# async_batch_insert.py
# High-performance async batch insertion
import asyncio
import asyncpg
from datetime import datetime, timedelta
import random

async def insert_metrics_batch(pool: asyncpg.Pool, metrics: list):
    """
    Insert metrics using async copy for maximum performance.
    COPY is faster than INSERT for large batches.

    Args:
        pool: asyncpg connection pool
        metrics: List of tuples (time, host, metric_name, value, tags)
    """
    async with pool.acquire() as conn:
        # Use COPY for bulk inserts - faster than INSERT
        await conn.copy_records_to_table(
            'app_metrics',
            records=metrics,
            columns=['time', 'host', 'metric_name', 'value', 'tags']
        )

async def insert_with_prepared_statement(pool: asyncpg.Pool, metrics: list):
    """
    Insert using prepared statements for repeated inserts.
    Prepared statements reduce query parsing overhead.
    """
    async with pool.acquire() as conn:
        # Prepare the statement once
        stmt = await conn.prepare("""
            INSERT INTO app_metrics (time, host, metric_name, value, tags)
            VALUES ($1, $2, $3, $4, $5)
        """)

        # Execute in batches using executemany
        await stmt.executemany(metrics)

async def buffered_writer(pool: asyncpg.Pool, buffer_size=1000, flush_interval=5):
    """
    Buffered writer that batches writes for efficiency.
    Flushes when buffer is full or interval expires.
    """
    buffer = []
    last_flush = datetime.now()

    async def flush():
        nonlocal buffer, last_flush
        if buffer:
            await insert_metrics_batch(pool, buffer)
            print(f"Flushed {len(buffer)} metrics")
            buffer = []
            last_flush = datetime.now()

    async def write(metric: tuple):
        nonlocal buffer
        buffer.append(metric)

        # Flush if buffer is full or time interval exceeded
        time_since_flush = (datetime.now() - last_flush).total_seconds()
        if len(buffer) >= buffer_size or time_since_flush >= flush_interval:
            await flush()

    return write, flush
```

--------------------------------

### Configure SQLAlchemy Engine and ORM Model

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Sets up the database connection pool and defines the SensorReading model mapped to a TimescaleDB hypertable.

```python
engine = create_engine(
    "postgresql://postgres:password@localhost:5432/timeseries_db",
    pool_size=10,           # Number of connections to keep
    max_overflow=20,        # Additional connections allowed
    pool_pre_ping=True      # Verify connections before use
)

Base = declarative_base()
Session = sessionmaker(bind=engine)

class SensorReading(Base):
    """
    SQLAlchemy model for sensor readings.
    Maps to the sensor_data hypertable.
    """
    __tablename__ = 'sensor_data'

    # Composite primary key of time and sensor_id
    time = Column(DateTime(timezone=True), primary_key=True)
    sensor_id = Column(Integer, primary_key=True)
    temperature = Column(Float)
    humidity = Column(Float)
    pressure = Column(Float)
    location = Column(String)

    def __repr__(self):
        return f"<SensorReading(sensor={self.sensor_id}, time={self.time}, temp={self.temperature})>"
```

--------------------------------

### Create Daily Continuous Aggregate

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Creates a daily aggregate view using the hourly aggregate as its source, optimizing for long-term trend analysis. This reduces query time for daily summaries. It also includes a daily refresh policy.

```Python
def create_daily_aggregate():
    """
    Create a daily aggregate for long-term trend analysis.
    Uses the hourly aggregate as source for efficiency.
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE MATERIALIZED VIEW sensor_daily_stats
                WITH (timescaledb.continuous) AS
                SELECT
                    time_bucket('1 day', bucket) AS day,
                    sensor_id,
                    location,
                    AVG(avg_temp) AS avg_temp,
                    MIN(min_temp) AS min_temp,
                    MAX(max_temp) AS max_temp,
                    AVG(avg_humidity) AS avg_humidity,
                    SUM(reading_count) AS total_readings
                FROM sensor_hourly_stats
                GROUP BY day, sensor_id, location
                WITH NO DATA;
            """)

            # Daily refresh policy with longer lookback
            cur.execute("""
                SELECT add_continuous_aggregate_policy(
                    'sensor_daily_stats',
                    start_offset => INTERVAL '7 days',
                    end_offset => INTERVAL '1 day',
                    schedule_interval => INTERVAL '1 day'
                );
            """)

            print("Continuous aggregate 'sensor_daily_stats' created")
```

--------------------------------

### TimescaleDB Query Optimization Patterns

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Illustrates optimized versus non-optimized SQL query patterns for TimescaleDB in Python, focusing on time-bounded queries, approximate counts, and indexed ordering.

```python
# performance_tips.py
# TimescaleDB performance optimization examples
from timescale_connection import get_connection

def optimized_query_patterns():
    """Examples of optimized vs non-optimized queries"""

    with get_connection() as conn:
        with conn.cursor() as cur:
            # BAD: Full table scan
            # cur.execute("SELECT * FROM sensor_data WHERE sensor_id = 1")

            # GOOD: Time-bounded query enables chunk exclusion
            cur.execute("""
                SELECT * FROM sensor_data
                WHERE sensor_id = 1
                  AND time > NOW() - INTERVAL '1 day'
            """)

            # BAD: Counting all rows
            # cur.execute("SELECT COUNT(*) FROM sensor_data")

            # GOOD: Use approximate count for large tables
            cur.execute("""
                SELECT approximate_row_count('sensor_data')
            """)

            # BAD: ORDER BY on non-indexed column
            # cur.execute("SELECT * FROM sensor_data ORDER BY temperature")

            # GOOD: ORDER BY on indexed time column
            cur.execute("""
                SELECT * FROM sensor_data
                WHERE time > NOW() - INTERVAL '1 hour'
                ORDER BY time DESC
                LIMIT 100
            """)

```

--------------------------------

### Perform Buffered Writes with asyncpg

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Uses a buffered writer pattern to batch metric insertions into TimescaleDB. Requires an active asyncpg connection pool.

```python
async def main():
    pool = await asyncpg.create_pool(
        "postgresql://postgres:password@localhost:5432/timeseries_db"
    )

    write, flush = await buffered_writer(pool)

    # Simulate writing metrics
    import json
    for i in range(5000):
        metric = (
            datetime.now(),
            "host-1",
            "cpu_usage",
            random.uniform(0, 100),
            json.dumps({"core": i % 4})  # JSONB tags
        )
        await write(metric)

    # Final flush to ensure all data is written
    await flush()
    await pool.close()

if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Async TimescaleDB Connection Handler

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Provides a class for asynchronous database operations with TimescaleDB using connection pooling. Ensure the DSN is correctly formatted.

```python
import asyncio
import asyncpg
from typing import Optional

class AsyncTimescaleDB:
    """
    Async database handler for TimescaleDB.
    Uses connection pooling for efficient resource management.
    """

    def __init__(self, dsn: str):
        # DSN format: postgresql://user:password@host:port/database
        self.dsn = dsn
        self.pool: Optional[asyncpg.Pool] = None

    async def connect(self):
        """
        Create connection pool with specified limits.
        min_size: Connections kept ready for immediate use
        max_size: Maximum concurrent connections
        """
        self.pool = await asyncpg.create_pool(
            self.dsn,
            min_size=5,
            max_size=20,
            command_timeout=60  # Query timeout in seconds
        )

    async def close(self):
        """Close all connections in the pool"""
        if self.pool:
            await self.pool.close()

    async def execute(self, query: str, *args):
        """Execute a query without returning results"""
        async with self.pool.acquire() as conn:
            return await conn.execute(query, *args)

    async def fetch(self, query: str, *args):
        """Execute a query and return all results"""
        async with self.pool.acquire() as conn:
            return await conn.fetch(query, *args)

    async def fetchrow(self, query: str, *args):
        """Execute a query and return single row"""
        async with self.pool.acquire() as conn:
            return await conn.fetchrow(query, *args)

# Usage example
async def main():
    db = AsyncTimescaleDB("postgresql://postgres:password@localhost:5432/timeseries_db")
    await db.connect()

    # Enable TimescaleDB extension
    await db.execute("CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE;")

    await db.close()

if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Create Hourly Continuous Aggregate

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Creates a materialized view that pre-computes hourly statistics from sensor data. It automatically updates as new data arrives. Use this for real-time or near-real-time hourly summaries.

```Python
# continuous_aggregates.py
# Creating and managing continuous aggregates
from timescale_connection import get_connection

def create_hourly_aggregate():
    """
    Create a continuous aggregate for hourly sensor statistics.
    Automatically updates as new data arrives.
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            # Drop existing aggregate if recreating
            cur.execute("""
                DROP MATERIALIZED VIEW IF EXISTS sensor_hourly_stats CASCADE;
            """)

            # Create continuous aggregate
            # WITH (timescaledb.continuous) enables automatic refresh
            cur.execute("""
                CREATE MATERIALIZED VIEW sensor_hourly_stats
                WITH (timescaledb.continuous) AS
                SELECT
                    time_bucket('1 hour', time) AS bucket,
                    sensor_id,
                    location,
                    AVG(temperature) AS avg_temp,
                    MIN(temperature) AS min_temp,
                    MAX(temperature) AS max_temp,
                    AVG(humidity) AS avg_humidity,
                    AVG(pressure) AS avg_pressure,
                    COUNT(*) AS reading_count
                FROM sensor_data
                GROUP BY bucket, sensor_id, location
                WITH NO DATA;
            """)

            # Configure automatic refresh policy
            # start_offset: How far back to refresh
            # end_offset: Gap from current time (for late-arriving data)
            # schedule_interval: How often to run refresh
            cur.execute("""
                SELECT add_continuous_aggregate_policy(
                    'sensor_hourly_stats',
                    start_offset => INTERVAL '3 days',
                    end_offset => INTERVAL '1 hour',
                    schedule_interval => INTERVAL '1 hour'
                );
            """)

            print("Continuous aggregate 'sensor_hourly_stats' created")
```

--------------------------------

### POST /sensor/readings/batch

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Efficiently records multiple sensor readings in a single operation.

```APIDOC
## POST /sensor/readings/batch

### Description
Records multiple sensor readings efficiently using bulk insertion.

### Method
POST

### Request Body
- **readings** (List[tuple]) - Required - A list of tuples containing sensor data fields.
```

--------------------------------

### Record Sensor Readings and Batch Data

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Methods for inserting individual sensor readings with threshold checks and performing efficient bulk inserts using copy_records_to_table.

```python
    async def record_reading(self, sensor_id: int, temperature: float,
                            humidity: float, pressure: float,
                            battery: float, rssi: int):
        """Record a sensor reading and check for alerts"""
        async with self.pool.acquire() as conn:
            # Insert reading
            await conn.execute("""
                INSERT INTO sensor_readings
                    (time, sensor_id, temperature, humidity, pressure, battery, rssi)
                VALUES (NOW(), $1, $2, $3, $4, $5, $6)
            """, sensor_id, temperature, humidity, pressure, battery, rssi)

        # Check thresholds and generate alerts
        await self._check_thresholds(sensor_id, {
            'temperature': temperature,
            'humidity': humidity,
            'battery': battery
        })
```

```python
    async def record_batch(self, readings: List[tuple]):
        """Record multiple readings efficiently"""
        async with self.pool.acquire() as conn:
            await conn.copy_records_to_table(
                'sensor_readings',
                records=readings,
                columns=['time', 'sensor_id', 'temperature', 'humidity',
                        'pressure', 'battery', 'rssi']
            )
```

--------------------------------

### Batch Insert with psycopg2

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Uses execute_values for efficient batch insertion of sensor readings into a PostgreSQL database.

```python
import psycopg2.extras
from datetime import datetime, timedelta
import random
from timescale_connection import get_connection

def insert_sensor_readings_batch(readings: list):
    """
    Insert multiple sensor readings in a single batch.
    Much faster than individual inserts for large datasets.

    Args:
        readings: List of tuples (time, sensor_id, temp, humidity, pressure, location)
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            # Use execute_values for efficient batch insert
            # page_size controls how many rows per INSERT statement
            psycopg2.extras.execute_values(
                cur,
                """
                INSERT INTO sensor_data
                    (time, sensor_id, temperature, humidity, pressure, location)
                VALUES %s
                """,
                readings,
                page_size=1000  # Rows per INSERT statement
            )

    print(f"Inserted {len(readings)} sensor readings")

def generate_sample_data(num_sensors=10, num_hours=24):
    """
    Generate sample sensor data for testing.
    Creates realistic temperature, humidity, and pressure readings.
    """
    readings = []
    locations = ["warehouse_a", "warehouse_b", "office_1", "office_2", "datacenter"]
    base_time = datetime.now() - timedelta(hours=num_hours)

    for hour in range(num_hours):
        for minute in range(60):
            timestamp = base_time + timedelta(hours=hour, minutes=minute)

            for sensor_id in range(num_sensors):
                # Generate realistic sensor values with some variation
                reading = (
                    timestamp,
                    sensor_id,
                    20 + random.gauss(0, 5),      # Temperature around 20C
                    50 + random.gauss(0, 10),     # Humidity around 50%
                    1013 + random.gauss(0, 5),    # Pressure around 1013 hPa
                    random.choice(locations)
                )
                readings.append(reading)

    return readings

# Example usage
if __name__ == "__main__":
    # Generate and insert sample data
    sample_readings = generate_sample_data(num_sensors=10, num_hours=24)
    insert_sensor_readings_batch(sample_readings)
```

--------------------------------

### Synchronous TimescaleDB Connection with psycopg2

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Establishes a synchronous database connection pool using psycopg2. The TimescaleDB extension must be enabled once per database.

```Python
# timescale_connection.py
# Basic TimescaleDB connection using psycopg2
import psycopg2
from psycopg2 import pool
from contextlib import contextmanager

# Database connection parameters
DB_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "database": "timeseries_db",
    "user": "postgres",
    "password": "password"
}

def create_connection_pool(min_conn=5, max_conn=20):
    """
    Create a connection pool for efficient connection reuse.
    Pools reduce overhead of creating new connections for each query.
    """
    return pool.ThreadedConnectionPool(
        minconn=min_conn,  # Minimum connections to keep ready
        maxconn=max_conn,  # Maximum connections allowed
        **DB_CONFIG
    )

# Initialize global connection pool
connection_pool = create_connection_pool()

@contextmanager
def get_connection():
    """
    Context manager for safe connection handling.
    Automatically returns connection to pool when done.
    """
    conn = connection_pool.getconn()
    try:
        yield conn
        conn.commit()  # Commit transaction if no errors
    except Exception as e:
        conn.rollback()  # Rollback on any error
        raise e
    finally:
        connection_pool.putconn(conn)  # Return to pool

def setup_timescaledb():
    """
    Enable TimescaleDB extension on the database.
    Must be run once before creating hypertables.
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            # Enable the TimescaleDB extension
            cur.execute("CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE;")
            print("TimescaleDB extension enabled successfully")
```

--------------------------------

### Load Sensor Data into DataFrame

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Retrieves sensor data from the database using parameterized queries for security and efficiency.

```python
def load_sensor_data_to_dataframe(sensor_id: int, days: int = 7) -> pd.DataFrame:
    """
    Load sensor data directly into a pandas DataFrame.
    Uses read_sql for efficient data transfer.
    """
    query = """
        SELECT time, temperature, humidity, pressure, location
        FROM sensor_data
        WHERE sensor_id = %(sensor_id)s
          AND time > NOW() - INTERVAL '%(days)s days'
        ORDER BY time
    """

    # read_sql handles connection and data type conversion
    df = pd.read_sql(
        query,
        engine,
        params={"sensor_id": sensor_id, "days": days},
        parse_dates=["time"],  # Parse time column as datetime
        index_col="time"       # Use time as index
    )

    return df
```

--------------------------------

### Create TimescaleDB Hypertable

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Converts a regular PostgreSQL table into a hypertable for efficient time-series data partitioning. Adjust chunk_time_interval based on query patterns.

```sql
CREATE TABLE sensor_data (
    time TIMESTAMPTZ NOT NULL,
    device_id INT NOT NULL,
    temperature NUMERIC(10, 2)
);

SELECT create_hypertable('sensor_data', 'time', chunk_time_interval => INTERVAL '1 day');
```

--------------------------------

### Retrieve readings with gap filling

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Uses time_bucket_gapfill to create continuous time series, filling gaps with locf or interpolation.

```python
def get_readings_with_gap_fill(sensor_id: int, interval: str = '5 minutes'):
    """
    Retrieve readings with gap filling for missing data points.
    Uses time_bucket_gapfill to create continuous time series.
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            # time_bucket_gapfill creates buckets even when no data exists
            # locf() fills gaps with last observed value (Last Observation Carried Forward)
            # interpolate() creates linear interpolation between points
            cur.execute("""
                SELECT
                    time_bucket_gapfill(%s, time) AS bucket,
                    sensor_id,
                    locf(AVG(temperature)) AS temperature,
                    interpolate(AVG(humidity)) AS humidity
                FROM sensor_data
                WHERE sensor_id = %s
                  AND time > NOW() - INTERVAL '24 hours'
                  AND time <= NOW()
                GROUP BY bucket, sensor_id
                ORDER BY bucket;
            "", (interval, sensor_id))

            return cur.fetchall()
```

--------------------------------

### Register Sensor Configuration

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Registers a sensor with its configuration details, including alert thresholds, within the monitoring service.

```python
    def register_sensor(self, config: SensorConfig):
        """Register a sensor with alert thresholds"""
        self.sensors[config.sensor_id] = config

```

--------------------------------

### Perform Bulk Data Insertion

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Uses SQLAlchemy's bulk_insert_mappings for high-performance data ingestion.

```python
def bulk_insert_readings(readings: list):
    """
    Bulk insert readings for better performance.
    Uses SQLAlchemy core for efficiency.
    """
    session = Session()
    try:
        # Convert to list of dicts for bulk insert
        reading_dicts = [
            {
                "time": r[0],
                "sensor_id": r[1],
                "temperature": r[2],
                "humidity": r[3],
                "pressure": r[4],
                "location": r[5]
            }
            for r in readings
        ]

        # Use bulk_insert_mappings for efficiency
        session.bulk_insert_mappings(SensorReading, reading_dicts)
        session.commit()
        print(f"Bulk inserted {len(readings)} readings")
    except Exception as e:
        session.rollback()
        raise e
    finally:
        session.close()
```

--------------------------------

### Execute TimescaleDB Specific Queries

Source: https://oneuptime.com/blog/post/2026-02-02-timescaledb-python/view

Uses raw SQL via SQLAlchemy's text() function to leverage TimescaleDB's time_bucket aggregation.

```python
def query_with_time_bucket(sensor_id: int, bucket_interval: str = '1 hour'):
    """
    Use TimescaleDB time_bucket with SQLAlchemy.
    Requires raw SQL for TimescaleDB-specific functions.
    """
    session = Session()
    try:
        # Use text() for raw SQL with TimescaleDB functions
        result = session.execute(text("""
            SELECT
                time_bucket(:interval, time) AS bucket,
                AVG(temperature) AS avg_temp,
                AVG(humidity) AS avg_humidity,
                COUNT(*) AS count
            FROM sensor_data
            WHERE sensor_id = :sensor_id
              AND time > NOW() - INTERVAL '7 days'
            GROUP BY bucket
            ORDER BY bucket DESC
        """), {"interval": bucket_interval, "sensor_id": sensor_id})

        return result.fetchall()
    finally:
        session.close()
```
Tokens:
Raw