sluicebox

High-throughput, validated and observable writes and queries for InfluxDB 2 and InfluxDB 3, built to drop into an application and push large volumes of data as fast as the server accepts them.

• Asynchronous writes: write() validates and serializes on the calling thread, buffers, and returns at once; background threads batch, gzip and send with up to 16 requests in flight, retrying transient failures. Await the returned future when you need the acknowledgement.
• Fast: about 1.15 M points/s serialization per thread with full validation, 3x the official clients; polars DataFrames are serialized vectorized. End to end, 5-50x the official clients (below).
• Validation that prevents bad data: field types are locked on first write and corrected from the server's own type errors (no more field type conflict batches), schemas can be declared, identifiers are checked against what each server version really accepts, NaN/inf, naive datetimes, out-of-range values and timestamps in the wrong unit are caught before sending.
• Content-aware tag injection: static tags, environment tags, request-scoped context tags, and rules that derive tags from a point's measurement, tags and field values (with regex captures).
• Configuration in TOML, secrets only from the environment / .env / secret files (pydantic SecretStr).
• Prometheus metrics, structured logging, per-stage profiling and a cProfile helper.
• Typed exceptions for every failure, including per-line attribution of partial writes.
• Sync and asyncio APIs; thread-safe, and on free-threaded Python 3.14t writes from several threads use several cores; fork-aware.

Queries go through the official clients: Arrow Flight (SQL / InfluxQL) for InfluxDB 3 via influxdb3-python, Flux for InfluxDB 2 via influxdb-client. The write path uses its own urllib3 transport, so ingest-only applications need neither.

Contents

• Install
• Quick start
• Writing
• Choosing tags
• Configuration and secrets
• Validation and schemas
• Tag injection
• Queries
• Errors
• Observability
• Running in production
• Performance and tuning
• Remote servers, TLS and proxies
• Server behaviour handled for you
• Development

Install

uv add sluicebox                    # writes only
uv add 'sluicebox[v3]'              # + InfluxDB 3 queries
uv add 'sluicebox[v2]'              # + InfluxDB 2 queries
uv add 'sluicebox[polars]'          # + polars DataFrames
uv add 'sluicebox[all]'             # all of the above, plus pandas

Python 3.11+; pip install takes the same specifiers. The development version installs from GitHub: uv add 'sluicebox[v3] @ git+https://github.com/davidson-engineering/sluicebox'.

Quick start

sluicebox.toml (non-secret settings; full reference in sluicebox.example.toml):

[connection]
url = "http://localhost:8181"
version = 3              # or 2 (then also: org = "my-org")
database = "telemetry"   # bucket on InfluxDB 2

.env (or real environment variables; never commit it):

SLUICEBOX_TOKEN=apiv3_...

from sluicebox import InfluxClient, Point

with InfluxClient.from_config() as client:  # reads ./sluicebox.toml and ./.env
    client.check()  # fail fast: wrong URL, version, token or bucket

    # Asynchronous: returns immediately, sent in the background.
    client.write({"measurement": "cpu", "tags": {"host": "web-1"}, "fields": {"usage": 0.42}})
    client.write(Point("cpu").tag("host", "web-2").field("usage", 0.17))

    # Synchronous: wait for the server's acknowledgement (raises on failure).
    batch = [
        {"measurement": "cpu", "tags": {"host": f"web-{i}"}, "fields": {"usage": i / 10}} for i in range(10)
    ]
    result = client.write(batch).result()

    for row in client.query("SELECT * FROM cpu WHERE time > now() - interval '1 hour'"):
        print(row)  # or .to_polars(), .to_pandas(), .to_arrow() with those extras
# Leaving the block flushes everything buffered.

InfluxDB 3 Core acknowledges a write when its write-ahead log is persisted (every second by default), so a single synchronous write takes up to a second. Throughput comes from many writes in flight, not from waiting on each; write.no_sync = true acknowledges in about a millisecond with weaker durability.

client.to_line_protocol(data) returns the lines write(data) would send, with validation and tag injection applied, without sending them.

Writing

Inputs

write() accepts any of these, or an iterable mixing them:

