anaplan-orm 0.1.0

An Object-Relational Mapper (ORM) for the Anaplan API using Pydantic.

anaplan-orm

A lightweight Python 3 library that abstracts the Anaplan API into an Object-Relational Mapper (ORM).

Current Status

🚀 Active Beta 🚀 Core data transformation, parsing engine, and Anaplan chunked API client are complete.

🌟 Features

• Pydantic Data Ingestion: Validates and maps Python objects to Anaplan models effortlessly.
• Enterprise Security: Supports standard Basic Authentication and Anaplan's proprietary RSA-SHA512 Certificate-based Authentication (mTLS).
• Resilient Networking: Built-in exponential backoff, automated retries to protect against dropped packets, and mid-flight authentication token refreshing for massive, long-running pipelines.
• Massive Payloads: Automatically handles chunked file uploads for multi-megabyte/gigabyte datasets without memory crashes.
• Smart Polling: Asynchronous process execution with configurable, patient polling for long-running database transactions.

---

🔐 Authentication

anaplan-orm uses a decoupled authentication strategy, allowing you to easily swap between development and production security standards.

1. Basic Authentication

Ideal for development and sandbox testing.

from anaplan_orm.client import AnaplanClient
from anaplan_orm.authenticator import BasicAuthenticator

auth = BasicAuthenticator("[e-mail]", "your_password")
client = AnaplanClient(authenticator=auth)

2. Certificate-Based Authentication (Enterprise Standard)

For production environments, Anaplan requires a custom RSA-SHA512 signature. The CertificateAuthenticator handles this cryptographic handshake automatically.

Note: The library expects a .pem file containing both your private key and public certificate. If your enterprise issues a .p12 keystore, you can extract it using your terminal:

openssl pkcs12 -in keystore.p12 -out certificate.pem

from anaplan_orm.client import AnaplanClient
from anaplan_orm.authenticator import CertificateAuthenticator

# 1. Initialize the Certificate Authenticator
auth = CertificateAuthenticator(
    cert_path="path/to/your/certificate.pem",
    # Omit if your private key is unencrypted
    cert_password="your_secure_password", 
    # Set to False if you need to bypass a corporate proxy
    verify_ssl=True
)

# 2. Inject it into the Anaplan Client
client = AnaplanClient(authenticator=auth)

# 3. Execute a request
status = client.ping()

---

Quick Start: XML Parsing & Data Upload

The anaplan-orm is designed to take raw XML strings (e.g., from MuleSoft or data pipeline payloads), validate them into Python objects, and stream them directly into Anaplan.

1. Define Your Model

Map your Anaplan target columns to Python using Pydantic fields. The alias parameter bridges the gap between external uppercase XML tags and internal Python snake_case variables.

from pydantic import Field
from anaplan_orm.models import AnaplanModel

class Developer(AnaplanModel):
    dev_id: int = Field(alias="DEV_ID")
    dev_name: str = Field(alias="DEV_NAME")
    dev_age: int = Field(alias="DEV_AGE")
    dev_location: str = Field(alias="DEV_LOCATION")

2. Parse, Serialize, and Upload

Use the XMLStringParser to ingest your XML string payload, then use the AnaplanClient to stream the chunked data to Anaplan.

from anaplan_orm.parsers import XMLStringParser
from anaplan_orm.client import AnaplanClient, BasicAuthenticator

# 1. Your incoming XML string payload
xml_string = """
<AnaplanExport>
    <Row>
        <DEV_ID>1001</DEV_ID>
        <DEV_NAME>Ada Lovelace</DEV_NAME>
        <DEV_AGE>36</DEV_AGE>
        <DEV_LOCATION>London</DEV_LOCATION>
    </Row>
</AnaplanExport>
"""

def run_pipeline():
    # 2. Parse and Validate the data using the ORM
    parser = XMLStringParser()
    developers = Developer.from_payload(payload=xml_string, parser=parser)
    
    # 3. Serialize to Anaplan-ready CSV (using a Pipe separator)
    csv_data = Developer.to_csv(developers, separator="|")

    # 4. Authenticate with Anaplan
    auth = BasicAuthenticator(
        email="ANAPLAN_EMAIL", 
        pwd="ANAPLAN_PASSWORD"
    )

    client = AnaplanClient(authenticator=auth)

    # 5. Stream the file chunks safely
    client.upload_file_chunked(
        workspace_id="YOUR_WORKSPACE_ID", 
        model_id="YOUR_MODEL_ID", 
        file_id="YOUR_FILE_ID", 
        csv_data=csv_data,
        chunk_size_mb=10
    )

    # 6. Execute the Import Process
    task_id = client.execute_process(
        workspace_id="YOUR_WORKSPACE_ID", 
        model_id="YOUR_MODEL_ID", 
        process_id="YOUR_PROCESS_ID"
    )
    
    # 7. Actively poll the database for success/failure
    status = client.wait_for_process_completion(
        workspace_id="YOUR_WORKSPACE_ID", 
        model_id="YOUR_MODEL_ID", 
        process_id="YOUR_PROCESS_ID", 
        task_id=task_id
    )

