Meerschaum 2¶

The 2.9 series added support for geometry data and improved the web console development experience.

v2.9.5¶

• Add the Query Data dropdown to pipes' cards. Similar to the Recent Data dropdown, you can now explore a pipe's data with the Query Data dropdown. This supports begin, end, params, and limit filtering.

• Additional fixes for in-place syncs on PostGIS. Geometry columns are now correctly coalesced for PostGIS.

v2.9.2 – v2.9.4¶

• Fix in-place syncs with GEOMETRY columns. The query to retrieve GEOMETRY data types for PostGIS has been fixed.

• Consolidate 1-minute chunks in Pipe.get_chunk_bounds(). The chunk bounds returned from Pipe.get_chunk_bounds() (without specifiying an end) now consolidate the 1-minute chunk into the last chunk:

import meerschaum as mrsm
pipe = mrsm.Pipe('demo', 'chunk_bounds', columns={'datetime': 'dt'}, instance='sql:local')
pipe.sync([
    {'dt': '2025-01-01'},
    {'dt': '2025-01-02'},
])
mrsm.pprint(pipe.get_chunk_bounds(bounded=True))
# [
#   (
#       datetime.datetime(2025, 1, 1, 0, 0, tzinfo=datetime.timezone.utc),
#       datetime.datetime(2025, 1, 2, 0, 1, tzinfo=datetime.timezone.utc)
#   )
# ]

• Remove unnecessary CREATE SCHEMA IF EXISTS checks.
• Fix unsupported syntax for Python 3.8.
• Truncate long connector keys in the Shell prompt.
• Fix null location keys behavior for pipes cards.
• Fix autoincrementing primary keys for custom schemas.

v2.9.0 – v2.9.1¶

• Add the dtype geometry (and geography). The new data type geometry adds support for syncing GIS data (e.g. WKB/WKT, GeoJSON, shapely objects, and GeoDataFrames).

import meerschaum as mrsm
pipe = mrsm.Pipe(
    'demo', 'geometry',
    instance='sql:local',
    columns={
        'primary': 'id',
    },
    dtypes={
        'geom': 'geometry',
        'id': 'int',
    },
)

The geometry dtype syntax supports constraints for Geometry type and/or CRS (SRID):

dtypes:
  id: int
  geom: geometry[MultiLineString, 4326]

Syncing a GeoDataFrame (without specifying an explicit dtype) will detect any CRS and geometry type:

import meerschaum as mrsm
import geopandas as gpd

path = 'path/to/shapefile.shp'
gdf = gpd.read_file(path)

pipe = mrsm.Pipe('demo', 'shapefile', instance='sql:local')
success, msg = pipe.sync(gdf)

print(pipe.dtypes['geometry'])
# geometry[Point, 6570]

• Add the SQLConnector flavor postgis. The new flavor postgis (built atop the postgresql flavor) allows pipes to take natively support GEOMETRY (and GEOGRAPHY) types.

• Add a pages sidebar to the Web Console. Clicking the Meerschaum logo on the Web Console will show the pages navigation sidebar, allowing you to easily expand the web app. Custom pages added via @web_page are grouped by plugin. You can override this behavior with the page_group parameter:

from meerschaum.plugins import web_page, dash_plugin

@dash_plugin
def init_dash(dash_app):
    import dash.html as html

    @web_page('users', page_group='Settings', login_required=True)
    def my_new_settings_page():
        return dbc.Container([
            html.P("A new page `/dash/users/` added to the Settings page group.")
        ])

• Add the property instance_keys to api connectors. The optional property instance_keys determines the value of instance_keys to be sent alongside pipe requests.

• Insert the Web Console navbar into custom pages. Custom pages added via @web_page will now include the simple navbar by default, to more tightly integrate custom pages into the Web Console. This behavior may be disabled by setting skip_navbar=True in @web_page:

# example.py

from meerschaum.plugins import web_page, dash_plugin

@dash_plugin
def init_app(dash_app):
    import dash_bootstrap_components as dbc
    import dash.html as html

    @web_page('example-page', skip_navbar=True)
    def example_page():
        return dbc.Container(html.H1('Hello, world!'))

• Create INT columns for dtypes int32, SMALLINT for int16. The SQLConnector now maps the Pandas dtypes int32 to INT and int16 (and int8) to SMALLINT rather than defaulting to BIGINT for everything.

• Fix serialization of valkey pipes without indices. Pipes synced without columns now correctly serialize documents' keys.

• Add API endpoints for clearing pipes and chunk bounds. The endpoints /pipes/{connector_keys}/{metric_key}/{location_key}/clear and /pipes/{connector_keys}/{metric_key}/{location_key}/chunk_bounds now allow API users to clear pipes (rather than using the legacy actions endpoint) and get the values from pipe.get_chunk_bounds().

• Skip venv locking on Windows.

• Shrink full Docker image size.

The 2.8 series introduced batches to verify pipes as well as more granular control over exposed instances via the API.

v2.8.4¶

• Allow for pattern matching in allowed_instance_keys. You may now generalize the instances exposed by the API by using Unix-style patterns in the list system:api:permissions:instances:allowed_instance_keys:

{
  "api": {
    "permissions": {
      "instances": {
        "allowed_instance_keys": [
          "valkey:*",
          "*_dev"
        ]
      }
    }
  }
}

• Return pipe attributes for the route /pipes/{connector}/{metric}/{location}. The API routes /pipes/{connector}/{metric}/{location} and /pipes/{connector}/{metric}/{location}/attributes both return pipe attributes.

• Check entire batches for verify rowcounts. The command verify rowcounts will now check batch boundaries before checking row-counts for individual chunks. This should moderately increase performance.

• Kill orphaned child processes when the parent job is killed. Jobs created with pipeline arguments should now kill associated child processes.

• Add --skip-hooks. The flag --skip-hooks prevents any sync hooks from firing when syncing pipes.

• Remove datetime rounding from parse_schedule(). Scheduled actions now behave as expected, the current timestamp is no longer rounded to the nearest minute, which was causing issues with the starting in delay feature.

• Fix allowed_instance_keys enforcement.

v2.8.3¶

• Increase username limit to 60 characters.
• Add chunk retries to Pipe.verify().
• Add instance keys to remaining pipes endpoints.
• Misc bugfixes.

v2.8.0 – v2.8.2¶

• Add batches to Pipe.verify(). Verification syncs now run in sequential batches so that they may be interrupted and resumed. See Pipe.get_chunk_bounds_batches() for more information:

from datetime import timedelta
import meerschaum as mrsm

pipe = mrsm.Pipe('demo', 'get_chunk_bounds', instance='sql:local')
bounds = pipe.get_chunk_bounds(
    chunk_interval=timedelta(hours=10),
    begin='2025-01-10',
    end='2025-01-15',
    bounded=True,
)
batches = pipe.get_chunk_bounds_batches(bounds, workers=4)
mrsm.pprint(
    [
        tuple(
            (str(bounds[0]), str(bounds[1]))
            for bounds in batch
        )
        for batch in batches
    ]
)
# [
#     (
#         ('2025-01-10 00:00:00+00:00', '2025-01-10 10:00:00+00:00'),
#         ('2025-01-10 10:00:00+00:00', '2025-01-10 20:00:00+00:00'),
#         ('2025-01-10 20:00:00+00:00', '2025-01-11 06:00:00+00:00'),
#         ('2025-01-11 06:00:00+00:00', '2025-01-11 16:00:00+00:00')
#     ),
#     (
#         ('2025-01-11 16:00:00+00:00', '2025-01-12 02:00:00+00:00'),
#         ('2025-01-12 02:00:00+00:00', '2025-01-12 12:00:00+00:00'),
#         ('2025-01-12 12:00:00+00:00', '2025-01-12 22:00:00+00:00'),
#         ('2025-01-12 22:00:00+00:00', '2025-01-13 08:00:00+00:00')
#     ),
#     (
#         ('2025-01-13 08:00:00+00:00', '2025-01-13 18:00:00+00:00'),
#         ('2025-01-13 18:00:00+00:00', '2025-01-14 04:00:00+00:00'),
#         ('2025-01-14 04:00:00+00:00', '2025-01-14 14:00:00+00:00'),
#         ('2025-01-14 14:00:00+00:00', '2025-01-15 00:00:00+00:00')
#     )
# ]

• Add --skip-chunks-with-greater-rowcounts to verify pipes. The flag --skip-chunks-with-greater-rowcounts will compare a chunk's rowcount with the rowcount of the remote table and skip if the chunk is greater than or equal to the remote count. This is only applicable for connectors which implement remote=True support for get_sync_time().

• Add verify rowcounts. The action verify rowcounts (same as passing --check-rowcounts-only to verify pipes) will compare row-counts for a pipe's chunks against remote rowcounts. This is only applicable for connectors which implement get_pipe_rowcount() with support for remote=True.

• Add remote to pipe.get_sync_time(). For pipes which support it (i.e. the SQLConnector), the option remote is intended to return the sync time of a pipe's fetch definition, like the option remote in Pipe.get_rowcount().

• Allow for the Web API to serve pipes from multiple instances. You can disable this behavior by setting system:api:permissions:instances:allow_multiple_instances to false. You may also explicitly allow which instances may be accessed by the WebAPI by setting the list system:api:permissions:instances:allowed_instance_keys (defaults to ["*"]).

• Fix memory leak for retrying failed chunks. Failed chunks were kept in memory and retried later. In resource-intensive syncs with large chunks and high failures, this would result in large objects not being freed and hogging memory. This situation has been fixed.

• Add negation to job actions. Prefix a job name with an underscore to select all other jobs. This is useful for filtering out noise for show logs.