Timestamps may be ints in the write precision, timezone-aware datetimes, ISO 8601 strings (nanoseconds kept), float epoch seconds, or numpy/pandas timestamps and integers. Values InfluxDB cannot store (before 1677 or after 2262) are rejected; integer timestamps that land before 1973 log a warning, since that is what epoch seconds written with precision="ns" look like. Points without a timestamp (raw line protocol too) get the time of the write() call (write.auto_timestamp), which makes retries idempotent. Untimed points of the same series written in one call therefore share a timestamp and overwrite each other, as with server-assigned times; sluicebox logs a warning when that happens (except for raw lines passed through unparsed).

DataFrames: a column named time (or a datetime column named timestamp, or a pandas DatetimeIndex) is the timestamp unless time_column= says otherwise; it may hold datetimes, integers in the write precision, or ISO 8601 strings. All other columns that are not tag_columns are fields.

Typed models declare their field types once, so they can never write an inconsistent type:

from dataclasses import dataclass
from datetime import datetime
from typing import Annotated
from sluicebox import Tag, measurement


@measurement("cpu")
@dataclass
class CpuSample:
    host: Annotated[str, Tag]
    usage: float  # declared float: an int value is written as 1.0
    cores: int
    time: datetime  # an attribute named "time"/"timestamp" is the timestamp


client.write([CpuSample("web-1", 1, 8, now)])

Asynchronous and synchronous use

write() returns a WriteFuture as soon as the data is buffered:

future = client.write(points)  # fire and forget: ignore the future
result = future.result(timeout=30)  # or wait: WriteResult(points=..., dropped=..., duration=...)
await future  # the same from asyncio
future.add_done_callback(lambda f: ack_upstream() if f.exception() is None else nack())  # on a sender thread

Waiting flushes the buffer, so a synchronous write returns as soon as the server answers rather than after flush_interval. While all senders are busy, the writes of everyone waiting queue up and leave together in the next request (up to batch_size), so many threads each waiting on their own writes still get batched requests. client.flush() waits for everything written so far; close() (or leaving the with block, or interpreter exit) flushes before shutting down.

Data is serialized before write() returns, so the objects you passed can be reused immediately.

Done-callbacks and on_error handlers run on the client's sender threads: they must not wait for the same client (flush(), close(), or result() of a pending write raise RuntimeError there), and should hand slow work to another thread.

Writing the same point twice

Requests are sent in parallel (write.concurrency, 16) and retried independently, so batches can reach the server in a different order than they were written. Lines within one request keep their order. This matters only when the same point (series, field and timestamp) is written more than once with different values, for example a correction, or a "latest state" kept at a fixed timestamp: the older value may be stored last. For such writes, wait for the first write's future before sending the next, or set write.concurrency = 1, which sends one request at a time in order, retries included.

Failures in the background

A batch that still fails after retries (or that the server rejects) is reported:

• to the futures of the write() calls whose points it carried: result() / await raise one error for the call, and future.errors lists every batch's error. When the server rejects single lines (InfluxDB 3 partial writes, and the lines InfluxDB 2, or InfluxDB 3 on /api/v2/write, refuses a whole request for, see Server behaviour handled for you), the rest of the batch is written and only the calls that contained a rejected line fail, with a PartialWriteError whose line_errors give each rejected line (as sent) and its number within that call;

• to on_error, if given, once per failed batch, with the exact line protocol that was not written (just the rejected lines, when the server rejected single lines), ready to dead-letter and replay. failure.retryable tells transient failures (network, outage, client closing) from data the server refused; it is true when some lines failed each way, as replaying them writes those that failed transiently:

  def dead_letter(failure: WriteFailure) -> None:
      if failure.retryable:  # replay later with client.write(lines, precision=failure.precision)
          DEAD_LETTERS.write_text("\n".join(failure.lines))
  
  
  client = InfluxClient.from_config(on_error=dead_letter)
  

• otherwise by flush() / close(), which raise a WriteError for failures nobody has seen (no result(), exception(), await or errors on their future), so fire-and-forget callers cannot lose data silently and synchronous callers are not told twice. An on_error handler that raises leaves the failure to flush() / close() as well;