if __name__ == "__main__":
    run_pipeline()

---

Advanced: Deeply Nested XML Extraction

If your XML payload is deeply nested or relies heavily on attributes (common with SOAP APIs), you can use Pydantic's json_schema_extra to define native XPath 1.0 mappings. The parser will automatically evaluate the XPath, extract both text nodes and attributes, and map them to your Anaplan aliases.

from pydantic import Field
from anaplan_orm.models import AnaplanModel

class NestedXMLDeveloper(AnaplanModel):
    # Use '@' to extract attributes
    # Use '/' to navigate nested text nodes
    dev_id: int = Field(
        alias="DEV_ID", 
        json_schema_extra={"path": "./EmployeeDetails/@empId"} 
    )
    dev_name: str = Field(
        alias="DEV_NAME", 
        json_schema_extra={"path": "./EmployeeDetails/Profile/FullName"} 
    )

To extract the repeating rows from the document, simply pass the base XPath expression to the parser using the data_key argument:

from anaplan_orm.parsers import XMLStringParser

# The parser will find every <Employee> node, and apply your XPath mappings to it!
developers = NestedXMLDeveloper.from_payload(
    payload=raw_xml_string, 
    parser=XMLStringParser(), 
    data_key=".//Employee" 
)

Below you can see the XML example used for the above example

<?xml version="1.0" encoding="UTF-8"?>
<EnterpriseExport status="success" timestamp="2026-03-16T08:00:00Z">
    <EmployeeRecords>
        <Employee status="active">
            <EmployeeDetails empId="1001">
                <Profile>
                    <FullName>Ada Lovelace</FullName>
                    <Age>36</Age>
                </Profile>
            </EmployeeDetails>
            <Office>
                <City>London</City>
                <Region>EMEA</Region>
            </Office>
        </Employee>
        <Employee status="active">
            <EmployeeDetails empId="1002">
                <Profile>
                    <FullName>Grace Hopper</FullName>
                    <Age>85</Age>
                </Profile>
            </EmployeeDetails>
            <Office>
                <City>New York</City>
                <Region>NAMER</Region>
            </Office>
        </Employee>
    </EmployeeRecords>
</EnterpriseExport>

---

Quick Start: JSON Parsing (REST APIs & Files)

For modern web integrations or local file processing, anaplan-orm provides a native JSONParser. It gracefully handles both flat JSON arrays and nested API responses by allowing you to pass targeted extraction keys directly through your Pydantic model.

1. Define Your Model

from pydantic import Field
from anaplan_orm.models import AnaplanModel

class Employee(AnaplanModel):
    emp_id: int = Field(alias="id")
    email: str = Field(alias="emailAddress")
    department: str = Field(alias="dept")

2. Parse and Upload

If your JSON data is nested inside a metadata wrapper: I.E:

{"status": "success", "data": [...] }

Simply pass the data_key argument to the from_payload method. The ORM will safely drill down, extract the records, and inflate your models.

import httpx
from anaplan_orm.parsers import JSONParser

# 1. Fetch JSON from an external REST API (or read a local .json file)
api_response = httpx.get("https://api.mycompany.com/v1/employees").text

# 2. Parse the JSON string (drilling into the "data" array)
parser = JSONParser()
employees = Employee.from_payload(
    payload=api_response, 
    parser=parser, 
    # The ORM passes this key directly to the parser.
    data_key="data" 
)

# 3. Convert to Anaplan CSV and Upload
csv_data = Employee.to_csv(employees)
client.upload_file_chunked(WORKSPACE_ID, MODEL_ID, FILE_ID, csv_data)

---

Advanced: Deeply Nested JSON Extraction

If your API returns a deeply nested JSON response, you do not need to write custom flattening loops. Simply use Pydantic's json_schema_extra to define a JMESPath mapping. The ORM will automatically traverse the JSON tree, extract the value, and assign it to the correct Anaplan column (alias).

from pydantic import Field
from anaplan_orm.models import AnaplanModel

class NestedEmployee(AnaplanModel):
    # 'alias' is the Anaplan CSV column's name. 
    # 'path' is where to find the data in the JSON.
    emp_id: int = Field(
        alias="DEV_ID", 
        json_schema_extra={"path": "employeeDetails.empId"}
    )
    city: str = Field(
        alias="LOCATION", 
        json_schema_extra={"path": "office.address.city"}
    )

