### Create and Slice TimeSeries with Generic Exporters (Python)

Source: https://context7.com/bobthebuidler/generic_exporters/llms.txt

Shows how to wrap a `Metric` with the `TimeSeries` class to enable time-based slicing and query plan creation. The example demonstrates creating a `TimeSeries` from a custom metric, performing arithmetic operations on it, and slicing it to define a query range.

```python
from datetime import datetime, timedelta
from decimal import Decimal
from generic_exporters import Metric, TimeSeries, Constant

class PriceMetric(Metric):
    @property
    def key(self) -> str:
        return "btc_price"

    async def produce(self, timestamp: datetime) -> Decimal:
        return Decimal("45000.00")

# Create a TimeSeries from a Metric
price_metric = PriceMetric()
price_series = TimeSeries(price_metric)

# Access the metric key
print(price_series.key)  # btc_price

# TimeSeries supports arithmetic operations
doubled_price = price_series + price_series
scaled_price = price_series * Constant(2)

# Slice to create a QueryPlan for a time range
start = datetime.utcnow() - timedelta(days=7)
end = datetime.utcnow()
interval = timedelta(hours=1)

query_plan = price_series[start:end:interval]
print(query_plan)  # <QueryPlan for <TimeSeries ...> interval=1:00:00>
```

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

### Utilize Constant Metrics in Generic Exporters (Python)

Source: https://context7.com/bobthebuidler/generic_exporters/llms.txt

Illustrates the usage of the `Constant` class, a special `Metric` that always returns a fixed value. This is useful for arithmetic operations involving constant numbers. The example shows creating constants and verifying their output across different timestamps.

```python
from decimal import Decimal
from datetime import datetime
from generic_exporters import Constant

# Create a constant value
multiplier = Constant(100)
percentage = Constant(Decimal("0.05"))

# Constants produce the same value at any timestamp
import asyncio

async def main():
    ts1 = datetime(2024, 1, 1)
    ts2 = datetime(2024, 12, 31)

    assert await multiplier.produce(ts1) == Decimal(100)
    assert await multiplier.produce(ts2) == Decimal(100)

    # Key is derived from class name
    print(multiplier.key)  # constant

asyncio.run(main())
```

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

### Create and Iterate Over QueryPlan with Python

Source: https://context7.com/bobthebuidler/generic_exporters/llms.txt

Shows how to create a QueryPlan by slicing a TimeSeries and how to access individual data points, await full materialization into a Dataset, and asynchronously iterate over the results. This class represents a planned query for time series data.

```python
from datetime import datetime, timedelta
from decimal import Decimal
from generic_exporters import Metric, TimeSeries, QueryPlan

class SensorMetric(Metric):
    @property
    def key(self) -> str:
        return "sensor_reading"

    async def produce(self, timestamp: datetime) -> Decimal:
        return Decimal("42.0")

import asyncio

async def main():
    sensor = SensorMetric()
    series = TimeSeries(sensor)

    # Create a query plan via slicing
    start = datetime.utcnow() - timedelta(hours=24)
    end = datetime.utcnow()
    interval = timedelta(minutes=30)

    query = series[start:end:interval]
    # Or create directly:
    # query = QueryPlan(series, start, end, interval)

    # Access individual timestamps
    ts = datetime.utcnow() - timedelta(hours=1)
    row = query[ts]  # Returns TimeDataRow

    # Await the full query to get a Dataset
    dataset = await query
    print(type(dataset))  # <class 'generic_exporters.dataset.Dataset'>

    # Async iterate over results
    async for timestamp, data in query:
        print(f"{timestamp}: {data}")
        break  # Just show first result

asyncio.run(main())
```

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

### Create and Query WideTimeSeries with Python

Source: https://context7.com/bobthebuidler/generic_exporters/llms.txt

Demonstrates how to create a WideTimeSeries object from multiple Metric or TimeSeries objects and how to slice it to create a QueryPlan for a specific time range. This is useful for comparing multiple metrics.