• Add Pipe.parent. As a quality-of-life improvement, the attribute Pipe.parent will return the first member of Pipe.parents (if available).

• Use the current instance for new tabs in the Webterm. Clicking "New Tab" will open a new tmux window using the currently selected instance on the Web Console.

• Other webterm quality-of-life improvements. Added a size toggle button to allow for the webterm to take the entire page.

• Additional refactoring work. The API endpoints code has been cleaned up.

• Added system configurations. New options have been added to the system configuration, such as max_response_row_limit, allow_multiple_instances, allowed_instance_keys.

The 2.7 series greatly improved indexing, numerics support, added the bytes type, and allowed for bypassing dtype enforcement (Pipe.enforce) as well as introducing persistent Webterm sessions.

v2.7.9 – v2.7.10¶

• Add persistent Webterm sessions. On the Web Console, the Webterm will attach to a persistent terminal for the current session's user.

• Reconnect Webterms after client disconnect. If a Webterm socket connection is broken, the client logic will attempt to reconnect and attach to the tmux session.

• Add tmux sessions to Webterms. Webterm sessions now connect to tmux sessions (tied to the user accounts). Set system:webterm:tmux:enabled to false to disable tmux sessions.

• Limit concurrent connections during verify pipes. To keep from exhausting the SQL connection pool, limit the number of concurrent intra-chunk connections.

• Return the precision and scale from a table's columns and types. Reading a table's columns and types with meerschaum.utils.sql.get_table_columns_types() now returns the precision and scale for NUMERIC (DECIMAL) columns.

v2.7.8¶

• Add support for user-supplied precision and scale for numeric columns. You may now manually specify a numeric column's precision and scale:

import meerschaum as mrsm

pipe = mrsm.Pipe(
    'demo', 'numeric', 'precision_scale',
    instance='sql:local',
    dtypes={'val': 'numeric[5,2]'},
)
pipe.sync([{'val': '123.456'}])
print(pipe.get_data())
#       val
# 0  123.46

• Serialize numeric columns to exact values during bulk inserts. Decimal values are serialized when inserting into NUMERIC columns during bulk inserts.

• Return a generator when fetching with SQLConnector. To alleviate memory pressure, skip loading the entire dataframe when fetching.

• Add json_serialize_value() to handle custom dtypes. When serializing documents, pass json_serialize_value as the default handler:

import json
from decimal import Decimal
from datetime import datetime, timezone
from meerschaum.utils.dtypes import json_serialize_value

print(json.dumps(
    {
        'bytes': b'hello, world!',
        'decimal': Decimal('1.000000001'),
        'datetime': datetime(2025, 1, 1, tzinfo=timezone.utc),
    },
    default=json_serialize_value,
    indent=4,
))
# {
#     "bytes": "aGVsbG8sIHdvcmxkIQ==",
#     "decimal": "1.000000001",
#     "datetime": "2025-01-01T00:00:00+00:00"
# }

• Fix an issue with the WITH keyword in pipe definitions for MSSQL. Previously, pipes with used with keyword WITH but not as a CTE (e.g. to specify an index) were incorrectly parsed.

v2.7.7¶

• Add actions drop indices and index pipes. You may now drop and create indices on pipes with the actions drop indices and index pipes or the pipe methods drop_indices() and create_indices():

import meerschaum as mrsm

pipe = mrsm.Pipe('demo', 'drop_indices', columns=['id'], instance='sql:local')
pipe.sync([{'id': 1}])
print(pipe.get_columns_indices())
# {'id': [{'name': 'IX_demo_drop_indices_id', 'type': 'INDEX'}]}

pipe.drop_indices()
print(pipe.get_columns_indices())
# {}

pipe.create_indices()
print(pipe.get_columns_indices())
# {'id': [{'name': 'IX_demo_drop_indices_id', 'type': 'INDEX'}]}

• Remove CAST() to datetime with selecting from a pipe's definition. For some databases, casting to the same dtype causes the query optimizer to ignore the datetime index.

• Add INCLUDE clause to datetime index for MSSQL. This is to coax the query optimizer into using the datetime axis.

• Remove redundant unique index. The two competing unique indices have been combined into a single index (for the key unique). The unique constraint (when upsert is true) shares the name but has the prefix UQ_ in place of IX_.

• Add pipe parameter null_indices. Set the pipe parameter null_indices to False for a performance improvement in situations where null index values are not expected.

• Apply backtrack minutes when fetching integer datetimes. Backtrack minutes are now applied to pipes with integer datetimes axes.

v2.7.6¶

• Make temporary table names configurable. The values for temporary SQL tables may be set in MRSM{system:connectors:sql:instance:temporary_target}. The new default prefix is '_', and the new default transaction length is 4. The values have been re-ordered to target, transaction ID, then label.

• Add connector completions to copy pipes. When copying pipes, the connector keys prompt will offer auto-complete suggestions.

• Fix stale job results. When polling for job results, the job result is dropped from in-memory cache to avoid overwriting the on-disk result.

• Format row counts and seconds into human-friendly text. Row counts and sync durations are now formatted into human-friendly representations.

• Add digits to generate_password(). Random strings from meerschaum.utils.misc.generate_password() may now contain digits.

v2.7.3 – v2.7.5¶

• Allow for dynamic targets in SQL queries. Include a pipe definition in double curly braces (à la Jinja) to substitute a pipe's target into a templated query.

import meerschaum as mrsm

pipe = mrsm.Pipe('demo', 'template', target='foo', instance='sql:local')
_ = pipe.register()

downstream_pipe = mrsm.Pipe(
    'sql:local', 'template',
    instance='sql:local',
    parameters={
        'sql': "SELECT *\nFROM {{Pipe('demo', 'template', instance='sql:local')}}"
    },
)

conn = mrsm.get_connector('sql:local')
print(conn.get_pipe_metadef(downstream_pipe))
# WITH "definition" AS (
#     SELECT *
#     FROM "foo"
# )
# SELECT *
# FROM "definition"

• Add --skip-enforce-dtypes. To override a pipe's enforce parameter, pass --skip-enforce-dtypes to a sync.

• Add bulk inserts for MSSQL. To disable this behavior, set system:connectors:sql:bulk_insert:mssql to false. Bulk inserts for PostgreSQL-like flavors may now be disabled as well.

• Fix altering multiple column types for MSSQL. When a table has multiple columns to be altered, each column will have its own ALTER TABLE query.

• Skip enforcing custom dtypes when enforce=False. To avoid confusion, special Meerschaum data types (numeric, json, etc.) are not coerced into objects when enforce=False.

• Fix timezone-aware casts. A bug has been fixed where it was possible to mix timezone-aware and -naive casts in a single query. This patch ensures that this no longer occurs.

• Explicitly cast timezone-aware datetimes as UTC in SQL syncs. By default, timezone-aware columns are now cast as time zone UTC in SQL. This may be skipped by setting enforce to False.

• Added virtual environment inter-process locks. Competing processes now cooperate for virtual environment verification, which protects installed packages.

v2.7.0 – v2.7.2¶

• Introduce the bytes data type. Instance connectors which support binary data (e.g. SQLConnector) may now take advantage of the bytes dtype. Other connectors (e.g. ValkeyConnector) may use meerschaum.utils.dtypes.serialize_bytes() to store binary data as a base64-encoded string.

import meerschaum as mrsm

pipe = mrsm.Pipe(
    'demo', 'bytes',
    instance='sql:memory',
    dtypes={'blob': 'bytes'},
)
pipe.sync([
    {'blob': b'hello, world!'},
])

df = pipe.get_data()
binary_data = df['blob'][0]
print(binary_data.decode('utf-8'))
# hello, world!

from meerschaum.utils.dtypes import serialize_bytes, attempt_cast_to_bytes
df['encoded'] = df['blob'].apply(serialize_bytes)
df['decoded'] = df['encoded'].apply(attempt_cast_to_bytes)
print(df)
#                blob               encoded           decoded
# 0  b'hello, world!'  aGVsbG8sIHdvcmxkIQ==  b'hello, world!'

• Allow for pipes to use the same column for datetime, primary, and autoincrement=True. Pipes may now use the same column as the datetime axis and primary with autoincrement set to True.

pipe = mrsm.Pipe(
    'demo', 'datetime_primary_key', 'autoincrement',
    instance='sql:local',
    columns={
        'datetime': 'Id',
        'primary': 'Id',
    },
    autoincrement=True,
)

• Only join on primary when present. When the index primary is set, use the column as the primary joining index. This will improve performance when syncing tables with a primary key.

• Add the parameter enforce. The parameter enforce (default True) toggles data type enforcement behavior. When enforce is False, incoming data will not be cast to the desired data types. For static datasets where the incoming data is always expected to be of the correct dtypes, then it is recommended to set enforce to False and static to True.

from decimal import Decimal
import meerschaum as mrsm

pipe = mrsm.Pipe(
    'demo', 'enforce',
    instance='sql:memory',
    enforce=False,
    static=True,
    autoincrement=True,
    columns={
        'primary': 'Id',
        'datetime': 'Id',
    },
    dtypes={
        'Id': 'int',
        'Amount': 'numeric',
    },
)
pipe.sync([
    {'Amount': Decimal('1.11')},
    {'Amount': Decimal('2.22')},
])

df = pipe.get_data()
print(df)
#    Id Amount
# 0   1   1.11
# 1   2   2.22

• Create the datetime axis as a clustered index for MSSQL, even when a primary index is specififed. Specifying a datetime and primary index will create a nonclustered PRIMARY KEY. Specifying the same column as both datetime and primary will create a clustered primary key (tip: this is useful when autoincrement=True).

