mammoth-io 0.3.1

Python SDK for Mammoth Analytics platform

mammoth-io

Python SDK for the Mammoth Analytics platform. Build data pipelines, apply transformations, and export results -- all from Python.

Installation

pip install mammoth-io

Requires Python 3.10+.

Quick Start

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"])

Views & Transformations

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)

Filter Rows

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,
)

Set Values (Conditional Labeling)

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"),
    ],
)

Math

# String expressions are parsed automatically
view.math("Price * Quantity", new_column="Revenue")
view.math("(Price + Tax) * 1.1", new_column="Grand Total")

Join

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"],
)

Pivot (Group By / Aggregate)

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"),
    ],
)

Window Functions

from mammoth import WindowFunction, SortDirection

view.window(
    function=WindowFunction.ROW_NUMBER,
    new_column="Rank",
    partition_by=["Region"],
    order_by=[["Sales", SortDirection.DESC]],
)

Text Operations

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")],
)

Date Operations

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")

Column Operations

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")])

Row Operations

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 and SQL

# 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")

Pipeline Management

# 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": [...]}})

All Transformation Methods

Method	Description
filter_rows()	Filter rows by condition
set_values()	Label/insert values with conditional logic
math()	Arithmetic expressions
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

Parameter Spec Dataclasses

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()

Conditions

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)

Supported Operators

Operator	Description
EQ, NE	Equal, not equal
GT, GTE, LT, LTE	Comparison
IN_LIST, NOT_IN_LIST	Value in/not in list
CONTAINS, NOT_CONTAINS	Text contains/not 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

File Upload

# 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)

Exports

Download as CSV

# 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")

Export to S3

# From a View object
result = view.export.to_s3(file_name="monthly_report.csv")

# From client with a known dataview ID
result = client.exports.to_s3(dataview_id=1039, file="monthly_report.csv")

Export to Database

# 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",
)

Branch Out (Export to Another Dataset)

# From a View object
view.export.to_dataset(dest_dataset_id=500)

# Or using the shorthand
view.branch_out(dest_dataset_id=500)

Other Export Targets

view.export.to_bigquery(...)
view.export.to_redshift(...)
view.export.to_elasticsearch(...)
view.export.to_ftp(host="ftp.example.com", path="/exports/data.csv", username="user", password="pass")
view.export.to_sftp(host="sftp.example.com", path="/exports/data.csv", username="user", password="pass")
view.export.to_email(recipients=["[e-mail]"])

MCP Server

The SDK includes a companion MCP (Model Context Protocol) server that lets AI assistants interact with Mammoth directly. Install it separately:

pip install mammoth-mcp

See the mammoth-mcp directory for configuration and usage details.