```python
from datetime import datetime, timedelta
from decimal import Decimal
from generic_exporters import Metric, TimeSeries, WideTimeSeries

class CryptoPrice(Metric):
    def __init__(self, symbol: str):
        super().__init__()
        self.symbol = symbol

    @property
    def key(self) -> str:
        return f"{self.symbol}_price"

    async def produce(self, timestamp: datetime) -> Decimal:
        # Fetch price for symbol at timestamp
        prices = {"BTC": "45000", "ETH": "2500", "SOL": "100"}
        return Decimal(prices.get(self.symbol, "0"))

# Create multiple metrics
btc = CryptoPrice("BTC")
eth = CryptoPrice("ETH")
sol = CryptoPrice("SOL")

# Create a WideTimeSeries from multiple metrics
portfolio = WideTimeSeries(btc, eth, sol)

# You can also pass TimeSeries objects
btc_series = TimeSeries(btc)
eth_series = TimeSeries(eth)
portfolio = WideTimeSeries(btc_series, eth_series)

# Slice to query multiple metrics over time
start = datetime.utcnow() - timedelta(days=1)
end = datetime.utcnow()
query = portfolio[start:end:timedelta(hours=1)]
```

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

### Materialize and Export Dataset with Python

Source: https://context7.com/bobthebuidler/generic_exporters/llms.txt

Illustrates how to materialize a QueryPlan into a Dataset object, which then allows for exporting the time series data to CSV format. The Dataset class is a container for materialized data and extends Python's dict.

```python
from datetime import datetime, timedelta
from decimal import Decimal
from generic_exporters import Metric, TimeSeries

class MetricData(Metric):
    @property
    def key(self) -> str:
        return "sample_data"

    async def produce(self, timestamp: datetime) -> Decimal:
        return Decimal("100.0")

import asyncio

async def main():
    metric = MetricData()
    series = TimeSeries(metric)

    start = datetime.utcnow() - timedelta(hours=1)
    end = datetime.utcnow()
    query = series[start:end:timedelta(minutes=10)]

    # Materialize the query into a Dataset
    dataset = await query

    # Export to CSV
    await dataset.to_csv("output.csv", index=True)

    # Plot the data (requires bqplot)
    # figure = await dataset.plot()

    # Export to a datastore
    # await dataset.export(my_datastore)

asyncio.run(main())
```

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

### Define and Use Custom Metrics with Generic Exporters (Python)

Source: https://context7.com/bobthebuidler/generic_exporters/llms.txt

Demonstrates how to define a custom metric by subclassing the `Metric` class, implementing the `produce` method, and composing metrics using arithmetic operations. It shows how to instantiate metrics, use `Constant` for fixed values, and produce metric values at a specific timestamp.

```python
from datetime import datetime
from decimal import Decimal
from generic_exporters import Metric, Constant

# Define a custom metric by subclassing Metric
class TemperatureMetric(Metric):
    def __init__(self, sensor_id: str):
        super().__init__()
        self.sensor_id = sensor_id

    @property
    def key(self) -> str:
        return f"temperature_{self.sensor_id}"

    async def produce(self, timestamp: datetime) -> Decimal:
        # Your logic to fetch temperature at timestamp
        # For example, query a database or external API
        return Decimal("72.5")

# Create metric instances
temp_sensor_1 = TemperatureMetric("sensor_1")
temp_sensor_2 = TemperatureMetric("sensor_2")

# Use Constant for fixed values
offset = Constant(10)

# Compose metrics using arithmetic operations
average_temp = (temp_sensor_1 + temp_sensor_2) / Constant(2)
adjusted_temp = temp_sensor_1 + offset

# Produce values at a specific timestamp
import asyncio

async def main():
    timestamp = datetime.utcnow()
    value = await temp_sensor_1.produce(timestamp)
    print(f"Temperature at {timestamp}: {value}")  # Temperature at 2024-01-15 10:30:00: 72.5

    # Composite metrics also produce values
    avg_value = await average_temp.produce(timestamp)
    print(f"Average temperature: {avg_value}")  # Average temperature: 72.5

asyncio.run(main())
```

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

### Convert QueryPlan to Pandas DataFrame with Python

Source: https://context7.com/bobthebuidler/generic_exporters/llms.txt

Demonstrates how to use the DataFramer processor to convert a QueryPlan into a pandas DataFrame. This is useful for further data analysis, visualization, or exporting to other formats. The conversion can also be done directly via a query method.

