Metadata-Version: 2.3
Name: anaplan-orm
Version: 0.2.0
Summary: An Object-Relational Mapper (ORM) for the Anaplan API using Pydantic.
License: MIT
Keywords: anaplan,orm,pydantic,data-engineering,api
Author: Valerio DAlessio
Author-email: valdal14@gmail.com
Requires-Python: >=3.10
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Dist: cryptography (>=46.0.5,<47.0.0)
Requires-Dist: httpx (>=0.28.1,<0.29.0)
Requires-Dist: jmespath (>=1.1.0,<2.0.0)
Requires-Dist: lxml (>=6.0.2,<7.0.0)
Requires-Dist: pydantic (>=2.12.5,<3.0.0)
Project-URL: Repository, https://github.com/valdal14/anaplan-orm
Description-Content-Type: text/markdown

# anaplan-orm

![CI Pipeline](https://github.com/valdal14/anaplan-orm/actions/workflows/ci.yml/badge.svg)
![Python](https://img.shields.io/badge/Python-3.10%2B-blue?style=flat&logo=python)
![License](https://img.shields.io/badge/License-MIT-green)

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

## Current Status
🚀 **Active Beta (v0.2.0)** 🚀
Core data transformation, parsing engine, chunked Anaplan API client, and custom strict-type validators are complete.

## 🌟 Features
* **Pydantic Data Ingestion:** Validates and maps Python objects to Anaplan models effortlessly.
* **Custom Anaplan Types:** Built-in `AnaplanDate` and `AnaplanBoolean` automatically sanitize and strictly format Python native types (like `datetime.date` or `True/False`) into the exact string formats required by Anaplan's API.
* **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.

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

auth = BasicAuthenticator("your_email@company.com", "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:

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

```python
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. 

```python
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.

```python
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.

```python
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:

```python
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
<?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
```python
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: 

```json
{"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.

```python
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](https://jmespath.org/) mapping. The ORM will automatically traverse the JSON tree, extract the value, and assign it to the correct Anaplan column (`alias`).

```python
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`.

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

--- 

## 🧰 Custom Anaplan Data Types (Sanitization)

The Anaplan API is notoriously strict about data formats during imports. A standard Python `datetime` object or a boolean `True` will cause a silent rejection if Anaplan expects `"YYYY-MM-DD"` or `"true"`. 

`anaplan-orm` provides custom Pydantic types that act as an automatic translation layer. They accept highly flexible Python inputs and strictly serialize them for Anaplan.

```python
from datetime import date
from pydantic import BaseModel
from anaplan_orm.types import AnaplanDate, AnaplanBoolean

class EmployeeRow(BaseModel):
    emp_id: int
    is_active: AnaplanBoolean
    start_date: AnaplanDate

# The ORM is highly forgiving with inputs:
row_1 = EmployeeRow(emp_id=1, is_active="Yes", start_date="2026-03-19")
row_2 = EmployeeRow(emp_id=2, is_active=True, start_date=date(2026, 4, 12))

# But it strictly serializes for the Anaplan API:
print(row_1.model_dump()) 
# {'emp_id': 1, 'is_active': 'true', 'start_date': '2026-03-19'}

print(row_2.model_dump())
# {'emp_id': 2, 'is_active': 'true', 'start_date': '2026-04-12'}
```

---

## ⬇️ 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.

```python
# 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.

```python
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):

```bash
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:

```bash
poetry run python3 -m ruff format .
```

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

```bash
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:

```bash
poetry run python3 -m pytest
```

### 4. The Pull Request Workflow

```bash
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.
```