• Increase the default chunk interval to 43200 minutes. New hypertables will use a default chunksize of 30 days (43200 minutes).

• Virtual environment bugfixes. Existing virtual environment packages are backed up before re-initializing a virtual environment. This fixes the issue of disappearing dependencies.

• Store numeric as TEXT for SQLite and DuckDB. Due to limited precision, numeric columns are now stored as TEXT, then parsed into Decimal objects upon retrieval.

• Show the Webterm by default when changing instances. On the Web Console, changing the instance select will make the Webterm visible.

• Improve dtype inference.

The 2.6 series added the primary index, autoincrement, and migrated to timezone-aware datetimes by default, as well as many quality-of-life improvements, especially for MSSQL.

v2.6.17¶

• Add relative deltas to starting in scheduler syntax. You may specify a delta in the job scheduler starting syntax:

mrsm sync pipes -s 'daily starting in 30 seconds'

• Fix drop pipes for pipes on custom schemas. Pipes created under a specific schema are now correctly dropped.

• Enhance editing pipeline jobs. Pipeline jobs now provide the job label as the default text to be edited. Pipeline arguments are now placed on a separate line to improve legibility.

• Disable the progress timer for jobs. The sync pipes progress timer will now be hidden when running through a job.

• Unset MRSM_NOASK for daemons. Now that jobs may accept user input, the environment variable MRSM_NOASK is no longer needed for jobs run as daemons (executor local).

• Replace Cx_Oracle with oracledb. The Oracle SQL driver is no longer required now that the default Python binding for Oracle is oracledb.

• Fix Oracle auto-incrementing for good. At long last, the mystery of Oracle auto-incrementing identity columns has been laid to rest.

v2.6.15 – v2.6.16¶

• Fix inplace syncs without a datetime axis. A bug introduced by a performance optimization has been fixed. Inplace pipes without a datetime axis will skip searching for date bounds. Setting upsert to true will bypass this bug for previous releases.

• Skip invoking get_sync_time() for pipes without a datetime axis. Invoking an instance connector's get_sync_time() method will now only occur when datetime is set.

• Remove guess_datetime() check from SQLConnector.get_sync_time(). Because sync times are only checked for pipes with a dedicated datetime column, the guess_datetime() check has been removed from the SQLConnector.get_sync_time() method.

• Skip persisting default target to parameters. The default target table name will no longer be persisted to parameters. This helps avoid accidentally setting the wrong target table when copying pipes.

• Default to "no" for syncing data when copying pipes. The action copy pipes will no longer sync data by default, instead requiring an explicit yes to begin syncing.

• Fix the "Update query" button behavior on the Web Console. Existing but null keys are now accounted for when update a SQL pipe's query.

• Fix another Oracle autoincrement edge case. Resetting the autoincrementing primary key value on Oracle will now behave as expected.

v2.6.10 – v2.6.14¶

• Improve datetime timezone-awareness enforcement performance. Datetime columns are only parsed for timezone awareness if the desired awareness differs. This drastically speeds up sync times.

• Switch to tz_localize() when stripping timezone information. The previous method of using a lambda to replace individual tzinfo attributes did not scale well. Using tz_localize() can be vectorized and greatly speeds up syncs, especially with large chunks.

• Add enforce_dtypes to Pipe.filter_existing(). You may optionally enforce dtype information during filter_existing(). This may be useful when implementing custom syncs for instance connectors. Note this may impact memory and compute performance.

import meerschaum as mrsm
import pandas as pd

pipe = mrsm.Pipe('a', 'b', instance='sql:local')
pipe.sync([{'a': 1}])

df = pd.DataFrame([{'a': '2'}])

### `enforce_dtypes=True` will suppress the differing dtypes warning.
unseen, update, delta = pipe.filter_existing(df, enforce_dtypes=True)
print(delta)

• Fix query_df() for null parameters. This is useful for when you may use query_df() with only select_columns or omit_columns.

• Fix autoincrementing IDs for Oracle SQL.

• Enforce security settings for creating jobs. Jobs and remote actions will only be accessible to admin users when running with --secure (system:permissions:actions:non_admin in config).

v2.6.6 – v2.6.9¶

• Improve metadata performance when syncing. Syncs via the SQLConnector now cache schema and index metadata, speeding up transactions.

• Fix upserts for MySQL / MariaDB. Upserts in MySQL and MariaDB now use ON DUPLICATE instead of REPLACE INTO.

• Fix dtype detection for index columns. A bug where new index columns were incorrectly created as INT has been fixed.

• Delete old keys when dropping Valkey pipes. Dropping a pipe from Valkey now clears all old index keys.

• Fix timezone-aware enforcement bugs.

v2.6.1 – v2.6.5¶

• Add Pipe.tzinfo. Check if a pipe is timezone-aware with tzinfo:

import meerschaum as mrsm

pipe = mrsm.Pipe(
    'demo', 'timezone', 'aware',
    columns={'datetime': 'dt'},
)
print(pipe.tzinfo)
# UTC

pipe = mrsm.Pipe(
    'demo', 'timezone', 'naive',
    columns={'datetime': 'dt'},
    dtypes={'dt': 'datetime64[ns]'},
)
print(pipe.tzinfo)
# None

• Improve timezone enforcement when syncing.
• Fix inplace syncs with upsert=True.
• Fix timezone-aware datetime truncation for MSSQL
• Fix timezone detection for existing timezone-naive tables.

v2.6.0¶

• Enforce a timezone-aware datetime axis by default. Pipes now enforce timezone-naive datetimes as UTC, even if the underlying column type is timezone-naive. To use datetime-naive datetime axes, you must explicitly set the dtype to datetime64[ns].

• Designate the index name primary for primary keys. Like the datetime index, the primary index is used for joins and will be created as the primary key in new tables.

import meerschaum as mrsm

pipe = mrsm.Pipe(
    'demo', 'primary_key',
    columns={'primary': 'pk'},
    instance='sql:local',
)
### Raises a `UNIQUE` constraint failure:
pipe.sync([
    {'pk': 1},
    {'pk': 1},
])

• Add autoincrement to Pipe.parameters Like upsert, you may designate an incremental integer primary key by setting autoincrement to True in the pipe parameters. Note that autoincrement will be True if you specify a primary index but do not specify a dtype or pass into the initial dataframe. This is only available for sql pipes.

import meerschaum as mrsm

pipe = mrsm.Pipe(
    'demo', 'autoincrement',
    columns={'primary': 'id'},
    autoincrement=True,
    instance='sql:local',
)
pipe.sync([
    {'color': 'red'},
    {'color': 'blue'},
])
print(pipe.get_data())
#    id color
# 0   1   red
# 1   2  blue

pipe.sync([
    {'color': 'green', 'id': 1},
])
print(pipe.get_data())
#    id  color
# 0   1  green
# 1   2   blue

• Add option static to Pipe.parameters to disable schema modification. Set static to True in a pipe's parameters to prevent any modification of the column's data types.

import meerschaum as mrsm

pipe = mrsm.Pipe(
    'demo', 'static',
    columns={
        'primary': 'id',
        'datetime': 'id',
    },
    dtypes={
        'meta': 'json',
        'id': 'int',
    },
    parameters={
        'static': True,
        'upsert': True,
        'autoincrement': True,
    },
)
pipe.sync([{'id': 1, 'meta': {'foo': 'bar'}}])

### Abort syncing new columns when `static` is set.
pipe.sync([{'id': 1, 'color': 'blue'}])

mrsm.pprint(pipe.get_columns_types())
# {'id': 'BIGINT', 'meta': 'JSONB'}

• Add get_create_table_queries() to build from dtypes dictionaries. You may get a CREATE TABLE query from a dtypes dictionary (in addition to a SELECT query). The function meerschaum.utils.sql.get_create_table_query() now also accepts an arguments primary_key and autoincrement to designate a primary key column.

from meerschaum.utils.sql import get_create_table_queries

queries = get_create_table_queries(
    {
        "id": "int",
        "is_active": "bool",
        "customer_id": "uuid",
        "metadata": "json",
    },
    'new_table',
    'postgresql',
    primary_key="id",
    autoincrement=True,
)
print(queries[0])
# CREATE TABLE "new_table" (
#     "id" BIGINT PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY NOT NULL,
#     "is_active" BOOLEAN,
#     "customer_id" UUID,
#     "metadata" JSONB
# )

• Create a multi-column index for columns defined in Pipe.columns. To disable this behavior, set unique to None in Pipe.indices.

• Default to BIT for boolean columns in MSSQL. The previous workaround was to store bool columns as INT. This change now defaults to BIT when creating new tables. Boolean columns cannot be nullable for MSSQL.

• Improve file protection in edit config. Writing an invalid config file will now stop you before committing the changes. The previous behavior would lead to data loss.

• Catch exceptions when creating chunk labels. If a datetime bound cannot be determined for a chunk, return pd.NA.

The 2.5.x series was short and sweet, primarily introducing features relating to Pipe.indices.

v2.5.1¶

• Update index information in the pipe card. The Indices section of the pipe card on the web console includes more detailed information, such as composite and multi-column indices.

• Print action results during scheduled jobs. Scheduled actions now print their result success tuples after firing.

• Other bugfixes. A few bugs from the migration of APScheduler to internal management have been fixed.

v2.5.0¶

• Add indices to Pipe.parameters. You may now explicitly state the indices to be created by defining indices (or indexes) in Pipe.parameters (or the Pipe constructor for your convenience).

import meerschaum as mrsm

