Python SDK for the Mammoth Analytics platform. Build data pipelines, apply transformations, and export results -- all from Python.
Prefer the terminal? This repository also ships
mammoth-cli, a command-line interface built for both people and autonomous agents. See Command-line interface below.
pip install mammoth-ioRequires Python 3.10+.
from mammoth import MammothClient
client = MammothClient(
api_key="your-api-key",
api_secret="your-api-secret",
workspace_id=11,
)
client.set_project_id(42)
# Get a view and inspect its columns
view = client.views.get(1039)
print(view.display_names) # ["Customer", "Region", "Sales", "Date"]
print(view.column_types) # {"Customer": "TEXT", "Region": "TEXT", "Sales": "NUMERIC", ...}
# After any transformation, display_names is automatically refreshed
# (including pipeline-added columns like those created by math/set_values/add_column).
# Use get_metadata() to inspect the full list:
view.math("Sales * 1.1", new_column="Revenue")
print(view.display_names) # now includes "Revenue"
meta = view.get_metadata() # [{"display_name": "Revenue", "internal_name": "column_x1y2", "type": "NUMERIC"}, ...]
# Fetch data — returns {"data": [rows...], "paging": {...}}
result = view.data(limit=100)
rows = result["data"]You can also extract IDs directly from a Mammoth URL:
from mammoth import MammothClient, parse_path
ids = parse_path("https://app.mammoth.io/#/workspaces/11/projects/42/views/1039")
# {"workspace_id": 11, "project_id": 42, "dataview_id": 1039}
client = MammothClient(
api_key="your-api-key",
api_secret="your-api-secret",
workspace_id=ids["workspace_id"],
)
client.set_project_id(ids["project_id"])
view = client.views.get(ids["dataview_id"])The View object is the central interface. It wraps a single dataview and exposes 25+ transformation methods. Each method sends a pipeline task to the API and automatically refreshes the view metadata — including any new columns added by the transformation.
view.math(expression="Price * Quantity", new_column="Revenue")
print("Revenue" in view.display_names) # True — refreshed automatically
# Inspect full column list (display_name, internal_name, type)
for col in view.get_metadata():
print(col)from mammoth import Condition, Operator, FilterType
# Keep rows where Sales >= 1000
view.filter_rows(Condition("Sales", Operator.GTE, 1000))
# Remove rows where Region is empty
view.filter_rows(
Condition("Region", Operator.IS_EMPTY),
filter_type=FilterType.REMOVE,
)from mammoth import SetValue, ColumnType
view.set_values(
new_column="Risk Level",
column_type=ColumnType.TEXT,
values=[
SetValue("High", condition=Condition("Sales", Operator.GTE, 10000)),
SetValue("Medium", condition=Condition("Sales", Operator.GTE, 5000)),
SetValue("Low"),
],
)# String expressions are parsed automatically
view.math("Price * Quantity", new_column="Revenue")
view.math("(Price + Tax) * 1.1", new_column="Grand Total")from mammoth import JoinType, JoinKeySpec
other_view = client.views.get(2050)
view.join(
foreign_view=other_view,
join_type=JoinType.LEFT,
on=[JoinKeySpec(left="Customer ID", right="Customer ID")],
select=["Category", "Tier"],
)from mammoth import AggregateFunction, AggregationSpec
view.pivot(
group_by=["Region"],
aggregations=[
AggregationSpec(column="Sales", function=AggregateFunction.SUM, as_name="Total Sales"),
AggregationSpec(column="Sales", function=AggregateFunction.AVG, as_name="Avg Sales"),
],
)from mammoth import WindowFunction, SortDirection
view.window(
function=WindowFunction.ROW_NUMBER,
new_column="Rank",
partition_by=["Region"],
order_by=[["Sales", SortDirection.DESC]],
)from mammoth import TextCase
# Change case
view.text_transform(["Customer Name"], case=TextCase.UPPER)
# Find and replace
view.replace_values(columns=["Status"], find="Pending", replace="In Progress")
# Split column
from mammoth import SplitColumnSpec
view.split_column(
"Full Name",
delimiter=" ",
new_columns=[SplitColumnSpec(name="First"), SplitColumnSpec(name="Last")],
)from mammoth import DateComponent, DateDiffUnit
# Extract year from a date column
view.extract_date("Order Date", DateComponent.YEAR, new_column="Order Year")
# Calculate difference between two dates
view.date_diff(DateDiffUnit.DAY, start="Start Date", end="End Date", new_column="Duration")
# Add 30 days to a date
from mammoth import DateDelta
view.increment_date("Ship Date", delta=DateDelta(days=30), new_column="Expected Arrival")from mammoth import CopySpec, ConversionSpec
# Add an empty column
view.add_column("Notes", ColumnType.TEXT)
# Delete columns
view.delete_columns(["Temp1", "Temp2"])
# Copy a column
view.copy_columns([CopySpec(source="Sales", as_name="Sales Backup", type=ColumnType.NUMERIC)])
# Combine (concatenate) columns
view.combine_columns(["First Name", "Last Name"], new_column="Full Name", separator=" ")
# Convert column type
view.convert_type([ConversionSpec(column="ZipCode", to=ColumnType.TEXT)])
view.convert_type([ConversionSpec(column="Order Date", to=ColumnType.DATE, format="MM/DD/YYYY")])from mammoth import FillDirection
# Fill missing values
view.fill_missing("Revenue", direction=FillDirection.LAST_VALUE)
# Keep top 100 rows
view.limit_rows(100)
# Remove duplicates
view.discard_duplicates()
# Unpivot columns to rows
view.unnest(["Q1", "Q2", "Q3", "Q4"], label_column="Quarter", value_column="Revenue")# AI-powered transformation
view.gen_ai(
prompt="Classify the sentiment of the review as Positive, Negative, or Neutral",
context_columns=["Review Text"],
new_column="Sentiment",
)
# Generate SQL from natural language (also adds pipeline task)
sql_query = view.generate_sql("count customers by region")
# Add a raw SQL query as a pipeline task
view.add_sql("SELECT region, COUNT(*) as cnt FROM data GROUP BY region")# List all tasks on a view
tasks = view.list_tasks()
# Delete a specific task
view.delete_task(task_id=123)
# Preview a task before applying
preview = view.preview_task({"MATH": {"EXPRESSION": [...]}})| Method | Description |
|---|---|
filter_rows() |
Filter rows by condition |
set_values() |
Label/insert values with conditional logic |
math() |
Arithmetic expressions |
small_large() |
Nth smallest or largest value across columns/constants |
join() |
Join with another view |
pivot() |
Group by and aggregate |
window() |
Window functions (rank, lag, running sum, etc.) |
crosstab() |
Pivot table |
text_transform() |
Change case, trim whitespace |
replace_values() |
Find and replace |
bulk_replace() |
Bulk find-and-replace with mapping |
split_column() |
Split by delimiter |
substring() |
Extract text by position or regex |
extract_date() |
Extract date components |
date_diff() |
Date difference |
increment_date() |
Add/subtract from dates |
add_column() |
Add empty column |
delete_columns() |
Remove columns |
copy_columns() |
Duplicate columns |
combine_columns() |
Concatenate columns |
convert_type() |
Change column data type |
fill_missing() |
Fill gaps forward/backward |
limit_rows() |
Keep top/bottom N rows |
discard_duplicates() |
Remove duplicate rows |
unnest() |
Unpivot columns to rows |
lookup() |
Lookup values from another view |
json_extract() |
Extract from JSON columns |
gen_ai() |
AI-powered transformation |
generate_sql() |
Generate SQL from natural language |
add_sql() |
Add raw SQL as pipeline task |
Methods that accept structured parameters use typed dataclasses for IDE autocomplete:
| Dataclass | Used by |
|---|---|
CopySpec |
copy_columns() |
ConversionSpec |
convert_type() |
AggregationSpec |
pivot() |
CrosstabSpec |
crosstab() |
JoinKeySpec |
join() on |
JoinSelectSpec |
join() select |
JsonExtractionSpec |
json_extract() |
The Condition class supports Python's & (AND), | (OR), and ~ (NOT) operators for composing filter logic.
from mammoth import Condition, Operator
# Simple conditions
high_sales = Condition("Sales", Operator.GTE, 10000)
west_region = Condition("Region", Operator.EQ, "West")
active = Condition("Status", Operator.IN_LIST, ["Active", "Pending"])
has_email = Condition("Email", Operator.IS_NOT_EMPTY)
# Combine with & (AND), | (OR), and ~ (NOT)
priority = high_sales & west_region # Both must be true
either = high_sales | west_region # At least one true
not_active = ~active # Negate a condition
complex_filter = (high_sales & west_region) | ~active # Nested logic
# Use anywhere conditions are accepted
view.filter_rows(priority)
view.set_values(
new_column="Flag",
column_type=ColumnType.TEXT,
values=[
SetValue("Priority", condition=high_sales & west_region),
SetValue("Normal"),
],
)
view.math("Sales * 1.1", new_column="Adjusted", condition=west_region)| Operator | Description |
|---|---|
EQ, NE |
Equal, not equal |
GT, GTE, LT, LTE |
Comparison |
IN_LIST, NOT_IN_LIST |
Value in/not in list |
IN_RANGE |
Between two bounds (inclusive); value is a 2-element list |
CONTAINS, NOT_CONTAINS |
Text contains/not contains |
ICONTAINS |
Case-insensitive text contains |
STARTS_WITH, ENDS_WITH |
Text prefix/suffix |
NOT_STARTS_WITH, NOT_ENDS_WITH |
Negated prefix/suffix |
IS_EMPTY, IS_NOT_EMPTY |
Null check |
IS_MAXVAL, IS_NOT_MAXVAL |
Max value in column |
IS_MINVAL, IS_NOT_MINVAL |
Min value in column |
# Keep rows where Amount is between 100 and 500 (inclusive)
view.filter_rows(Condition("Amount", Operator.IN_RANGE, [100, 500]))# Match "York", "york", "YORK" in City column
view.filter_rows(Condition("City", Operator.ICONTAINS, "york"))Pass a DateFunction enum member as the value with value_is_date_fn=True to compare against a dynamic date/time:
from mammoth import Condition, Operator, DateFunction
# Rows where Order Date is after today
view.filter_rows(Condition("Order Date", Operator.GT, DateFunction.TODAY, value_is_date_fn=True))
# Rows where Timestamp is on or after the current instant
view.filter_rows(Condition("Timestamp", Operator.GTE, DateFunction.NOW, value_is_date_fn=True))
# Rows where Date equals the maximum date value in that column
view.filter_rows(Condition("Date", Operator.EQ, DateFunction.MAX, value_is_date_fn=True))Available DateFunction values: NOW, TODAY, MAX, MIN, SYSTEM_DATE, SYSTEM_TIME.
# Upload a single file (returns dataset ID)
dataset_id = client.files.upload("sales_data.csv")
# Upload multiple files
dataset_ids = client.files.upload(["sales.csv", "customers.xlsx"])
# Upload an entire folder
dataset_ids = client.files.upload_folder("./data/")Supported formats: CSV, TSV, PSV, XLS, XLSX, ZIP, BZ2, GZ, TAR, 7Z, PDF, TIFF, JPEG, PNG, HEIC, WEBP. Maximum file size: 50 MB.
After upload, get a view for the new dataset:
dataset_id = client.files.upload("sales_data.csv")
views = client.views.list()
view = next(v for v in views if v.dataset_id == dataset_id)
print(view.display_names)# From a View object
path = view.export.to_csv("output.csv")
# From client with a known dataview ID
path = client.exports.to_csv(dataview_id=1039, output_path="output.csv")# From a View object
result = view.export.to_s3(file_name="monthly_report.csv")
# From client with a known dataview ID (parameter name is `file=`, not `file_name=`)
result = client.exports.to_s3(dataview_id=1039, file="monthly_report.csv")# PostgreSQL
view.export.to_postgres(
host="db.example.com",
port=5432,
database="analytics",
table="sales_summary",
username="user",
password="pass",
)
# MySQL
view.export.to_mysql(
host="db.example.com",
port=3306,
database="analytics",
table="sales_summary",
username="user",
password="pass",
)# From a View object — creates a new dataset named "Q1 snapshot"
new_dataset_id = view.export.to_dataset(dataset_name="Q1 snapshot")
# Shorthand
new_dataset_id = view.branch_out(dataset_name="Q1 snapshot")
# Append into an existing dataset
view.branch_out(dataset_name="Sales Archive", target_ds_id=500)# FTP — parameters: domain, directory, file, username, password
view.export.to_ftp(
domain="ftp.example.com",
directory="/exports",
file="sales.csv",
username="user",
password="pass",
)
# SFTP — password auth
view.export.to_sftp(
host="sftp.example.com",
username="user",
password="pass",
directory="/exports",
file_name="data.csv",
)
# SFTP — private-key auth
view.export.to_sftp(
host="sftp.example.com",
username="user",
ssh_key_authentication=True,
private_key="-----BEGIN OPENSSH PRIVATE KEY-----\n...",
)# `emails` is a list of recipient addresses (not `recipients`)
view.export.to_email(
emails=["team@example.com"],
subject="Q1 Sales Report",
)from mammoth import BigQueryExportType
view.export.to_bigquery(
selected_profile={"name": "my_dataset", "value": [["project_id", "dataset_id"]]},
selected_identity={"identity_config": {...}, "host": "sa@project.iam.gserviceaccount.com"},
table="sales_summary",
export_type=BigQueryExportType.REPLACE,
)from mammoth import OdbcType
# Publishes to a Mammoth-managed Postgres or BigQuery connection
view.export.publish_to_db(table="sales_dashboard", odbc_type=OdbcType.POSTGRES)from mammoth import RestAuthType, HttpMethod
view.export.to_rest_api(
base_url="https://api.example.com",
endpoint_path="/v1/records",
auth_type=RestAuthType.BEARER,
http_method=HttpMethod.POST,
auth={"token": "my-bearer-token"},
batch_size=500, # 1–10 000
timeout_seconds=30, # 5–300
)view.export.to_redshift(...)
view.export.to_elasticsearch(...)
view.export.to_azure_blob(...)
view.export.to_sharepoint(...)
view.export.to_onedrive(...)
view.export.to_tableau(...)
view.export.to_powerbi(...)
view.export.to_mssql(...)The SDK validates arguments before making any network call and raises specific exceptions so failures are caught early with an actionable message.
MammothError # base; all SDK exceptions inherit from this
├── MammothValidationError # invalid SDK arguments (raised before any API call)
├── MammothAPIError # API returned an error response
│ └── MammothAuthError # HTTP 401 — invalid credentials
├── MammothColumnError # column display name not found in view metadata
├── MammothExportError # export submitted but result dataset id didn't resolve
├── MammothJobTimeoutError # async job polling exceeded the timeout
├── MammothJobFailedError # async job completed with a failure status
└── MammothTransformError # pipeline transformation task failed
All exceptions expose a .message attribute. MammothAPIError additionally exposes .status_code (int | None) and .response_body (dict).
from mammoth import (
MammothClient,
MammothValidationError,
MammothAPIError,
)
client = MammothClient(api_key="...", api_secret="...", workspace_id=11)
client.set_project_id(42)
view = client.views.get(1039)
try:
# Validation fires before any network call
view.export.to_email(emails=[]) # raises MammothValidationError
except MammothValidationError as e:
print("Bad arguments:", e.message)
try:
client.views.get(99999) # raises MammothAPIError if not found
except MammothAPIError as e:
print(f"API error {e.status_code}: {e.response_body}")The SDK exposes typed resource APIs on the client for managing workspace entities. All public types are importable directly from mammoth.
from mammoth import ExternalKeyType
key = client.external_keys.create(
key_type=ExternalKeyType.ANTHROPIC,
key_name="My Claude key",
secure_key="sk-ant-...",
)
client.external_keys.delete(key["id"])# Add 50 GB storage
client.addons.add_storage(additional_storage_gb=50)
# Add 5 user seats
client.addons.add_users(user_count=5)
# Add a connector addon (single or bulk)
client.addons.add_connector(connector_id=42)
client.addons.add_connector(connector_ids=[42, 43])dashboard = client.dashboards.create(
intent="Show quarterly revenue by region and product",
source=[1039, 1040], # dataview IDs
enable_filters=True,
)from mammoth import (
AutomationTaskSpec, AutomationTaskType,
TaskDetailsSpec, DataRefreshConfig,
)
automation = client.automations.create(
name="Nightly refresh",
description="Pulls cloud data every night",
tasks=[
AutomationTaskSpec(
task_type=AutomationTaskType.RUN_DATA_RETRIEVAL,
details=TaskDetailsSpec(
ds_details=[DataRefreshConfig(ds_id=42)],
),
)
],
)from datetime import datetime
from mammoth import ScheduleCreateSpec, RruleSpec, RruleFrequency
schedule = client.automations.create_schedule(
spec=ScheduleCreateSpec(
rrule=RruleSpec(
frequency=RruleFrequency.DAILY,
start=datetime(2025, 1, 1, 6, 0),
),
)
)from mammoth import WorkspacePatchOp, WorkspacePatchPath, UserRolePatchOp, WorkspaceRoleType
# Rename the workspace
client.workspaces.update(patches=[
WorkspacePatchOp(op="replace", path=WorkspacePatchPath.NAME, value="My Workspace"),
])
# Change a user's role
client.workspaces.update_user(
user_id="user-uuid-here",
patches=[UserRolePatchOp(op="replace", path="role", value=WorkspaceRoleType.WORKSPACE_ADMIN)],
)# Create a PostgreSQL connection (config shape varies by connector_key)
conn = client.connectors.create_connection(
connector_key="postgres",
config={
"hostname": "db.example.com",
"port": 5432,
"database": "analytics",
"username": "user",
"password": "pass",
},
)
# Create a data source config (query-based for DB connectors)
ds_config = client.connectors.create_ds_config(
connector_key="postgres",
connection_key=conn["connection_key"],
query="SELECT * FROM sales WHERE year = 2024",
validate=True,
)mammoth-cli drives the same platform from your terminal — for people at a shell and for autonomous agents. In a terminal it prints readable tables; when output is piped it emits a stable JSON envelope with documented exit and error codes, so agents and CI need no special flags.
uv tool install mammoth-cli # or: pipx install mammoth-cli / pip install mammoth-cli
mammoth auth login -w 4 # log in once; the secret goes to your OS keyring
mammoth project list # a table in a terminal, JSON when pipedIt covers uploads, pipeline transformations, dashboards, automation, and administration, and it installs a bundled agent skill for Claude Code, Codex, and Cursor (mammoth skill install). See the mammoth-cli directory and its documentation.
The SDK includes a companion MCP (Model Context Protocol) server that lets AI assistants interact with Mammoth directly. Install it separately:
pip install mammoth-mcpSee the mammoth-mcp directory for configuration and usage details.
Maintainer release process (SDK + CLI, PyPI + GitHub releases, tag-triggered CI and the local fallback) is documented in RELEASING.md.