```python
from datetime import datetime, timedelta
from decimal import Decimal
from generic_exporters import Metric, TimeSeries
from generic_exporters.processors import DataFramer

class AnalyticsMetric(Metric):
    @property
    def key(self) -> str:
        return "page_views"

    async def produce(self, timestamp: datetime) -> Decimal:
        return Decimal("1500")

import asyncio

async def main():
    metric = AnalyticsMetric()
    series = TimeSeries(metric)

    start = datetime.utcnow() - timedelta(days=7)
    end = datetime.utcnow()
    query = series[start:end:timedelta(days=1)]

    # Create DataFramer and await to get DataFrame
    framer = DataFramer(query)
    df = await framer

    # Or use the query's built-in method
    df = await query.dataframe()

    print(df.head())
    # Process with pandas
    print(f"Total views: {df['page_views'].sum()}")

asyncio.run(main())
```

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

### SQL Time Series Key-Value Storage with SQLTimeSeriesKeyValueStore

Source: https://context7.com/bobthebuidler/generic_exporters/llms.txt

The SQLTimeSeriesKeyValueStore provides SQL database storage for time series data using Pony ORM. It supports SQLite by default and can be configured for other SQL databases like PostgreSQL. This allows for persistent storage and querying of time series metrics.

```python
from datetime import datetime, timedelta
from decimal import Decimal
from generic_exporters import Metric, TimeSeries
from generic_exporters.processors.exporters import TimeSeriesExporter
from generic_exporters.processors.exporters.datastores.timeseries.sql import SQLTimeSeriesKeyValueStore

class TemperatureMetric(Metric):
    @property
    def key(self) -> str:
        return "temperature"

    async def produce(self, timestamp: datetime) -> Decimal:
        return Decimal("72.5")

import asyncio

async def main():
    # Use default SQLite storage
    datastore = SQLTimeSeriesKeyValueStore()

    # Or configure PostgreSQL
    # datastore = SQLTimeSeriesKeyValueStore(
    #     provider='postgres',
    #     host='localhost',
    #     database='mydb',
    #     user='user',
    #     password='pass'
    # )

    metric = TemperatureMetric()
    series = TimeSeries(metric)

    start = datetime.utcnow() - timedelta(hours=24)
    end = datetime.utcnow()
    query = series[start:end:timedelta(hours=1)]

    # Check if data exists
    ts = datetime.utcnow()
    exists = await datastore.data_exists("temperature", ts)
    print(f"Data exists: {exists}")

    # Export data
    exporter = TimeSeriesExporter(query, datastore)
    await exporter.run()

asyncio.run(main())
```

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

### Generate Line Charts with Plotter Processor

Source: https://context7.com/bobthebuidler/generic_exporters/llms.txt

The Plotter processor generates line chart visualizations from a QueryPlan using bqplot. It can save plots to PNG files or return interactive Figure objects for Jupyter notebooks. It requires bqplot for visualization.

```python
from datetime import datetime, timedelta
from decimal import Decimal
from generic_exporters import Metric, TimeSeries
from generic_exporters.processors import Plotter

class StockPrice(Metric):
    @property
    def key(self) -> str:
        return "AAPL"

    async def produce(self, timestamp: datetime) -> Decimal:
        return Decimal("175.50")

import asyncio

async def main():
    stock = StockPrice()
    series = TimeSeries(stock)

    start = datetime.utcnow() - timedelta(days=30)
    end = datetime.utcnow()
    query = series[start:end:timedelta(days=1)]

    # Create a plotter
    plotter = Plotter(query)

    # Get the figure object
    figure = await plotter.plot()

    # Run to save as PNG (saves to ~/.generic_exporters/plots/)
    await plotter.run()

    # Or use QueryPlan's built-in method
    figure = await query.plot()

asyncio.run(main())
```

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

### Perform Arithmetic Operations on Metrics

Source: https://context7.com/bobthebuidler/generic_exporters/llms.txt