pipe = mrsm.Pipe(
    'software', 'versions',
    instance='sql:local',
    columns=['major', 'minor', 'patch'],
    indices={
        'all': ['major', 'minor', 'patch'],
    },
)
mrsm.pprint(pipe.get_indices())
# {
#     'major': 'IX_software_versions_major',
#     'minor': 'IX_software_versions_minor',
#     'patch': 'IX_software_versions_patch',
#     'all': 'IX_software_versions_major_minor_patch'
# }

You may also use the key index_template to change the format of the generated index names (defaults to IX_{target}_{column_names}, where target is the table name and column_names consists of all of the index's columns joined by an underscore).

pipe.parameters['index_template'] = "foo_{target}_{column_names}"
mrsm.pprint(pipe.get_indices())
# {'all': 'foo_software_versions_major_minor_patch'}

• Enable chunking for MSSQL To improve memory usage, chunksize is now accepted by SQLConnector.read() for the flavor mssql.

• Disable pyodbc pooling. To properly recycle the connection pool in the SQLAlchemy engine, the internal pyodbc pooling must be disabled. See the SQLAlchemy documentation.

• Bugfixes Other miscellaneous bugfixes have been included in this release, such as resolving broken imports during certain edge cases.

The 2.4.x series added the ValkeyConnector, relative --begin and --end, pipeline timeouts, and improved MSSQL support.

v2.4.13¶

• Add --timeout to pipeline arguments. You may now designate the maximum number of seconds to run a pipeline with --timeout. This will run the entire pipeline in a subprocess rather than a persistent session.

mrsm sync pipes + clear pipes --end '1 month ago' : -s daily --timeout 3600

• Add auto-complete to edit jobs and bootstrap jobs.

• Improve the editing experience for edit jobs and bootstrap jobs.

• Fixed plugin detection for Python 3.9.

v2.4.12¶

• Add the actions edit jobs and bootstrap jobs. The action edit jobs lets you easily tweak the arguments for an existing job, so there's no need to delete and recreate jobs. The bootstrap jobs wizard also gives you a chance to review your changes before starting a job.

• Fix nested CTEs for MSSQL. Pipes may now use definitions containing a WITH clause for Microsoft SQL Server.

• Added wrap_query_with_cte to meerschaum.utils.sql. Reference a subquery in an encapsulating parent query, even if the subquery contains CTEs itself.

from meerschaum.utils.sql import wrap_query_with_cte

sub_query = """
WITH [foo] AS (
  SELECT 1 AS [val]
)
SELECT ([val] * 2) AS [newval]
FROM [foo]
"""

parent_query = "SELECT (newval * 3) FROM [src]"

query = wrap_query_with_cte(
    sub_query,
    parent_query,
    'mssql',
    cte_name='src',
)
print(query)
# WITH [foo] AS (
#   SELECT 1 AS [val]
# ),
# [src] AS (
# SELECT ([val] * 2) AS [newval]
# FROM [foo]
#
# )
# SELECT (newval * 3) FROM [src]

• Fix --yes when running in background jobs. The flags --yes and --noask are now properly handled when running a background job which contains prompts.

• Add an external page for jobs to the Web Console. Like the shareable /pipes/ links, you may now link to a specific job at the path /dash/job/{name}. Click the name of the job on the card to open a job in a new tab.

• Preserve the original values for --begin and --end. When creating jobs in the shell, the original string values for --begin and --end will be preserved, such as in the case of --begin 1 month ago.

• Fix Pipe formatting for small terminals. Pipes with long names are now properly rendered in small terminal windows.

• Enable shell suggestions for chained actions. The shell auto-complete now works with chained actions.

v2.4.9 – v2.4.11¶

• Add relative formats to --begin and --end. The flags --begin and --end support values in the format [N] [unit] ago:

mrsm sync pipes --begin '3 days ago'

Add a second delta format (recommended to be denoted by the keyword rounded) to round the timestamp to a clean value:

mrsm clear pipes --end '1 month ago rounded 1 day'

Supported units are seconds, minutes, hours, days, weeks, months (ago only), and years.

• Respect --begin, --end, and --params in show rowcounts. The flags --begin, --end, and --params are now handled in the action show rowcounts.

• Fix an issue with Pipe.get_backtrack_data(). An incorrect calculation was fixed the produce the correct backtrack interval.

v2.4.8¶

• Allow for syncing against DATETIMEOFFSET columns in MSSQL. When syncing an existing table with a DATETIMEOFFSET column, timestamps correctly coerced into timezone-naive UTC timestamps. Note this behavior will likely change to timezone-aware-by-default in a future release.

• Default to DATETIME2 for MSSQL. To preserve precision, MSSQL now creates datetime columns as DATETIME2.

• Remove temporary tables warnings. Failure to drop temporary tables no longer raises a warning (added IF EXISTS check).

• Fix an issue with the sql action.

• Fix UUID support for SQLite, MySQL / MariaDB.

• Set IS_THREAD_SAFE to False for Oracle.

v2.4.6 – v2.4.7¶

• Prefix temporary tables with ##. Temporary tables are now prefixed with ## to take advantage of tempdb in MSSQL.

• Add the uuid dtype. The uuid dtype adds support for Python UUID objects and maps to the appropriate UUID data type per SQL flavor (e.g. UNIQUEIDENTIFIER for mssql).

• Add upsert support to MSSQL. Setting upsert in a pipe's parameters will now upsert rows in a single transaction (via a MERGE query).

• Add SQLConnector.get_connection(). To simplify connection management, you may now obtain an active connection with SQLConnector.get_connection(). To force a new connection, pass rebuild=True.

• Improve session management for MSSQL. Transactions and connections are now more gracefully handled when working with MSSQL.

v2.4.2 – v2.4.5¶

• Fix bootstrap connectors. Revert a breaking change to the bootstrap connectors wizard.

• Respect disabling uv for package installation. Setting system:experimental:uv_pip to false will now disable uv for certain.

• Default to a query string for options when bootstrapping MSSQL connectors. Although dictionaries are supported for options, using a dictionary as a default was breaking serialization. The default for options is now the string driver=ODBC Driver 17 for SQL Server&UseFMTONLY=Yes.

• Default to MSSQL ODBC Driver 18. The default driver to be used by MSSQL connectors is version 18.

v2.4.1¶

• Add instance to the external pipe links. When sharing pipe links on the Web Console, the instance will now be included in the URL.

• Fix an issue with remote actions. An import error has been patched.

v2.4.0¶

• Add valkey instance connectors. Introducing a new first-class instance connector: the ValkeyConnector. Valkey, a fork of Redis, is a high-performance in-memory database often used for caching. The valkey service has been added to the Meerschaum stack and is accessible via the built-in connector valkey:main.

• Cache Web Console sessions in Valkey when running with --production. Starting the web API with --production will now store sessions in valkey:main. This results in a smoother experience in the event of a web server restart. By default, sessions expire after 30 days.

  You may disable this behavior by setting system:experimental:valkey_session_cache to false.

• Allow for a default executor. Setting the key meerschaum:executor will set the default executor (overriding the check for systemd). This is useful for defaulting to remote actions in a multi-node deployment.

• Allow querying for None in query_df(). You may now query for null rows:

import pandas as pd
from meerschaum.utils.dataframe import query_df

df = pd.DataFrame({'a': [1, 2, pd.NA]}).astype('Int64')

result = query_df(df, {'a': None})
print(result)
#       a
# 2  <NA>

result = query_df(df, {'a': [None, 1]})
print(result)
#       a
# 0     1
# 2  <NA>

• Improve query_df() performance. Dataframe vlues are no longer serialized by default in query_df(), meaning that parameters must match the data type. Pass coerce_types=True to restore legacy behavior.

• Add Pipe.copy_to(). Copy pipes between instances with Pipe.copy_to():

import meerschaum as mrsm

pipe = mrsm.Pipe('plugin:noaa', 'weather')
pipe.copy_to('valkey:main')

• Add include_unchanged_columns to Pipe.filter_existing(). Pass include_unchanged_columns=True to return entire documents in the update dataframe. This is useful for situations where you are unable to update individual fields:

import meerschaum as mrsm
import pandas as pd

pipe = mrsm.Pipe('a', 'b', columns=['id'])
pipe.sync([
    {'animal': 'cat', 'name': 'Meowth', 'id': 1},
    {'animal': 'dog', 'name': 'Fluffy', 'id': 2},
])

df = pd.DataFrame([{'id': 1, 'breed': 'tabby'}])

unseen, update, delta = pipe.filter_existing(df)
print(update)
#    id  breed
# 0   1  tabby

unseen, update, delta = pipe.filter_existing(
    df,
    include_unchanged_columns=True,
)
print(update)
#   animal    name  id  breed
# 0    cat  Meowth   1  tabby

• Add a share button to the Pipe card. On the web dashboard, you may now more easily share pipes by clicking the "share" icon and copying the URL. This opens the pipe card in a new, dedicated tab.

• Add OPTIONAL_ATTRIBUTES to connectors. Connectors may now set OPTIONAL_ATTRIBUTES, which will add skippable prompts in bootstrap connector.

• Remove progress bar for syncing via remote actions. Executing sync pipes remotely will no longer print the timer progress bar.

• Fix bug with stack in the shell. Note that stack actions may not be chained.

• Fix scheduler dependency. To fix the installation of APScheduler, attrs is now held back to 24.1.0.

The 2.3 series was short but brought significant improvements, notably the Job API, remote jobs, and action chaining.

v2.3.5 – v2.3.6¶

• Properly handle remote jobs. Long-running remote jobs are now properly handled, allowing for graceful API shutdown.

• Detect when creating a remote pipeline. Running a pipeline action with a remote executor will pass through the pipeline to the API server:

mrsm show version + show arguments : --loop -e api:main

• Remove actions websocket endpoint with temporary jobs.

• Properly quote environment variables in systemd services.

• Remove ~/.local volume from api service in the stack. This was overwriting the new Docker image version and such needed to be removed.

v2.3.0 – v2.3.4¶

• Add the Job class. You may now manage jobs with Job:

import meerschaum as mrsm
job = mrsm.Job('syncing-engine', 'sync pipes --loop')
job.start()

If you are running on systemd, jobs will be created as user services. Otherwise (e.g. running in Docker) jobs are created as Unix daemons and kept alive by the API server.

You may choose the executor with -e (--executor-keys). Supported values are local, systemd, and the keys for any API instance. See the jobs documentation for more information.

• Chain actions with +. Run multiple commands by joining them with +, similar to && in bash but with better performance (one process).

$ mrsm show pipes + sync pipes

Adding -d (--daemon) will escape these joiners and run all of the chained commands in the job:

$ mrsm show pipes + sync pipes --loop -d

• Run chained actions as a pipeline with :. You can schedule chained actions by adding : to the end of your command:

mrsm sync pipes -i sql:local + sync pipes : -s 'daily starting 00:00' -d

Other supported flags are --loop, --min-seconds, and the number of times to run the pipeline (e.g. x2 or 2):

mrsm sync pipes + verify pipes : --loop --min-seconds 600

mrsm show pipes + sync pipes : x2

• Add --restart. Your job will be automatically restarted if you use any of flags --loop, --schedule, or --restart.

• Execute actions remotely. You may execute an action on an API instance by setting the executor to the connector keys. You may run the executor command in the Meercshaum shell (like instance) or pass the flag -e (--executor-keys).

mrsm sync pipes -e api:main

The output is streamed directly from the API instance (via a websocket).

• Add from_plugin_import(). You may now easily access attributes from a plugin's submodule with meerschaum.plugins.from_plugin_import().

from meerschaum.plugins import from_plugin_import

get_defined_pipes = from_plugin_import('compose.utils.pipes', 'get_defined_pipes')

The 2.2.x series introduced new features improvements, such as the improved scheduler, the switch to uv, the @dash_plugin and @web_page() decorators, and much more.

v2.2.7¶

• Fix daemon stability. Broken file handlers are now better handled, and this should keep background jobs from crashing.

• Improve show jobs output. The show jobs table now includes the SuccessTuple of the most recent run (when jobs are stopped).

• Use a plugin's __doc__ string as the default description. When registering a new plugin, the __doc__ string will be used as the default value for the description.

• Pipes without connectors are no longer considered errors when syncing. When a pipe has an ordinary string in place of a connector (e.g. externally managed), return early and consider success rather than throwing an error.

import meerschaum as mrsm
pipe = mrsm.Pipe('a', 'b')
pipe.sync()
# (True, "Pipe('a', 'b') does not support fetching; nothing to do.")

• Add update announcements. When new Meerschaum releases become available, you will now be presented with an update message when starting the shell. Update checks may be disabled by setting shell:updates:check_remote to false.

• Enforce a 10-minute max timeout for APIConnectors.

v2.2.6¶

• Fix a critical login issue. The previous release (v2.2.5) broke the login functionality of the Web UI and has been yanked. If you are running v2.2.5, it is urgent that you upgrade immediately.

• Add environment variable MRSM_CONFIG_DIR. You may now isolate your configuration directory outside of the root (like with MRSM_PLUGINS_DIR, and MRSM_VENVS_DIR). This will be useful in certain production deployments where secrets need to be segmented and isolated.

• Add register connector. Like bootstrap connector, you may now programmatically create connectors.

• Allow for job names to contain spaces and parentheses. Jobs may now be created with more dynamic names. This issue in particular affected Meerschaum Compose.

• Allow for type annotations in required. Plugins may now annotate required:

# plugins/example.py

__version__: str = '0.0.1'
required: list[str] = ['requests']

• Automatically include --noask and --yes in remote actions. For your convenience, the flags --noask and --yes are included in remote actions sent by APIConnector.do_action().

• Fixed an issue with URIs for api connectors. Creating an APIConnector via a URI connection string now properly handles the protocol.

• Fixed a formatting issue with show logs.

v2.2.5¶

• Add bootstrap plugin. The bootstrap plugin wizard provides a convenient way to create new plugins from templates.

• Add edit plugin. The action edit plugin will open a plugin's source file in your editor ($EDITOR or pyvim).

• Allow actions, fetch(), and sync() to omit pipe and **kwargs. Adding **kwargs (and pipe) is now optional, and you may instead explicitly state only the arguments required.

from meerschaum.actions import make_action

@make_action
def do_the_thing(shell=False):
    msg = (
        "You're in the mrsm shell."
        if shell
        else "You called this from bash."
    )
    return True, msg


# You may now omit the positional argument `pipe`.
def fetch(begin=None, end=None):
    return [{'ts': '2024-01-01'}]

• Fixed minor bug with subaction detection. Subaction functions must now explicitly begin with the name of the parent action (underscore prefix allowed).

from meerschaum.actions import make_action

@make_action
def sing():
    return True, "~la la la~"

def sing_song():
    return True, "~do re mi~"

def _sing_tune():
    return True, "~fo so la ti~"

• Allow for --begin None. Explicitly setting --begin None will now pass None to fetch().

• Persist user packages in stack Docker container. The stack Docker Compose file now persists user-level packages (under ~/.local).

• Throw a warning if a @dash_plugin function raises an exception.

• Added connector type to show connectors. Append a connector type to the show connectors command (e.g. show connectors sql) to see only connectors of a certain type.

• Allow dprint to be imported from meerschaum.utils.warnings. For convenience, you may now import dprint alongside info, warn, and error.

from meerschaum.utils.warnings import dprint, warn, info, error

• Add positional arguments filtering (filter_positional() and filter_arguments()) In addition to keyword argument filtering, you may now filter positional arguments.

from meerschaum.utils.misc import filter_arguments, filter_positional

def foo(a, b, c=0):
    return (a * b) + c

filter_positional(foo, 1, 2, 3)
# (1, 2)

filter_arguments(foo, 1, 2, c=3, d=4)
# ((1, 2), {'c': 3})

• Cleaned up OAuth flow (/login).

v2.2.2 – v2.2.4¶

• Speed up package installation in virtual environments. Dynamic dependencies will now be installed via uv, which dramatically speeds up installation times.

• Add sub-cards for children pipes. Pipes with children defined now include cards for these pipes under the Parameters menu item. This is especially useful when working managing pipeline hierarchies.

• Add "Open in Python" to pipe cards. Clicking "Open in Python" on a pipe's card will now launch ptpython with the pipe object already created.

# Clicking "Open in Python" executes the following:
# $ mrsm python "pipe = mrsm.Pipe('plugin:noaa', 'weather', 'gvl', instance='sql:main')"
>>> import meerschaum as mrsm
>>> pipe = mrsm.Pipe('plugin:noaa', 'weather', 'gvl', instance='sql:main')

• Add the decorators @web_page and @dash_plugin. You may now quickly add your own pages to the web console by decorating your layout functions with @web_page:

# example.py
from meerschaum.plugins import dash_plugin, web_page

@dash_plugin
def init_dash(dash_app):

    import dash.html as html
    import dash_bootstrap_components as dbc
    from dash import Input, Output, no_update

    @web_page('/my-page', login_required=False)
    def my_page():
        return dbc.Container([
            html.H1("Hello, World!"),
            dbc.Button("Click me", id='my-button'),
            html.Div(id="my-output-div"),
        ])

    @dash_app.callback(
        Output('my-output-div', 'children'),
        Input('my-button', 'n_clicks'),
    )
    def my_button_click(n_clicks):
        if not n_clicks:
            return no_update
        return html.P(f'You clicked {n_clicks} times!')

• Use ptpython for the python action. Rather than opening a classic REPL, the python action will now open a ptpython shell.

• Allow passing flags to venv ptpython binaries. You may now pass flags directly to the ptpython binary of a virtual environment (by escaping with []):

mrsm python [--help]

• Allow for custom connectors to implement a sync() method. Like module-level sync() functions for plugin connectors, any custom connector may implement sync() instead of fetch().

# example.py
from typing import Any
import meerschaum as mrsm
from meerschaum.connectors import Connector, make_connector

@make_connector
class ExampleConnector(Connector):

    def register(self, pipe: mrsm.Pipe) -> dict[str, Any]:
        return {
            'columns': {
                'datetime': 'ts',
                'id': 'example_id',
            },
        }

    def sync(self, pipe: mrsm.Pipe, **kwargs) -> mrsm.SuccessTuple:
        ### Implement a custom sync.
        return True, f"Successfully synced {pipe}!"

v2.2.1¶

• Fix --schedule in the interactive shell. The --schedule flag may now be used from both the CLI and the Shell.

• Fix the SQLConnector CLI. The sql action now correctly opens an interactive CLI.

• Bumped duckdb to >=1.0.0. The upstream breaking changes that required duckdb to be held back have to do with how indices behave. For now, index creation has been disabled so that duckdb may be upgraded to 1.0+.

v2.2.0¶

New Features

• New job scheduler The job scheduler has been rewritten with a simpler syntax.

mrsm sync pipes -s 'daily & mon-fri starting 00:00 tomorrow' -d

• Add show schedule. Validate your schedules' upcoming timestamps with show schedule.

mrsm show schedule 'daily & mon-fri starting 2024-05-01'

Next 5 timestamps for schedule 'daily & mon-fri starting 2024-05-01':

  2024-05-01 00:00:00+00:00
  2024-05-02 00:00:00+00:00
  2024-05-03 00:00:00+00:00
  2024-05-06 00:00:00+00:00
  2024-05-07 00:00:00+00:00

• Added timestamps to log file lines. Log files now prepend the current minute to each line of the file, and the timestamps are also printed when viewing logs with show logs. To disable this behavio, set MRSM{jobs:logs:timestamps:enabled} to false.

You may change the timestamp format under the config keys MRSM{jobs:logs:timestamps:format} (timestamp written to disk) and MRSM{jobs:logs:timestamps:follow_format} (timestamp printed when following via show logs.).

• Add --skip-deps. When installing plugins, you may skip dependencies with --skip-deps. This should improve the iteration loop during development.

mrsm install plugin noaa --no-deps

• Add logs buttons to job cards on the Web UI. For your convenience, "Follow logs" and "Download logs" buttons have been added to jobs' cards.

• Add a Delete button to job cards on the Web UI. You may now delete a job from its card (once stopped, that is).

• Add management buttons to pipes' cards. For your convenience, you may now sync, verify, clear, drop, and delete pipes directly from cards.

• Designate your packages as plugins with the meerschaum.plugins entry point. You may now specify your existing packages as Meerschaum plugins by adding the meerschaum.plugins Entrypoint to your package metadata:

from setuptools import setup

setup(
    ...,
    entry_points = {
        'meerschaum.plugins': [
            'foo = foo',
        ],
    },
)

or if you are using pyproject.toml:

[project.entry-points."meerschaum.plugins"]
foo = "foo"

• Pre- and post-sync hooks are printed separately. The results of sync hooks are now printed right after execution rather than after the sync.

Bugfixes

• Fixed a filtering bug on the Web UI when changing instances. When changing instances on the Web Console, the connector, metric, and location choices will reset appropriately.

• Ctrl+C when exiting show logs. Pressing Ctrl+C will now exit the show logs immediately.

Breaking Changes

• No longer supporting the old scheduler syntax. If you have jobs with the old scheduler syntax (e.g. using the keyword before), you may need to delete and recreate your jobs with an updated schedule.

• Upgraded to psycopg from psycopg2. The upgrade to psycopg (version 3) should provide better performance for larger transactions.

• Daemon.cleanup() now returns a SuccessTuple.

Other changes

• Bumped xterm.js to v5.5.0.
• Added tags to the pipes card.
• Replaced watchgod with watchfiles.
• Replaced rocketry with APScheduler.
• Removed pydantic from dependencies.
• Removed passlib from dependencies.
• Bumped default TimescaleDB image to latest-pg16-oss.
• Held back duckdb to <0.10.3.

The 2.1.x series added high-performance upserts, improved numerics support and temporary tables performance, and many other bugfixes and improvements.

v2.1.7¶

• Add query_df() to meerschaum.utils.dataframe. The function query_df() allows you to filter dataframes by params, begin, and end.

import pandas as pd
df = pd.DataFrame([
    {'a': 1, 'color': 'red'},
    {'a': 2, 'color': 'blue'},
    {'a': 3, 'color': 'green'},
    {'a': 4, 'color': 'yellow'},
])

from meerschaum.utils.dataframe import query_df
query_df(df, {'color': ['red', 'yellow']})
#    a   color
# 0  1     red
# 3  4  yellow
query_df(df, {'color': '_blue'}, reset_index=True)
#    a   color
# 0  1     red
# 1  3   green
# 2  4  yellow
query_df(df, {'a': 2}, select_columns=['color'])
#   color
# 1  blue

from meerschaum.utils.misc import get_in_ex_params
params = {'color': ['red', '_blue', 'green']}
in_ex_params = get_in_ex_params(params)
in_ex_params
# {'color': (['red', 'green'], ['blue'])}
in_vals, ex_vals = in_ex_params['color']

• Add coerce_numeric to pipe.enforce_dtypes(). Setting this to False will not cast floats to Decimal if the corresponding dtype is int.

• Improve JSON serialization when filtering for updates.

• Add date_bound_only to pipe.filter_existing(). The argument date_bound_only means that samples retrieved by pipe.get_data() will only use begin and end for bounding. This may improve performance for custom instance connectors which have limited searchability.

• Add safe_copy to pipe.enforce_types(), pipe.filter_existing(), filter_unseen_df(). By default, these functions will create copies of dataframes to avoid mutating the input dataframes. Setting safe_copy to False may be more memory efficient.

• Add multiline support to extract_stats_from_message. Multiple messages separated by newlines may be parsed at once.

from meerschaum.utils.formatting import extract_stats_from_message
extract_stats_from_message("Inserted 10, upserted 3\ninserted 11, upserted 4")
# {'inserted': 21, 'updated': 0, 'upserted': 7}

• Remove order by check in SQL queries.

• Improve shell startup performance by removing support for cmd2. The package cmd2 never behaved properly, so support has been removed and only the built-in cmd powers the shell. As such, the configuration key shell:cmd has been removed.

v2.1.6¶

• Move success_tuple from arg to kwarg for @post_sync_hook functions. To match the signature of @pre_sync_hook functions, @post_sync_hook functions now only accept pipe as the positional argument. The return value of the sync will now be passed as the kwarg success_tuple. This allows you to use the same callback function as both the pre- and post-sync hooks.

import meerschaum as mrsm
from meerschaum.plugins import pre_sync_hook, post_sync_hook

@pre_sync_hook
@post_sync_hook
def log_sync(
        pipe: mrsm.Pipe,
        success_tuple: mrsm.SuccessTuple | None = None,
    ):
    if success_tuple is None:
        print(f"About to sync {pipe}!")
    else:
        success, msg = success_tuple
        print(f"Finished syncing {pipe} with message:\n{msg}")

• Add sync_timestamp and sync_complete_timestamp to sync hooks. The UTC datetime right before the sync is added to the sync hook kwargs, allowing for linking the two callbacks to the same datetime. For convenience, the UTC datetime is also captured at the end of the sync and is passed as sync_complete_timestamp.

from datetime import datetime
import meerschaum as mrsm
from meerschaum.plugins import pre_sync_hook, post_sync_hook

@pre_sync_hook
@post_sync_hook
def log_sync(
        pipe: mrsm.Pipe,
        sync_timestamp: datetime | None = None,
        sync_complete_timestamp: datetime | None = None,
        sync_duration: float | None = None,
    ) -> mrsm.SuccessTuple:

    if sync_complete_timestamp is None:
        print(f"About to sync {pipe} at {sync_timestamp}.")
        return True, "Success"

    msg = (
        f"It took {sync_duration} seconds to sync {pipe}:\n"
        + "({sync_timestamp} - {sync_complete_timestamp})"
    )
    return True, msg

• Improved performance of sync hooks. Sync hooks are now called asynchronously in their own threads to avoid slowing down or crashing the main thread.

• Rename duration to sync_duration for sync hooks. To avoid potential conflicts, the kwarg duration is prefixed with sync_ to denote that it was specifically added to provide context on the sync.

• Allow for sync hooks to return SuccessTuple. If a sync hook returns a SuccessTuple (Tuple[bool, str]), the result will be printed.

import meerschaum as mrsm
from meerschaum.plugins import post_sync_hook

@post_sync_hook
def log_sync(pipe) -> mrsm.SuccessTuple:
    return True, f"Logged sync for {pipe}."

• Add is_success_tuple() to meerschaum.utils.typing. You can now quickly check whether an object is a SuccessTuple:

from meerschaum.utils.typing import is_success_tuple
assert is_success_tuple((True, "Success"))
assert not is_success_tuple(("foo", "bar"))

• Allow for index-only pipes when upsert=True. If all columns are indices and upsert is True, then the upsert will insert net-new rows (ignore duplicates).

import meerschaum as mrsm
pipe = mrsm.Pipe(
    'a', 'b',
    columns = ['a', 'b'],
    parameters = {'upsert': True},
    instance = 'sql:local',
)
pipe.sync([{'a': 1, 'b': 2}])

• Allow for prelimary null index support for upsert=True (inserts only, PostgreSQL only). Like with regular syncs, upsert syncs now coalesce indices to allow for syncing null values. NOTE: the transaction will fail if a null index is synced again, so this is only for the initial insert.

• Remove automatic instance table renaming. This patch removes automatic detection and renaming of old instance tables to the new names (e.g. users -> mrsm_users). Users migrating from an old installation will need to rename the tables manually themselves.

v2.1.5¶

• Add the action tag pipes. Tags may be added or removed with the tag pipes action. Note that the flag --tags applies to existing tags for filtering; flags to be added or removed are positional arguments.

### Tag all `sql:main` pipes with `production` and `sync-daily`.
mrsm tag pipes production sync-daily -c sql:main

### Remove the `production`` tag from all pipes tagged as `production`.
mrsm tag pipes _production --tags production

• Add --tags support to register pipes. The action register pipe with the --tags flag will auto-tag the new pipes.

• Clean up warnings on Python 3.12. All instances of datetime.utcnow() have been replaced by datetime.now(timezone.utc).replace(tzinfo=None) (to preserve behavior). A full migration to timezone-aware datetimes would have to happen in a minor release.

• Improve timezone-aware datetime support for MSSQL. Passing a timezone-aware datetime as a date bound for MSSQL should now be fixed.

• Add an explicit VOLUME to the Dockerfile. The path /meerschaum is now explicitly set as a VOLUME in the Docker image.

• Add --tags filtering to the show tags action.

• Improve global ThreadPool handling. Global pools are now created on a per-worker (and per-class) basis, allowing for switching between workers within the same process. Note that global pools are maintained to allow for nested chunking and the limit the number of connections (e.g. avoid threads creating their own pools).

• Fix a bug when selecting only by negating tags. Pipes may now be selected by only specifying negated tags.

• Rename meerschaum.utils.get_pipes to meerschaum.utils._get_pipes to avoid namespace collissions. The functions get_pipes() and fetch_pipes_keys() are available at the meerschaum.utils module namespace.

v2.1.3 – v2.1.4¶

• Add the decorators @pre_sync_hook and @post_sync_hook. The new decorators @pre_sync_hook and @post_sync_hook let you intercept a Pipe immediately before and after a sync, capturing its return tuple and the duration in seconds.

import meerschaum as mrsm
from meerschaum.plugins import pre_sync_hook, post_sync_hook

@pre_sync_hook
def prepare_for_sync(pipe: mrsm.Pipe, **kwargs):
    print(f"About to sync {pipe} with kwargs:\n{kwargs}")

@post_sync_hook
def log_sync(
        pipe: mrsm.Pipe,
        return_tuple: mrsm.SuccessTuple,
        duration: float = 0.0
    ):
    print(f"It took {round(duration, 2)} seconds to sync {pipe} with result:")
    mrsm.pprint(return_tuple)

• Add the action show tags. The action show tags will now display panels of pipes grouped together by common tags. This is useful for large deployments which share common tags.

• Add dropdowns and inputs for flags with arguments to the Web Console. Leverage the full power of the Meerschaum CLI in the Web Console with the new dynamic flags dropdowns.

• Fix shell crashes in Docker containers. Reloading the running Meerschaum session from an interactive shell via a Docker container will no longer cause crashes on custom commands.

• Improve reloading times. Reloading the running Meerschaum session has been sped up by several seconds (due to skipping the internal shell modules).

• Improve virtual environments in the Docker image. Initial startup of Docker containers on a fresh persistent volume has been sped up due to preloading the default virtual environment creation. Additionally, the environment variable $MRSM_VENVS_DIR has been unset, reverting the virtual environments to be stored under /meerschaum/venvs.

v2.1.1 – v2.1.2¶

• Add upsert for high-performance pipes. Setting upsert under pipe.parameters will create a unique index and combine the insert and update stages into a single upsert. This is particularly useful for pipes with very large tables.

import meerschaum as mrsm

pipe = mrsm.Pipe(
    'demo', 'upsert',
    instance = 'sql:local',
    columns = ['datetime', 'id'],
    parameters = {'upsert': True},
)

mrsm.pprint(
    pipe.sync(
        [
            {'datetime': '2023-01-01', 'id': 1, 'val': 1.1},
            {'datetime': '2023-01-02', 'id': 2, 'val': 2.2},
        ]
    )
)
#  🎉 Upserted 2 rows.

• Add internal schema _mrsm_internal for temporary tables. To avoid polluting your database schema, temporary tables will now be created within _mrsm_internal (for databases which support PostgreSQL-style schemas).

• Added mrsm_temporary_tables. Temporary tables are logged to mrsm_temporary_tables and deleted when dropped.

• Drop stale temporary tables. Temporary tables which are more than 24 hours will be dropped automatically (configurable under system:sql:instance:stale_temporary_tables_minutes).

• Prefix instance tables with mrsm_. Internal Meerschaum tables still reside on the default schema but will be denoted with a mrsm_ prefix. The existing instance tables will be automatically renamed (e.g. pipes will become mrsm_pipes) to add the prefix (this renaming detection will be removed in later releases).

• Fix an issue with bootstrap. Refactoring work for 2.1.0 had broken the bootstrap action.

• Fix an issue with pause jobs.

• Fix an issue when selecting inverse pipes. Null location keys are now coalesced when selecting pipes to produce expected behavior.

• Avoid a system exit when exiting the SQL CLI.

v2.1.0¶

• Replace term.js with xterm.js. This has been a long time coming. The webterm has been migrated to xterm.js which has continuous support from term.js which was last updated almost 10 years ago.

• Deprecate the legacy web pseudo-terminal. Clicking the "Execute" button on the web console will now execute the command directly in the webterm. Additionally, changing the instance select will now automatically switch the webterm's context to the desired instance.

• Fix an issue when starting existing jobs. A bug has been fixed which prevented jobs from restarting specifically by name.

• Add MRSM_VENVS_DIR. Like MRSM_PLUGINS_DIR, you can now designate a virtual environments directory separate from the root directory. This is particularly useful for production deployments, and MRSM_VENVS_DIR has been set to /home/meerschaum/venvs in the official Docker images to allow for mounting /meerschaum to persistent volumes.

• Allow syncing NULL values into indices. Syncing None within an index will now be coalesced into a magic value when applying updates.

import meerschaum as mrsm

pipe = mrsm.Pipe(
    'allow', 'null', 'indices',
    instance = 'sql:local',
    columns = ['a', 'b'],
)
pipe.sync([{'a': 1, 'b': 1}])
pipe.sync([{'b': 1}])
pipe.sync([{'a': 1}])
pipe.sync([{'c': 1}])

print(pipe.get_data())
#       a     b     c
# 0  <NA>  <NA>     1
# 1  <NA>     1  <NA>
# 2     1  <NA>  <NA>
# 3     1     1  <NA>

• Syncing Decimal objects will now enforce numeric dtypes. For example, syncing a Decimal onto a integer column will update the dtype to numeric, like when syncing a float after an integer.

import meerschaum as mrsm
from decimal import Decimal

pipe = mrsm.Pipe(
    'demo', 'decimal', 'coersion',
    instance = 'sql:local',
    columns = ['id'],
)
pipe.sync([{'id': 1, 'foo': 10}])
pipe.sync([{'id': 1, 'foo': Decimal('20')}])
print(pipe.dtypes)
# {'id': 'int64[pyarrow]', 'foo': 'numeric'}

df = pipe.get_data()
print(f"{df['foo'][0]=}")
# df['foo'][0]=Decimal('20')

• Improve IS NULL and IS NOT NULL checks for params. Mixing null-like values (e.g. NaN, <NA>, None) in params will now separate out nulls.

from meerschaum.utils.sql import build_where
print(build_where({'a': ['_<NA>', '_1', '_2']}))
# WHERE
#   ("a" NOT IN ('1', '2')
#   AND "a" IS NOT NULL)
print(build_where({'a': ['NaN', '1', '2']}))
# WHERE
#   ("a" IN ('1', '2')
#   OR "a" IS NULL)

• Add colors to mrsm show columns.

• Fix a unicode decoding error when showing logs.

• Remove xstatic dependencies. The xterm.js files are now bundled as static assets, so the term.js files are no longer needed. Hurray for removing dependencies!

• Other bugfixes. A handful of minor bugfixes have been included in this release:

  • Removed non-connector environment variables like MRSM_WORK_DIR from the mrsm show connectors output.
  • Improving symlinks handling for multi-processed situations (mrsm start api).

At long last, 2.0 has arrived! The 2.0 releases brought incredible change, from standardizing chunking to adding Pipe.verify() and Pipe.deduplicate() to introducing first-class numeric support. See the full release notes below for the complete picture.

v2.0.8 – v2.0.9¶

• Cast None to Decimal('NaN') for numeric columns. To allow for all-null numeric columns, None (and other null-like types) are coerced to Decimal('NaN').

• Schema bugfixes. A few minor edge cases have been addressed when working with custom schemas for pipes.

• Remove APIConnector.get_backtrack_data(). Since 1.7 released, the get_backtrack_data() method for instance connectors has been optional.

  NOTE: the backtrack_data API endpoint has also been removed.

• Other bugfixes. Issues with changes made to session authentication have been addressed.

v2.0.5 – v2.0.7¶

• Add the numeric dtype (i.e. support for NUMERIC columns). Specifying a column as numeric will coerce it into decimal.Decimal objects. For SQLConnectors, this will be stored as a NUMERIC column. This is useful for syncing a mix of integer and float values.

import meerschaum as mrsm
pipe = mrsm.Pipe(
    'demo', 'numeric',
    instance = 'sql:main',
    columns = ['foo'],
    dtypes = {'foo': 'numeric'},
)
pipe.sync([{'foo': '1'}, {'foo': '2.01234567890123456789'}])
df = pipe.get_data()
print(df.to_dict(orient='records'))
# [{'foo': Decimal('1')}, {'foo': Decimal('2.01234567890123456789')}]

NOTE: Due to implementation limits, numeric has strict precision issues in embedded databases (SQLite and DuckDB: NUMERIC(15, 4)). PostgreSQL-like database flavors have the best support for NUMERIC; MySQL and MariaDB use a scale and precision of NUMERIC(38, 20), MSSQL uses NUMERIC(28, 10), and Oracle and PostgreSQL are not capped.

• Mixing int and float will cast to numeric. Rather than always casting to TEXT, a column containing a mix of int and float will be coerced into numeric.

import meerschaum as mrsm
pipe = mrsm.Pipe('mix', 'int', 'float', instance='sql:local')
pipe.sync([{'a': 1}])
pipe.sync([{'a': 1.1}])
print(pipe.dtypes)
# {'a': 'numeric'}

• Add schema to SQLConnectors. Including the key schema or as an argument in the URI will use this schema for created tables. The argument search_path will also set schema (i.e. for PostgreSQL).

export MRSM_SQL_FOO='{
  "username": "foo",
  "password": "bar",
  "port": 5432,
  "flavor": "timescaledb",
  "host": "localhost",
  "database": "db",
  "schema": "myschema"
}'

