aptdata

v0.1.0 · A declarative, extensible framework for building smart data pipelines in Python.

Overview

aptdata is built around three universal abstractions — System, Flow, and Component — that cover every data-processing paradigm in a single, coherent model:

flowchart TD
    I["IComponent / IFlow / ISystem\n@dataclass + ABC — pure interfaces"]
    B["BaseComponent / BaseFlow / BaseSystem\n@pydantic_dataclass — validated fields"]
    Y["Your concrete implementations"]

    I --> B --> Y

Datasets remain the fundamental data-exchange contract (IDataset / BaseDataset). Every outcome from the CLI is emitted as a machine-readable JSON line, making aptdata a natural fit for AI orchestrators, CI/CD pipelines and scripted workflows.

Requirements

• Python ≥ 3.10
• Poetry (for development)

Installation

From PyPI

pip install aptdata

Optional extras

pip install aptdata[pandas]   # pandas support
pip install aptdata[spark]    # PySpark support
pip install aptdata[plugins]  # REST, PostgreSQL, Parquet I/O
pip install aptdata[ai]       # MCP server for AI agents
pip install aptdata[all]      # everything

From source (development)

git clone https://github.com/strondata/smart-data.git
cd aptdata
poetry install

Quick start

from pydantic.dataclasses import dataclass as pydantic_dataclass
from aptdata.core import (
    BaseDataset, IDataset,
    BaseComponent, ComponentMeta, ComponentKind,
    BaseFlow, IFlow,
    BaseSystem,
)

@pydantic_dataclass
class MemoryDataset(BaseDataset):
    def __post_init__(self): self._data = None
    def read(self): return self._data
    def write(self, data): self._data = data

@pydantic_dataclass
class DoubleComponent(BaseComponent):
    def validate_inputs(self, inputs: list[IDataset]) -> bool:
        return len(inputs) == 1
    def execute(self, inputs: list[IDataset]) -> list[IDataset]:
        out = MemoryDataset(uri="memory://out")
        out.write([x * 2 for x in inputs[0].read()])
        return [out]

@pydantic_dataclass
class ETLFlow(BaseFlow):
    def __post_init__(self):
        self._nodes = {}
        self._edges = []
        self._compiled = False
    def add_component(self, c): self._nodes[c.component_id] = c
    def connect(self, src, tgt, condition=None): ...
    def compile(self): self._compiled = True
    def run(self, inputs): return inputs  # wire your logic here

@pydantic_dataclass
class MySystem(BaseSystem):
    def __post_init__(self): self._flows: list[IFlow] = []
    def register_flow(self, flow): self._flows.append(flow)
    def run(self):
        for flow in self._flows:
            flow.run([])

# Register and run via CLI
from aptdata.plugins import registry
registry.register("my_system", MySystem)

aptdata run my_system
# {"event": "pipeline.started", "pipeline": "my_system", "env": "dev", "dry_run": false, "trace_id": null}
# {"event": "pipeline.completed", "pipeline": "my_system", "env": "dev", "dry_run": false, "elapsed_seconds": 0.001, "trace_id": null}

aptdata run resolves the pipeline in its own process, so the registration must be importable there — load it with aptdata plugin load my_package.module (or run a declarative YAML with aptdata config run) instead of relying on a script that already exited.

CLI reference

aptdata run PIPELINE [--env ENV] [--dry-run]
aptdata monitor [--refresh SECONDS]
aptdata scaffold PROJECT_NAME [--template TEMPLATE] [--output PATH]
aptdata schema export --output schema.json
aptdata system list [--json]
aptdata system info NAME [--json]
aptdata system validate NAME
aptdata plugin list [--json]
aptdata plugin inspect NAME [--json]
aptdata plugin preview READER [--limit N]
aptdata plugin load MODULE_PATH
aptdata config validate PATH
aptdata config init [--output PATH] [--template]
aptdata config show PATH
aptdata config run PATH [--env ENV]
aptdata telemetry status [--json]
aptdata telemetry export [--format json]
aptdata mesh list [--dir DIR] [--json]
aptdata mesh run COMPONENT [--dir DIR] [--dry-run] [--json]
aptdata mesh build COMPONENT [--dir DIR] [--json]
aptdata agents list [--file PATH] [--enabled] [--json]
aptdata agents send AGENT_ID PROMPT [--file PATH] [--json]
aptdata agents route TEXT [--file PATH] [--json]
aptdata agents dispatch TEXT [--file PATH] [--json]
aptdata agents resolve CAPABILITY [--file PATH] [--json]
aptdata project init NAME [--out PATH] [--json]
aptdata project plan PROJECT_FILE [--file PATH] [--json]
aptdata project run PROJECT_FILE [--file PATH] [--json]
aptdata setup [--file PATH]            # wizard de diagnóstico + configuração
aptdata setup --check [--json]         # diagnóstico não interativo (CI)
aptdata converse TEXT [--session ID] [--file PATH] [--yes] [--json]
aptdata converse --confirm DECISION_ID [--choose AGENT] [--session ID]
aptdata telegram [--file PATH] [--token-env VAR]
aptdata viz [--file PATH] [--host HOST] [--port PORT]
aptdata obs summary [--json]
aptdata obs tail [--limit N] [--kind KIND] [--run-id ID] [--json]
aptdata mcp-start [--transport TRANSPORT]
aptdata interactive

