Metadata-Version: 2.4
Name: pam-python
Version: 0.2.4
Summary: Pam Python Library
Author-email: Narongrit Kanhanoi <narongrit@pams.ai>
Project-URL: Homepage, https://github.com/heart/pam-python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Operating System :: OS Independent
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Customer Service
Classifier: Topic :: Software Development :: Libraries
Requires-Python: <3.13,>=3.12
Description-Content-Type: text/markdown
License-File: LICENSE.txt
Requires-Dist: Flask>=3.0.2
Requires-Dist: requests>=2.32.3
Requires-Dist: PyYAML>=6.0.2
Dynamic: license-file

# pam-python-data-plugin-framework

This repository provides the `pam` CLI and runtime framework to build Data Plugin services for PAM Real CDP. It generates a ready-to-run project, standardizes service lifecycle, and handles common tasks like input handling, temp storage, uploads, and service monitoring.

This README is a practical, step-by-step guide you can follow to create and run a real service.

**What you get**

- CLI to initialize a project and scaffold services
- Service lifecycle contract (start, data input, upload, exit)
- Temp file and SQLite helpers
- Managed Report Store V2 with typed Table, Metric, and ApexCharts-compatible builders
- Named `ResultBatchUploader` streams for bounded DataFrame uploads
- A monitoring loop for service timeouts and periodic cleanup

---

**Table of Contents**

1. Prerequisites
2. Install
3. Initialize a Project
4. Create a Service
5. Understand the Lifecycle
6. Using Temp Files Correctly
7. Running the Server
8. Testing a Service
9. Configuration
10. Project Structure
11. Troubleshooting

---

**Prerequisites**

- Python 3.12
- `uv` (recommended) or `pip`

---

**Create a uv project**

Install PAM inside each project instead of installing the CLI globally:

```bash
mkdir my_data_plugin
cd my_data_plugin
uv init --python 3.12
uv add pam-python
uv run pam init
```

`pam init` recognizes a fresh uv project, keeps its uv configuration, and adds
the PAM scaffold. It refuses projects whose generated source files have already
been edited.

Pip remains supported for new projects that do not use uv:

```bash
mkdir my_data_plugin
cd my_data_plugin
python3 -m venv venv
source venv/bin/activate
python -m pip install pam-python
pam init
```

When run in an empty directory, `pam init` still asks whether to create a uv or
pip project. PAM never migrates an existing application.

---

**Create a Service**

```bash
uv run pam new service rfm_segment
```

The generator asks you to select Polars (default) or Pandas. Polars services use
lazy Parquet processing. Pandas services also receive Parquet and install
`pandas` plus `pyarrow`. The selected direct dependencies are added to the
project automatically.

This creates a new folder (e.g. `rfm_segment/`) with:

- a service class (`RfmSegmentSvc.py`)
- `functions.py` for your logic
- `service.yaml` for registration
- a test file

---

**Understand the Lifecycle**
The runtime calls your service in two main phases.

1. `on_start`

- Called once at the beginning
- Read parameters from `self.request.runtime_parameters`
- Should return quickly (start a thread for long work)

2. `on_data_input`

- Called when CDP sends input files
- `req.input_files` is the ordered positional data-source contract configured in PAM
- `req.file_format` is the typed format sent by PAM
- Should also return quickly (use a thread if needed)

When your service is done:

- Upload ordinary CDP result rows through `ResultBatchUploader`
- Publish managed reports through `self.reports`; do not build report JSON manually
- Call `self._exit()` to signal completion

---

**Using Temp Files Correctly**
Temp storage is managed by the framework. Do not delete temp files manually.

Standard helpers:

- `TempfileUtils.get_temp_path_for_service(self, self.service_name)`
- `TempfileUtils.get_temp_file_name_for_service(self, self.service_name, prefix, extension)`

Notes:

- `get_temp_path_for_service(...)` returns a directory path without a trailing slash.
- The temp path includes date/service/token in this structure:
  `TEMP_DATASOURCE_PATH/YYYY_MM_DD/<service>/<token>`

---

**Uploading Results in Batches**
Polars `LazyFrame`/`DataFrame` and Pandas `DataFrame` results all pass through
`ResultBatchUploader`. It streams lazy results in bounded chunks, buffers incomplete
batches across calls, writes complete batches as CSV, and uploads them through PAM.
The framework does not prescribe page-local/global computation or intermediate
storage; submit rows when your business result is ready.