export MRSM_SQL_FOO='timescaledb://foo:bar@localhost:5432/db?options=--search_path%3Dmyschema'

• Add schema to pipe.parameters. In addition to the default schema at the connector level, you may override this by setting schema under pipe.parameters.

import meerschaum as mrsm
pipe = mrsm.Pipe('a', 'b', parameters={'schema': 'myschema'})

• Add schema to meerschaum.utils.sql.sql_item_name(). You may now pass an optional schema when quoting:

from meerschaum.utils.sql import sql_item_name
print(sql_item_name('foo', 'mssql', schema='dbo'))
# '[dbo].[foo]'

• Add options to SQLConnector. The key options will now contain a sub-dictionary of connection options, such as driver, search_path, or any other query parameters.

• Disable the "Sync Documents" accordion item when the session is not authenticated. When running the API with --secure, only admin users will be able to access the "Sync Documents" accordion items on the pipes' cards.

• Remove dtype_backend from SQLConnector.read(). This argument previously had no effect. When applied, it was coercing JSON columns into strings, so it was removed.

• Remove meerschaum.utils.daemon.Log. This had been replaced by meerschaum.utils.daemon.RotatingLog and had been broken since the 2.0 release.

• Remove params from Pipe.filter_existing(). To avoid confusion, filter parameters are instead derived from the incoming DataFrame. This will improve performance when repeatedly syncing chunks which span the same interval. The default limit of 250 unique values may be configured under pipes:sync:filter_params_index_limit.