This snippet illustrates how to perform arithmetic operations (+, -, *, /, //, **) on `Metric` objects to create composite metrics. When a composite metric's `produce` method is called, it calculates the values of its operands and applies the operation. The keys of these composite metrics reflect the performed operation.

```python
from datetime import datetime
from decimal import Decimal
from generic_exporters import Metric, Constant

class Revenue(Metric):
    @property
    def key(self) -> str:
        return "revenue"

    async def produce(self, timestamp: datetime) -> Decimal:
        return Decimal("10000")

class Expenses(Metric):
    @property
    def key(self) -> str:
        return "expenses"

    async def produce(self, timestamp: datetime) -> Decimal:
        return Decimal("6000")

import asyncio

async def main():
    revenue = Revenue()
    expenses = Expenses()

    # Create composite metrics
    profit = revenue - expenses
    profit_margin = profit / revenue
    scaled_revenue = revenue * Constant(100)
    growth = revenue ** Constant(2)

    timestamp = datetime.utcnow()

    # Each composite metric computes its value
    print(f"Profit: {await profit.produce(timestamp)}")        # 4000
    print(f"Margin: {await profit_margin.produce(timestamp)}") # 0.4
    print(f"Scaled: {await scaled_revenue.produce(timestamp)}")# 1000000

    # Keys reflect the operation
    print(profit.key)        # (revenue-expenses)
    print(profit_margin.key) # ((revenue-expenses)/revenue)

asyncio.run(main())
```

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

### Export Time Series Data to VictoriaMetrics

Source: https://context7.com/bobthebuidler/generic_exporters/llms.txt

This snippet demonstrates how to export time series data to Victoria Metrics using the `VictoriaMetrics` datastore. It requires configuring the Victoria Metrics URL and optional labels for metric identification. The output is sent to the specified Victoria Metrics endpoint.

```python
from datetime import datetime, timedelta
from decimal import Decimal
from generic_exporters import Metric, TimeSeries
from generic_exporters.processors.exporters import TimeSeriesExporter
from generic_exporters.processors.exporters.datastores.timeseries.victoria import VictoriaMetrics

class ServerMetric(Metric):
    def __init__(self, metric_name: str):
        super().__init__()
        self._key = metric_name

    @property
    def key(self) -> str:
        return self._key

    async def produce(self, timestamp: datetime) -> Decimal:
        return Decimal("95.5")

import asyncio

async def main():
    # Configure Victoria Metrics connection
    datastore = VictoriaMetrics(
        url="http://localhost:8428/api/v1/import",
        key_label_name="__name__",
        extra_labels={
            "env": "production",
            "host": "server-01"
        }
    )

    cpu_metric = ServerMetric("cpu_usage")
    series = TimeSeries(cpu_metric)

    start = datetime.utcnow() - timedelta(hours=1)
    end = datetime.utcnow()
    query = series[start:end:timedelta(minutes=1)]

    # Export to Victoria Metrics
    exporter = TimeSeriesExporter(query, datastore)
    await exporter.run()

asyncio.run(main())
```

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

### Export Time Series Data with TimeSeriesExporter

Source: https://context7.com/bobthebuidler/generic_exporters/llms.txt

The TimeSeriesExporter exports time series data to external datastores. It handles backfilling historical data and supports running continuously for real-time exports. Implement data_exists to check for existing data and avoid duplicates. Requires a custom TimeSeriesDataStoreBase implementation.

```python
from datetime import datetime, timedelta
from decimal import Decimal
from generic_exporters import Metric, TimeSeries
from generic_exporters.processors.exporters import TimeSeriesExporter
from generic_exporters.processors.exporters.datastores.timeseries._base import TimeSeriesDataStoreBase

class MyMetric(Metric):
    @property
    def key(self) -> str:
        return "my_metric"

    async def produce(self, timestamp: datetime) -> Decimal:
        return Decimal("42.0")

# Custom datastore implementation
class MyDataStore(TimeSeriesDataStoreBase):
    def __init__(self):
        self.data = {}

    async def data_exists(self, key, ts: datetime) -> bool:
        return (key, ts) in self.data

    async def push(self, key, ts: datetime, data: Decimal) -> None:
        self.data[(key, ts)] = data
        print(f"Stored {key}={data} at {ts}")

import asyncio

async def main():
    metric = MyMetric()
    series = TimeSeries(metric)

    start = datetime.utcnow() - timedelta(hours=1)
    end = datetime.utcnow()
    query = series[start:end:timedelta(minutes=10)]

    datastore = MyDataStore()

    # Create exporter with 5-minute buffer
    exporter = TimeSeriesExporter(
        query,
        datastore,
        buffer=timedelta(minutes=5),
        concurrency=10
    )

    # Export historical data
    await exporter.run(run_forever=False)

    # Or run continuously for real-time exports
    # await exporter.run(run_forever=True)

asyncio.run(main())
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.