### 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"" ``` -------------------------------- ### 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() ```