• Add forwarded_allow_ips and proxy_headers to the web API. The default values forwarded_allow_ips='*' and proxy_headers=True are set when running Uvicorn or Gunicorn and will help when running Meerschaum behind a proxy.

• Bump dash-extensions to >=1.0.4. The bug that was holding back the version was due to including enrich.ServersideTransform in the dash proxy without actually utilizing it.

v2.0.3 – v2.0.4¶

• Fix an issue with --timeout-seconds. Previous refactoring efforts had broken the --timeout-seconds polling behavior.

• Fix a formatting issue when pretty-printing pipes. Pipes may now be correctly printed if both single and double quotes appear in a message.

• Allow omitting port for APIConnectors. You may now omit the port attribute for APIConnectors to use the protocol-default port (e.g. 443 for HTTPS). Note you will need to delete the key api:default:port via mrsm edit config if it's present.

• Add optional verify key to API connectors. Client API connectors may now be used with self-signed HTTPS instances.

• Bump duckdb to version 0.9.0. This adds complete support for PyArrow data types to DuckDB.

v2.0.2¶

• Syncing with --skip-check-existing will not apply the backtrack interval. Because --skip-check-existing (or check_existing=False) is guaranteed to produce duplicates, the backtrack interval will be set to 0 when running in insert-only mode.