• always by an ERROR log (rate limited) and the points_failed_total metric.

Records rejected by validation with on_invalid = "drop" are counted in result.dropped, and result.rejected holds their ValidationErrors (index, code, key, message; DataFrame rows by row number).

Backpressure

write.max_pending_bytes (128 MiB) bounds buffered plus in-flight data. When the buffer is full, write.on_full decides: block (default) waits up to block_timeout (60 s, then BufferFullError), drop discards new data (counted in points_dropped_total{reason="buffer_full"} and the future's dropped), and raise raises BufferFullError at once. BufferFullError, ClientClosedError and ValidationError carry points_enqueued: the points of that call that were already buffered and will be sent (do not re-send those).

asyncio

async with AsyncInfluxClient.from_config() as client:
    future = await client.write(points)  # buffered; never blocks the event loop
    result = await future  # acknowledged
    rows = await client.query("SELECT ...")
    async for chunk in client.query_stream("SELECT ..."):
        ...

The asyncio client shares the thread-based engine: large inputs are serialized and backpressure is awaited in worker threads, and streams of small writes yield to the event loop regularly. AsyncInfluxClient.wrap(client) puts an asyncio facade over an existing InfluxClient. Calling the synchronous client's blocking methods (result(), flush(), query(), close()) on a running event loop logs a warning.

Choosing tags

Every distinct combination of tag values is a series, and InfluxDB's cost grows with the number of series (InfluxDB 2 keeps an index of all series in memory). Tag values should come from a bounded set: host, region, service, endpoint, status class, sensor id. Values that are unique or unbounded (request ids, user or session ids, trace ids, timestamps, free text, measured values) belong in fields. sluicebox logs a warning when a measurement exceeds 150,000 distinct tag sets in one process.

Configuration and secrets

Settings come from, highest priority first: keyword overrides, environment variables (SLUICEBOX_WRITE__BATCH_SIZE=50000), a .env file, a secrets directory (one file per secret, e.g. Docker/Kubernetes /run/secrets/sluicebox_token), the TOML file, then defaults.

load_settings("sluicebox.toml")  # explicit file
load_settings("app.toml", section="services.influx")  # a table inside your app's config
load_settings(env_prefix="ARCHIVE_")  # a second client: ARCHIVE_TOKEN, ...
load_settings(secrets_dir="/run/secrets")
InfluxClient.from_config(write={"batch_size": 50_000})  # same arguments, plus overrides
InfluxClient(settings, write={"concurrency": 4})  # settings plus overrides
settings.with_overrides(query={"timeout": "5m"})  # a modified copy

Some environment variables steer loading itself (they may also be in .env):

The token is a pydantic.SecretStr: never printed, logged or put in a repr. sluicebox refuses to load a TOML file that contains a token (at the top level or in a table such as [connection]; fields or tags named token are fine) or a proxy password, since that file is meant to be committed. Unknown keys in the TOML file, keyword overrides or prefixed .env entries are errors with a "did you mean" suggestion; unknown SLUICEBOX_* environment variables are logged with the setting they probably meant (SLUICEBOX_DATABASE -> SLUICEBOX_CONNECTION__DATABASE); other applications' .env entries are ignored. Invalid settings raise ConfigurationError listing every problem. Without a token, a 401 error says where the token was looked for.

Validation and schemas

Every record is checked while it is serialized, before anything is buffered:

validation.on_invalid = "raise" (default) raises ValidationError with a machine-readable code, the record index, measurement, key and a position-free message; "drop" skips the record, logs it (rate limited per code, measurement and key), counts it in points_dropped_total{reason=<code>} and stats().write.points_dropped, and returns it in WriteResult.rejected.

Codes: type_conflict, non_finite, out_of_range, string_too_long, invalid_name, reserved_name, invalid_tag_value, tag_field_conflict, unexpected_tag, missing_tag, unexpected_field, missing_field, unknown_measurement, no_fields, invalid_time, naive_datetime, malformed_record, unsupported_type, unsupported_record, invalid_line, invalid_encoding.

Raw line protocol is passed through unchanged by default (fastest), except that lines without a timestamp get the write() time like any other point; raw_lines = "validate" parses it so validation and tag injection apply too. Either way, a newline inside a quoted string field value belongs to the value, so the output of to_line_protocol() and WriteFailure.lines can be written back as is. Lines are never sent if the server would read them together with the next line, which could be another write() call's: a line that leaves a string value open (m s="cut 17000, e.g. truncated) or ends in a backslash, or one whose multi-line string value is not followed by a valid end of line (a string left open, then closed by the quote of a later line). Those are invalid_line, also when passed through: on the server they would take the lines after them along, without an error (see Server behaviour). Only lines with a quote or a backslash need this check. A str or bytes is one record, so with on_invalid = "drop" an invalid line drops the text it is in; write lines one by one, or as a list, to keep the others.

[validation]
int_as_float = true          # common choice for sensor data: 25 and 25.5 are the same field type

[measurements.cpu]
fields = { usage = "float", cores = "integer" }
tags = ["host", "region"]
required_tags = ["host"]

Tag injection

[tags]
static = { env = "production" }
from_env = { region = "AWS_REGION", pod = { var = "POD_NAME", default = "local" } }
on_conflict = "keep"           # the point's own value wins (or "overwrite", "error")

[[tags.rules]]                 # tags derived from content; named groups become template values
when.tags = { device = '^(?P<site>[a-z]+)-(?P<rack>\d+)$' }
when.measurement = "^sensor_"
set = { site = "{site}", rack = "{rack}" }

[[tags.rules]]
when.fields = { temperature = { gt = 80 } }   # gt ge lt le eq ne in regex
set = { alert = "overheating" }

Rule conditions (all given ones must hold): measurement (regex), tags (regex per tag), has_tags, missing_tags, has_fields, fields (comparisons per field). A from_env variable given as a plain name must be set; with { var, default } it is optional (an empty default leaves the tag out).

from sluicebox import tag_context

with tag_context(tenant=req.tenant, region=req.region):  # per thread / asyncio task
    client.write(points)

client = InfluxClient.from_config(
    tags={"host": socket.gethostname()},
    enrichers=[lambda measurement, tags, fields: {"unit": "celsius"} if "temperature" in fields else None],
)

Order: context tags, then static tags, then rules (which see the tags added so far), then enrichers. Templates can also use {measurement}, {tags[key]} and {fields[key]}; a rule whose template references a missing value does not apply. client.to_line_protocol(...) shows the result. The tags= argument adds static tags; [tags] settings go in the config file (or load_settings(tags={...})).

Queries

client.query("SELECT host, usage FROM cpu WHERE host = $host", params={"host": "web-1"})  # InfluxDB 3
client.query("SELECT * FROM cpu WHERE time >= $since", params={"since": since})  # aware datetime
client.query("SELECT * FROM cpu", language="influxql")
client.query("from(bucket: params.bucket) |> range(start: -1h)", params={"bucket": "b"})  # InfluxDB 2
for chunk in client.query_stream("SELECT * FROM big_table"):  # bounded memory
    process(chunk.to_polars())

Prefer params to string formatting. On InfluxDB 3, datetime parameters are sent as RFC 3339 strings. On InfluxDB 2, Flux parameters are bound as escaped literals in an option params = {...} record (placed after the query's import statements), so params.name works on InfluxDB OSS too (the official client relies on a Cloud-only API feature); pass datetimes for Flux times. Writes are asynchronous: flush() before querying data you just wrote.

Results: InfluxDB 3 timestamp columns are timezone-aware (UTC). InfluxDB 2 results come from the official client's Flux parser: they include Flux's result, table, _start, _stop columns, times have microsecond resolution, and string values containing \n come back as \r\n. An empty result converts to an empty DataFrame.

Moving from InfluxDB 2 to 3: writes need no code changes (set version and the database). Queries do: InfluxDB 3 speaks SQL and InfluxQL, InfluxDB 2 Flux, and a query in the other language fails with a hint saying so.

Errors

All exceptions derive from SluiceboxError:

Transient failures (connection errors, timeouts, 429, 5xx) are retried with exponential backoff and jitter, honouring Retry-After, for up to retry.max_elapsed (5 minutes, so outages that long lose nothing); a broken pooled connection is retried immediately. Retries are safe because every point carries an explicit timestamp: a re-sent point overwrites itself.

Observability

Metrics (Prometheus, namespace sluicebox, label client = settings name): points_written_total, points_failed_total, points_dropped_total{reason}, write_batches_total{outcome}, write_bytes_total{kind=raw|sent}, write_retries_total{reason}, write_request_duration_seconds, write_batch_duration_seconds, write_batch_points, write_buffer_bytes, write_buffer_limit_bytes, write_max_batch_bytes (lowered after HTTP 413), write_last_success_timestamp_seconds, write_inflight_requests, queries_total, query_duration_seconds, query_rows_total, errors_total{operation,error}, stage_duration_seconds{stage}, client_info. The counters alerts need exist at zero from the start. Pass registry= to use your own registry; set [metrics] port to serve /metrics directly (one port per process). Clients with the same name share series.

Alerts worth having: increase(sluicebox_points_failed_total[5m]) > 0, increase(sluicebox_points_dropped_total[5m]) > 0, sluicebox_write_buffer_bytes / sluicebox_write_buffer_limit_bytes > 0.8, time() - sluicebox_write_last_success_timestamp_seconds > 300, and a rising sluicebox_write_retries_total.

Logging goes to the sluicebox logger hierarchy (sluicebox.write, .query, .validation, .transport, .client, .config); the library only adds a NullHandler, so records reach your own handlers. Records carry structured context (client, database, points, error, status, code, measurement, key) in record.influx, which JsonFormatter emits as JSON keys along with any extra= fields. For applications without logging setup, configure_logging(level="INFO", format="json") or [logging] configure = true installs a stderr handler on the sluicebox logger (which then stops propagating). Repeated warnings are rate limited, and close() reports how many were suppressed. Tokens never reach log records. Data lost at interpreter exit is also printed to stderr when no logging is configured.

Profiling: stage timings (serialize, backpressure, queue, compress, request, batch, query) feed stage_duration_seconds and client.stats().stages (count, total and max seconds per stage; serialize and batch are counted per batch). Slow batches and queries are logged with their duration. For a deep dive:

with client.profile("write.prof", memory=True) as report:
    client.write(points).result()
print(report.summary(limit=15))  # top functions, wall time, peak memory

memory=True traces every allocation and slows code down many times over; use it on short runs. client.stats() returns counters (points written/failed/dropped, retries, bytes) and buffer state.

Running in production

• Check at startup: client.check() verifies the server version, token, organization and bucket (with an empty write) so misconfiguration fails the deploy, not the first batch.
• Shutdown: call client.close() (or use with). Python's default SIGTERM handler exits without running cleanup, so install one that triggers your normal shutdown. close_timeout (20 s) stays below Kubernetes' 30 s grace period; data that could not be sent in time is reported (log, on_error, WriteError), never dropped silently.
• Telemetry that must never slow the application: on_full = "drop" (drops are counted and logged). Pipelines that must not lose data keep block (with a block_timeout).
• Memory: process memory can grow by about 3x max_pending_bytes while the buffer is full (an outage); size it for the container's limit. On free-threaded Python every thread that writes also keeps serializer caches of its own (up to 50,000 serialized tag sets per measurement and 200,000 escaped tag values): with many distinct series written from many threads, that adds up.
• Outages: batches are retried for retry.max_elapsed (5 minutes); meanwhile the buffer fills and backpressure applies.
• Processes: create one client per process. With pre-fork servers (gunicorn) or multiprocessing with fork, create clients after the fork (e.g. in post_fork); a client inherited across fork() keeps working in the child, but data buffered before the fork belongs to the parent, and InfluxDB 3 queries (gRPC) cannot run in a child forked after the parent created a query client (sluicebox raises instead of hanging). For metrics from several processes, use prometheus_client's multiprocess mode or a port per process.

Performance and tuning

Measured on an Apple M5 Pro against local Docker containers (InfluxDB 2.9.1, InfluxDB 3.12 Core with an in-memory object store), 3 tags and 4 fields per point, full validation on, with every run verifying the stored point count. uv run python benchmarks/bench_write.py --help reproduces them.

Serialization alone: 1.15 M points/s per thread (dicts, validation and type locking on) versus 0.36-0.38 M points/s for the official clients' Point serialization (benchmarks/bench_serialize.py). Against a local sink that accepts everything instead of a server, the client itself writes 860 k points/s from a list of dicts and 380 k points/s with one write() per point, checking that every point arrives exactly once (benchmarks/bench_client.py).

What matters most:

• Requests in flight. InfluxDB 3 acknowledges a write when its WAL flushes (every second by default), so its throughput is roughly concurrency x batch_size per second. The defaults (16 x 25,000 lines) were 6-7x faster on InfluxDB 3 than 4 x 10,000, and the official clients send one request at a time. Over a 40 ms round trip the defaults reached 460 k points/s on InfluxDB 3 versus 20 k points/s with 4 x 5,000.
• gzip (on by default) barely matters on localhost but is decisive on real links: on a 1 Mbit/s uplink it was over twice as fast.
• DataFrames are serialized vectorized: write() returns 4-6x sooner than for the same rows as dicts, leaving the server as the only bottleneck.
• write.no_sync = true (InfluxDB 3 Core/Enterprise) acknowledges before the WAL is persisted: lower latency, weaker durability.
• One client per process, shared by all threads, is the intended setup. Serialization of records runs at about 1.15 M points/s on one core. With the GIL, threads do not add to that: to use more cores, write DataFrames (vectorized in Rust) or run one client per process. On free-threaded Python (3.14t) threads do add: each thread that writes gets state of its own in the client (a staging lane in the buffer, serializer caches), so threads writing at once share no lock or object. One write() per point against a local sink went from 400 k points/s on one thread to 1.1 M on four and 1.6 M on eight (benchmarks/bench_client.py --threads 2,4,8, Apple M5 Pro); before, eight threads reached half the rate of one. With write.concurrency = 1, whose requests leave in the order the lines were written, all threads share one lane.

Remote servers, TLS and proxies

Beyond loopback, the test suite runs every write scenario through Toxiproxy (latency, bandwidth limits, connection resets, responses lost in a blackhole, outages, truncated uploads, sliced TCP streams), against servers with native TLS, behind nginx with a 256 KB body limit or mutual TLS, and through an HTTP forward proxy. The client:

• pools keep-alive connections with TCP keepalive (dead peers and NAT/LB idle drops are detected);
• retries a broken pooled connection immediately, everything else with backoff;
• learns a request size limit from HTTP 413 responses (including a proxy's HTML page) and splits batches accordingly, instead of re-uploading oversized bodies (and stops splitting when even tiny requests are refused, which is not about size);
• never duplicates data when a response is lost after the server stored the batch, and the servers never ingest a truncated upload (both verified).

A path in connection.url (e.g. https://gateway.internal/influx) prefixes every API path, for servers behind path-routing proxies. TLS: connection.ca_cert for a private CA, client_cert/client_key for mutual TLS (applied to writes and to InfluxDB 3's Flight queries; tested through an nginx that requires client certificates), verify_ssl = false to disable verification (logged once). Python 3.13+ verifies certificates strictly (VERIFY_X509_STRICT): an internal CA certificate needs basicConstraints and keyUsage extensions. Proxies: connection.proxy, with credentials only via the environment (SLUICEBOX_CONNECTION__PROXY=http://user:pass@proxy:3128); InfluxDB 3 queries tunnel through it.

InfluxDB 3 Cloud Serverless, Dedicated and Clustered only offer the v2-compatible endpoint: set write.api = "v2" (a 404 from /api/v3/write_lp says so in the error).

Server behaviour handled for you

Found by testing against InfluxDB 2.9 and InfluxDB 3.12, and handled or validated:

• An InfluxDB 2 server answers the InfluxDB 3 write endpoint with its web UI and HTTP 200: a web page is never taken for a successful write, and check() names the version mismatch.
• Backslashes: InfluxDB 3 unescapes \\ in names and tags, InfluxDB 2 does not. sluicebox escapes per version, so values like c:\dir round-trip on both.
• InfluxDB 2 silently drops points whose measurement starts with # (a comment line), and stores but can never query measurement names containing =. Both are rejected client-side.
• Names or tag values ending in a backslash, tabs in names (InfluxDB 3), NaN, int64 overflow, timestamps beyond 2262 and strings over 1 MiB are rejected by the servers. Validation rejects them before they are sent.
• InfluxDB 2 parses a whole request before storing any of it, so one line it cannot parse (raw line protocol is sent unvalidated by default) fails the whole batch, other write() calls' points included, and a replay of the dead-lettered batch fails again. Its HTTP 400 quotes each such line: sluicebox takes those out and sends the rest again, so every valid line is written and only the bad ones fail, each for its own write() call. When the quotes do not match the lines sent, the request is halved instead, as for HTTP 413. This costs at most 64 extra requests per batch, and a request the server may have partly stored is never resent.
• Both servers end a line at a newline only outside string values, and take a backslash to escape any character, the newline too. A line leaving a string value open, or ending in a backslash, therefore runs on into the next line of the request: InfluxDB 3 neither stores nor reports the lines it swallows (write_lp named 1 error for 3 lines, and stored 1 of the 3 valid ones), and InfluxDB 2 stores two such lines and the lines between as one point with HTTP 204. Lines from many write() calls share a request, so write() rejects such lines (invalid_line), unvalidated raw line protocol too. InfluxDB 2 writes the backslash of a name like a\ b as a\\ b, which its line splitting reads as an escaped backslash and a space; points whose lines it would then read differently are rejected (invalid_name).
• On InfluxDB 3, /api/v3/write_lp (the default for version 3) keeps the valid lines of a batch and names the rejected ones. /api/v2/write (write.api = "v2") stores nothing of a request with one line it cannot parse, or that gives a column another type than its table has, and names the reason but not the line. sluicebox sends the lines that reason can be about (lines it cannot parse either, or that give the column the refused type, a request per table) apart from the others, which go in one request. Lines of one table that all give a field the refused type fail together, and their type conflict teaches the client the server's type. Otherwise the request is halved, which costs about 2 x log2(lines) requests per refused line: about 30 in a 25,000-line batch. This matters because a request the server stores takes up to a second (it acknowledges on its WAL flush), while one it refuses takes milliseconds. The same limit of 64 extra requests per batch applies.
• InfluxDB 3 Cloud Serverless, Dedicated and Clustered keep the valid lines of a request by default (Serverless always), so after their HTTP 400 for some lines nothing is resent: the whole request is reported as failed (replaying it is harmless, as every point carries its timestamp).
• Flux parameters (params.x) are a Cloud-only feature; sluicebox binds them so they work on OSS.

Development

uv sync                                         # environment with all extras and dev tools
docker compose up -d --wait                     # InfluxDB 2 (:18086) and 3 Core (:18181)
docker/tls/generate.sh && docker compose --profile network up -d --wait   # + TLS, Toxiproxy, nginx

uv run pytest tests/unit                        # fast, no servers needed
uv run pytest                                   # everything (server tests skip when unavailable)
uv run ruff check . && uv run ruff format --check . && uv run mypy
uv run python benchmarks/bench_write.py --server 3 --points 500000
uv run python benchmarks/bench_client.py        # the client alone, against a local sink

On first start the compose stack generates random throwaway credentials into docker/secrets/ (git-ignored), where the tests and benchmarks read them; the servers keep their data in memory. InfluxDB 3 Core allows five databases, so the tests share one.

Releasing: set version in pyproject.toml, commit, and publish a GitHub release tagged v<version> (gh release create v0.1.0 --generate-notes). The release workflow checks that the tag matches the version, runs the unit tests, builds, and uploads to PyPI by trusted publishing.

License

MIT; see LICENSE.