Recommended usage:

```python
from pam.result_batch_uploader import ResultBatchUploader

batch_uploader = ResultBatchUploader(self, batch_size=50000)
batch_uploader.upload(df, name="main")
batch_uploader.flush()
batch_uploader.wait_for_uploader()
status = batch_uploader.get_status()
```

Notes:

- `request.runtime_parameters["batch_size"]` overrides the constructor default when valid.
- `name` separates result streams and enforces one stable column schema per stream.
- `options` may be passed to `upload(...)` and are forwarded to `_upload_result`.
- Uploads run in a bounded background queue so processing can continue while PAM receives CSV batches.
- `flush()` closes input and queues the final remainder; it does not wait for network completion.
- Always call `wait_for_uploader()` after `flush()` and before `_exit()`.
- Failed uploads retry with exponential backoff, then log/record the failed batch and continue.
- `get_status()` includes buffered, uploaded, failed, and retried counts per stream.
- Polars `LazyFrame` is supported without collecting the complete result at once.

---

**Managed Reports (Report Store V2)**

Managed reports are static facts stored in framework-owned SQLite. Plugin code
selects a typed builder, writes source-derived `DAY`, `MONTH`, or `NONE` facts, and
publishes only after the full calculation succeeds.

```python
fresh = self.request.get_runtime_bool("fresh", default=False)
store = (
    self.reports.create("reports.sqlite")
    if fresh
    else self.reports.open("reports.sqlite")
)

# Choose the builder agreed in the customer requirement.
# See generated REPORTS.md for every supported declaration and write API.
```

Important rules:

- Use only typed builders documented in `REPORTS.md`; there is no raw-report API.
- `open()` resumes the remote file or creates it only after an authoritative 404.
- Network/5xx download failures never create a replacement store.
- Use `create()` only for a human-requested fresh restart.
- Delete a partition, write its recalculated facts, then call `store.publish()` once.
- Low-level `_request_sqlite`/`_upload_sqlite` remain for custom plugin state, not reports.

---

**Running the Server**
The generated `main.py` runs the Flask server.

```bash
uv run python main.py  # uv project
python main.py         # pip project
```

By default it binds to `0.0.0.0:8000`. You can override with:

```bash
export SERVER_HOST=0.0.0.0
export SERVER_PORT=8000
```

---

**Testing a Service**
For uv projects, use one cross-platform command:

```bash
uv run python -m unittest discover -s rfm_segment -p "test_*.py"
```

Pip projects include shell-specific scripts:

macOS/Linux:

```bash
./run_unit_test.sh rfm_segment
```

Windows PowerShell:

```powershell
.\run_unit_test.ps1 rfm_segment
```

Windows Command Prompt:

```bat
run_unit_test.bat rfm_segment
```

Place custom tests in the service folder and name them `test_<service>.py`.

---

**Configuration**
Environment variables you can set:

- `SERVER_HOST`
- `SERVER_PORT`
- `TEMP_BASE_PATH` (default `/app/data`)
- `TEMP_DATASOURCE_PATH` (default `/app/data/data_sources`)
- `TEMP_CLEAN_DAYS` (default `10`)
- `TEMP_CLEAN_INTERVAL_HOURS` (default `6`, set empty to disable periodic cleanup)

---

**Project Structure**
After `pam init` and one service:

```
.
├── main.py
├── AGENT.md
├── REPORTS.md
├── Dockerfile
├── pyproject.toml
├── uv.lock
├── .python-version
├── rfm_segment/
│   ├── RfmSegmentSvc.py
│   ├── functions.py
│   ├── service.yaml
│   └── test_rfm_segment.py
└── .pam-project
```

---

**Troubleshooting**

- If `pam` is missing, verify the uv tool installation or activate the pip virtualenv.
- If uv initialization fails, verify `uv` is on `PATH` and package indexes are reachable.
- If `pam new service` fails, confirm the service name is provided.
- If temp cleanup is too frequent or too slow, adjust `TEMP_CLEAN_INTERVAL_HOURS` and `TEMP_CLEAN_DAYS`.

---

**Next Steps**

- Implement your logic in `functions.py`.
- Wire it into `on_start` and `on_data_input` in your service class.
- Use the temp utilities to write intermediate files.
- Keep Polars transformations lazy and submit final results through `ResultBatchUploader`.
- Read `REPORTS.md` before implementing a managed report.