• Allow for columns to be a list. Note that building a pipe with columns as a list must have the datetime column named datetime.

import meerschaum as mrsm
pipe = mrsm.Pipe('a', 'b', columns=['a'])
print(pipe.columns)
# {'a': 'a'}

• Bump default SQLAlchemy pool size to 8 connections.

• Consider the number of checked out connections when choosing workers. For pipes on sql instances, pipe.get_num_workers() will now consider the number of checked out connections rather than only the number of active threads.

• Fix pipe.get_data(as_dask=True) for JSON columns.

v2.0.1¶

• Fix syncing bools within in-place SQL pipes. SQL pipes may now sync bools in-place. For database flavors which lack native BOOLEAN support (e.g. sqlite, oracle, mysql), then the boolean columns must be stated in pipe.dtypes.

• Fix an issue with multiple users managing jobs. Extra validation was added to the web UI to allow for multiple users to interact with jobs.

• Fix a minor formatting bug with highlight_pipes(). Improved validation logic was added to prevent incorrectly prepending the Pipe( prefix.

• Hold back pydantic to <2.0.0 Pydantic 2 is supported in all features except --schedule. Until rocketry supports Pydantic 2, it will be held back.

v2.0.0¶

Breaking Changes

• Removed redundant Pipe.sync_time property. Use pipe.get_sync_time() instead.

• Removed SQLConnector.get_pipe_backtrack_minutes(). Use pipe.get_backtrack_interval() instead.

• Replaced pipe.parameters['chunk_time_interval'] with pipe.parameters['verify']['chunk_minutes'] For better security and cohesiveness, the TimescaleDB chunk_time_interval value is now derived from the standard chunk_minutes value. This also means pipes with integer date axes will be created with a new default chunk interval of 1440 (was previously 100,000).

• Moved choose_subaction() into meerschaum.actions. This function is for internal use and as such should not affect any users.

Features

• Added verify pipes and --verify. The command mrsm verify pipes or mrsm sync pipes --verify will resync pipes' chunks with different rowcounts to catch any backfilled data.

import meerschaum as mrsm
foo = mrsm.Pipe(
    'foo', 'src',
    target = 'foo',
    columns = {'datetime': 'dt'},
    instance = 'sql:local'
)
docs = [
    {'dt': '2023-01-01'},
    {'dt': '2023-01-02'},
]
foo.sync(docs)

pipe = mrsm.Pipe(
    'sql:local', 'verify',
    columns = {'datetime': 'dt'},
    parameters = {
        'query': f'SELECT * FROM "{foo.target}"'
    },
    instance = 'sql:local',
)
pipe.sync(docs)

backfilled_docs = [
    {'dt': '2022-12-30'},
    {'dt': '2022-12-31'},
]
foo.sync(backfilled_docs)
mrsm.pprint(pipe.verify())
assert foo.get_rowcount() == pipe.get_rowcount()

• Added deduplicate pipes and --deduplicate. Running mrsm deduplicates pipes or mrsm sync pipes --deduplicate will iterate over pipes' entire intervals, chunking at the configured chunk interval (see pipe.get_chunk_interval() below) and clearing + resyncing chunks with duplicate rows.

If your instance connector implements deduplicate_pipe() (e.g. SQLConnector), then this method will override the default pipe.deduplicate().

pipe = mrsm.Pipe(
    'demo', 'deduplicate',
    columns = {'datetime': 'dt'},
    instance = 'sql:local',
)
docs = [
    {'dt': '2023-01-01'},
    {'dt': '2023-01-01'},
]
pipe.sync(docs)
print(pipe.get_rowcount())
# 2
pipe.deduplicate()
print(pipe.get_rowcount())
# 1

• Added pyarrow support. The dtypes enforcement system was overhauled to add support for pyarrow data types.

import meerschaum as mrsm
import pandas as pd

df = pd.DataFrame(
    [{'a': 1, 'b': 2.3}]
).convert_dtypes(dtype_backend='pyarrow')

pipe = mrsm.Pipe(
    'demo', 'pyarrow',
    instance = 'sql:local',
    columns = {'a': 'a'},
)
pipe.sync(df)

• Added bool support. Pipes may now sync DataFrames with booleans (even on Oracle and MySQL):

import meerschaum as mrsm
pipe = mrsm.Pipe(
    'demo', 'bools',
    instance = 'sql:local',
    columns = {'id': 'id'},
)
pipe.sync([{'id': 1, 'is_blue': True}])
assert 'bool' in pipe.dtypes['is_blue']

pipe.sync([{'id': 1, 'is_blue': False}])
assert pipe.get_data()['is_blue'][0] == False

• Added preliminary dask support. For example, you may now return Dask DataFrames in your plugins, pass into pipe.sync(), and pipe.get_data() now has the flag as_dask.

import meerschaum as mrsm
pipe = mrsm.Pipe(
    'dask', 'demo',
    columns = {'datetime': 'dt'},
    instance = 'sql:local',
)
pipe.sync([
    {'dt': '2023-01-01', 'val': 1},
    {'dt': '2023-01-02', 'val': 2},
    {'dt': '2023-01-03', 'val': 3},
])
ddf = pipe.get_data(as_dask=True)
print(ddf)
# Dask DataFrame Structure:
#                            dt             val
# npartitions=4
#                datetime64[ns]  int64[pyarrow]
#                           ...             ...
#                           ...             ...
#                           ...             ...
#                           ...             ...
# Dask Name: from-delayed, 5 graph layers

print(ddf.compute())
#           dt  val
# 0 2023-01-01    1
# 0 2023-01-02    2
# 0 2023-01-03    3

pipe2 = mrsm.Pipe(
    'dask', 'insert',
    columns = pipe.columns,
    instance = 'sql:local',
)
pipe2.sync(ddf)
assert pipe.get_data().to_dict() == pipe2.get_data().to_dict()

pipe.sync([{'dt': '2023-01-01', 'val': 10}])
pipe2.sync(pipe.get_data(as_dask=True))
assert pipe.get_data().to_dict() == pipe2.get_data().to_dict()

• Added chunk_minutes to pipe.parameters['verify']. Like pipe.parameters['fetch']['backtrack_minutes'], you may now specify the default chunk interval to use for verification syncs and iterating over the datetime axis.

import meerschaum as mrsm
pipe = mrsm.Pipe(
    'a', 'b',
    instance = 'sql:local',
    columns = {'datetime': 'dt'},
    parameters = {
        'verify': {
            'chunk_minutes': 2880,
        }
    },
)
pipe.sync([
    {'dt': '2023-01-01'},
    {'dt': '2023-01-02'},
    {'dt': '2023-01-03'},
    {'dt': '2023-01-04'},
    {'dt': '2023-01-05'},
])
chunk_bounds = pipe.get_chunk_bounds(bounded=True)
for chunk_begin, chunk_end in chunk_bounds:
    print(chunk_begin, '-', chunk_end)

# 2023-01-01 00:00:00 - 2023-01-03 00:00:00
# 2023-01-03 00:00:00 - 2023-01-05 00:00:00