---

Quick Start: SQL Databases (Relational Data to Anaplan)

If your source data lives in a relational database (Snowflake, PostgreSQL, SQL Server), anaplan-orm provides an SQLCursorParser. This allows you to stream live database queries directly into Pydantic models without ever saving a CSV to disk.

1. Execute your query and pass the cursor

The parser accepts any standard DB-API 2.0 cursor object, dynamically extracts the column headers, and maps them to your model aliases.

import psycopg2 # Or sqlite3, snowflake.connector, etc.
from anaplan_orm.parsers import SQLCursorParser

# 1. Connect to your database and execute a query
conn = psycopg2.connect("dbname=enterprise user=admin password=secret")
cursor = conn.cursor()
cursor.execute("SELECT emp_id AS id, email_address, department FROM employees WHERE active = true")

# 2. Pass the active cursor directly into the ORM
employees = Employee.from_payload(
    payload=cursor, 
    parser=SQLCursorParser()
)

# 3. Convert to Anaplan CSV and Upload
csv_data = Employee.to_csv(employees)
client.upload_file_chunked(WORKSPACE_ID, MODEL_ID, FILE_ID, csv_data)

conn.close()

---

⬇️ Extracting Data (Outbound Pipeline)

anaplan-orm supports two distinct architectural patterns for extracting data, depending on your pipeline's requirements.

Option A: Pure Extract & Load (ELT)

If your goal is simply to move data from Anaplan to a Data Lake (like AWS S3) to ingest later. You can stream the raw CSV text directly.

# 1. Trigger the export and wait for completion
task_id = client.execute_export(WORKSPACE_ID, MODEL_ID, EXPORT_ID)
client.wait_for_export_completion(WORKSPACE_ID, MODEL_ID, EXPORT_ID, task_id)

# 2. Download the raw CSV string in chunks
raw_csv_string = client.download_file_chunked(WORKSPACE_ID, MODEL_ID, EXPORT_ID)

# 3. Write directly to a file (or upload to AWS S3 using boto3)
with open("anaplan_export.csv", "w", encoding="utf-8") as f:
    f.write(raw_csv_string)

Option B: In-Flight Processing (The True ORM)

If you need to validate data types, perform cross-column mathematical transformations, or mask sensitive PII before routing the data to another microservice, you can seamlessly inflate the CSV into strongly-typed Pydantic models.

from pydantic import BaseModel, Field, ValidationError
from anaplan_orm.parsers import CSVStringParser

# Define your data contract
class FinancialRow(BaseModel):
    cost_center: str = Field(alias="Cost Center")
    outlook_eur: float = Field(alias="Outlook in Local Currency")

# 1. Parse the raw CSV string into a list of dictionaries
parsed_rows = CSVStringParser.parse(raw_csv_string)

# 2. Inflate and validate the Pydantic models
valid_models = []
for row in parsed_rows:
    try:
        valid_models.append(FinancialRow(**row))
    except ValidationError as e:
        print(f"Quarantined invalid row: {e}")

# You now have a list of mathematically safe Python objects for an easy transformations 
for model in valid_models:
    model.outlook_eur = model.outlook_eur * 1.05

🤝 Contributing to anaplan-orm

We welcome contributions! To maintain enterprise-grade code quality, this project uses strict formatting, linting, and testing pipelines.

Prerequisites

• Python 3.10+
• Poetry (Dependency management)

1. Local Setup

Clone the repository and install all dependencies (including the dev group tools like Pytest and Ruff):

git clone https://github.com/valdal14/anaplan-orm.git
cd anaplan-orm
poetry install

2. Formatting & Linting (Ruff)

This project enforces strict PEP 8 compliance using Ruff. Before submitting any code, you must format and lint your changes. If you do not run these commands, the GitHub Actions CI pipeline will fail your Pull Request.

Run the formatter to automatically fix spacing, quotes, and line breaks:

poetry run python3 -m ruff format .

Run the linter to catch unused imports, bad variables, and logical style issues:

poetry run python3 -m ruff check --fix .

(Tip: I highly recommend installing the Ruff extension in your IDE and setting it to "Format on Save").

3. Running Tests (Pytest)

Every feature and bug fix must be covered by unit tests. The test suite heavily utilizes Python's unittest.mock to simulate Anaplan network responses without requiring live API credentials.

Run the entire test suite:

poetry run python3 -m pytest

4. The Pull Request Workflow

1 - Create a feature branch (e.g., feature/ORM-123-new-parser).
2 - Write your code and your tests.
3 - Run Ruff (format and check) and Pytest.
4 - Push your branch to GitHub and open a Pull Request against main.
4 - Wait for the automated CI pipeline to verify your build before merging.