Incremental Extraction

Incremental extraction fetches records new since the last run, using a persisted watermark (timestamp, sequence ID, or cursor). PyCharter provides standalone state helpers; your scheduled job must read the watermark, inject it into the source configuration, and decide when it is safe to save a new one.

The native batch DAG does not call these helpers. Its retained settings.state_store field currently has no runtime effect. The flow below describes manual job orchestration, not automatic DAG checkpointing.

How it works

Run N
  ├─ Load last watermark from state store (e.g. "2024-06-01T00:00:00")
  ├─ Pass watermark to extractor as a parameter
  ├─ Extract only records where updated_at > watermark
  ├─ Load records
  └─ Save new watermark to state store (e.g. "2024-06-02T00:00:00")

Run N+1
  ├─ Load watermark: "2024-06-02T00:00:00"
  └─ Extract only records since that timestamp

State persistence is handled by a StateStore. Two implementations ship with PyCharter:

Store	Best for
FileStateStore	Single-process scripts, local development
SqliteStateStore	Scheduled jobs, lightweight production, multi-pipeline tracking

---

Quick start

1 · Choose a state store

FileStateStoreSqliteStateStore

from pycharter.pipeline_generator.state import FileStateStore

store = FileStateStore(path="./.state")
# Creates .state/<pipeline_name>.json per pipeline

from pycharter.pipeline_generator.state import SqliteStateStore

store = SqliteStateStore(path="./.state/state.db")
# Creates a pipeline_state table in the SQLite file

2 · Read and update the watermark

import asyncio
from pycharter import Pipeline
from pycharter.pipeline_generator.state import SqliteStateStore, PipelineState

PIPELINE = "orders"


async def run(next_watermark: str):
    store = SqliteStateStore("./.state/state.db")

    # Load last state (None on first run)
    state = store.get(PIPELINE)
    last_watermark = state.watermark if state else None

    pipeline = Pipeline.from_dir(
        "pipelines/orders/",
        variables={
            # Inject watermark as a variable — use it in the extract step source
            "SINCE": last_watermark or "1970-01-01T00:00:00",
            "UNTIL": next_watermark,
        },
    )

    result = await pipeline.arun()

    if result.success and result.rows_failed == 0 and result.record_outcomes.complete:
        # The caller chose a source-consistent upper bound before extraction.
        # This example assumes skipped/quarantined records are handled separately
        # and destination acknowledgement is sufficient for this job's policy.
        store.save(
            PipelineState(
                pipeline_name=PIPELINE,
                watermark=next_watermark,
                last_run_id=result.run_id,
                last_run_at=result.start_time.isoformat() if result.start_time else None,
                run_count=(state.run_count if state else 0) + 1,
            )
        )
        print(f"Loaded {result.rows_loaded} rows. Next watermark: {next_watermark}")
    else:
        print(f"Pipeline failed — watermark NOT updated")
        for err in result.errors:
            print(f"  {err}")


# Replace this illustrative bound with one obtained from your source.
asyncio.run(run("2024-06-02T00:00:00Z"))

The source must honor both bounds. A successful result alone does not establish transactional delivery, and a failed loader can have written records before raising. Define replay/idempotency and skipped/quarantined-record handling in the job before advancing a production watermark. PipelineResult does not calculate a maximum source timestamp; a run UUID is not a source watermark.

3 · Use the watermark in the extract step

# pipeline.yaml — extract step (HTTP)
- id: extract_orders
  type: extract
  source:
    type: http
    url: https://api.example.com/orders
    params:
      updated_after: "${SINCE}"
      updated_before: "${UNTIL}"
      page_size: 500

# pipeline.yaml — extract step (database)
- id: extract_orders
  type: extract
  source:
    type: database
    connection_string: "${DATABASE_URL}"
    query: |
      SELECT * FROM orders
      WHERE updated_at > '${SINCE}'
        AND updated_at <= '${UNTIL}'
      ORDER BY updated_at

---

State model

PipelineState is a frozen dataclass:

@dataclass(frozen=True, slots=True)
class PipelineState:
    pipeline_name: str
    watermark: str | None = None  # High-water mark (ISO 8601, sequence ID, etc.)
    last_run_id: str | None = None  # UUID of the last successful run
    last_run_at: str | None = None  # ISO 8601 timestamp of last run
    run_count: int = 0  # Total number of successful runs
    metadata: dict[str, Any] = field(default_factory=dict)  # Arbitrary extras

Serialize/deserialize with .to_dict() / PipelineState.from_dict(data).

---

Managing state manually

from pycharter.pipeline_generator.state import FileStateStore, PipelineState

store = FileStateStore()

# Inspect state
state = store.get("orders")
print(state.watermark, state.run_count)

# List all tracked pipelines
print(store.list_pipelines())

# Reset state (re-process from the beginning)
store.delete("orders")

# Backfill from a specific point
store.save(
    PipelineState(
        pipeline_name="orders",
        watermark="2024-01-01T00:00:00",
    )
)

---

Creating a store from config

If you drive pipeline configuration from a YAML/dict config, use the factory helper:

from pycharter.pipeline_generator.state import create_state_store

store = create_state_store(
    {
        "backend": "sqlite",  # "file" or "sqlite"
        "path": "./.state/state.db",
    }
)

---

Choosing a watermark strategy

Strategy	When to use
Timestamp (updated_at)	Source has a reliable modification timestamp
Sequence / auto-increment ID	Source has a monotonically increasing integer key
Cursor / page token	Pagination API returns a cursor for the next page

Use a run UUID to identify execution evidence; it cannot substitute for a source position when the source has no natural watermark.

Watermark reliability

Timestamps can be unreliable if clock skew exists between producer and consumer. Add a small overlap (updated_at >= watermark - 5 minutes) to avoid missing recently written records.

---

FileStateStore vs SqliteStateStore

Attribute	FileStateStore	SqliteStateStore
Storage	JSON files	SQLite table
Concurrent access	No file locking	SQLite transactional writes; callers must serialize checkpoint read/update for the same pipeline
Multiple pipelines	One file each	One table, one row each
Portability	Copy/move files	Copy SQLite file
Dependencies	None	sqlite3 (stdlib)
Best for	Local dev, single process	Scheduled jobs, multi-pipeline

---

See also

• Building ETL Pipelines
• Async Execution Model
• Testing Pipelines — test incremental logic with MockExtractor
• PipelineState API reference