Top-level commands (run, scaffold, schema export, mcp-start) always emit machine-readable JSON lines. Sub-commands marked [--json] above are dual-mode: Rich tables/panels by default, one JSON line with --json. The remaining text-only commands (system validate, plugin preview, plugin load, config *; telemetry export uses --format json) do not take --json yet.

Scaffold templates

aptdata scaffold my_lakehouse --template medallion
aptdata scaffold my_job --template job-wheel
aptdata scaffold my_service --template docker-compose-app

Processing Engines

Engine-agnostic transformation wrappers for pandas and PySpark:

from aptdata.plugins.transform import PandasTransformer

def clean(df):
    return df.dropna().drop_duplicates()

transformer = PandasTransformer("clean", clean)
result = transformer.transform(my_dataset)

See Transform Engines docs for PySpark usage.

Data Quality & Contracts

from aptdata.plugins.quality import (
    EnforcementMode, ExpectColumnToNotBeNull,
    QualityValidator, SchemaContract,
)

validator = QualityValidator(
    expectations=[ExpectColumnToNotBeNull("id")],
    enforcement=EnforcementMode.ABORT,
)
clean_data = validator.validate(raw_df)

See Quality docs for all built-in expectations.

Data Governance

from aptdata.plugins.governance import (
    BusinessRule, DatasetCatalog, DatasetCatalogEntry, LineageStore,
)
from aptdata.core.lineage import LineageGraph, LineageNode, LineageEventType

# Lineage tracking
graph = LineageGraph(run_id="run-1", workflow_name="etl")
graph.add_node(LineageNode(dataset_uri="s3://raw/data", event_type=LineageEventType.READ))

store = LineageStore()
store.save(graph)

See Governance docs for the full API.

Setup

O ponto de partida é o wizard de setup — ele diagnostica o ambiente com transparência (agents.yaml, router, política de roteamento, Telegram, observabilidade, viz) e configura o que faltar:

aptdata setup                 # wizard guiado (cria agents.yaml inicial, canal Telegram)
aptdata setup --check --json  # relatório de saúde para CI/painéis (exit 1 se incompleto)

O token do Telegram nunca é gravado em arquivo — fica na env TELEGRAM_BOT_TOKEN; o agents.yaml guarda só o nome da variável. Depois do setup: aptdata converse (conversa headless), aptdata viz (painel + traço ao vivo), aptdata telegram (bot fino). Veja docs/telegram.md.

AI Agents & MCP Server

aptdata ships with a built-in Model Context Protocol server (mcp-start). This transforms AI assistants (like Claude, Copilot, or Devin) into autonomous data engineers with direct access to:

• Pipeline Execution: Trigger and monitor data flows (run_flow).
• Agents: List the agent registry and route/dispatch prompts (list_agents, dispatch).
• Data Quality: Audit the latest quality test results (quality://reports/...) — placeholder data for now.
• Data Governance: Read business rules to prevent violations (governance://rules) — placeholder data for now.
• Lineage: Trace upstream dependencies and column-level provenance (get_pipeline_lineage) — placeholder data for now.

The quality/governance/lineage resources currently return illustrative mock payloads (see aptdata/mcp/server.py); they define the contract while the real stores are being built.

aptdata mcp-start --transport stdio

See the MCP Documentation for setup instructions.

Release process

Releases are automated via the Release workflow. After a PR is merged into main, the CI reads its labels and bumps the version accordingly.

The workflow will:

1. Detect the merged PR and its labels.
2. Run bump-my-version bump <part> to update pyproject.toml and aptdata/__init__.py.
3. Create a chore(release): bump version to X.Y.Z commit and a vX.Y.Z tag.
4. Push the commit and tag to main.
5. The tag push automatically triggers the Publish to PyPI workflow.

Branch protection note: GitHub Actions must have read and write permissions (Settings → Actions → General → Workflow permissions) and, if branch protection is enabled on main, the rule must allow GitHub Actions to bypass it.

Development

make install   # install all dependencies
make test      # run the test suite
make lint      # lint with ruff
make docs      # build the documentation

Documentation

Full documentation is available in the docs/ directory and can be served locally with:

mkdocs serve

License

MIT