Metadata-Version: 2.4
Name: soigia
Version: 0.1.24
Summary: Python SDK for Soi Gia
Author-email: Soi Gia <sojgja@gmail.com>
License-Expression: BSD-3-Clause
Keywords: soigia,sdk,datatable,pipeline,streamlit
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pandas
Requires-Dist: jinja2
Requires-Dist: loguru
Requires-Dist: Faker
Requires-Dist: pyarrow
Requires-Dist: pyyaml
Requires-Dist: tabulate
Requires-Dist: rich
Requires-Dist: sqlalchemy
Provides-Extra: dev
Requires-Dist: coverage; extra == "dev"
Requires-Dist: hypothesis; extra == "dev"
Requires-Dist: mypy; extra == "dev"
Requires-Dist: pandas-stubs; extra == "dev"
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Dynamic: license-file

# soigia

> **Important: `fake_vars`**
>
> `soigia.fake_vars` is a test-only escape hatch for advanced runtime cases where you must
> inject values into a running function's local variables.
>
> Use it only when you have no safer option.
>
> Important constraints:
>
> - CPython-specific: it relies on frame internals and `ctypes`
> - test/debug only: do not treat it as a stable production API
> - it only updates locals that already exist in the target frame
> - missing names are intentionally ignored unless your test asserts otherwise

### `fake_vars` example

Use `fake_vars` when you want to run a function normally, but replace a few local variables while it is running.

#### Pytest example

```python
from soigia import fake_vars


class OrderService:
    def __init__(self):
        self.qty = 1

    def process(self):
        subtotal = 100
        tax = 20
        total = subtotal + tax
        return subtotal, tax, total


def test_order_method_with_fake_locals():
    service = OrderService()

    result = fake_vars.set_vars(service.process, subtotal=300, tax=0)

    assert result.subtotal == 300
    assert result.tax == 0
    assert result.total == 300
```

#### unittest example

```python
import unittest
from soigia import fake_vars


class OrderService:
    def __init__(self):
        self.audit = []

    def process(self):
        status = "PENDING"
        self.audit.append(status)
        return status


class TestOrderService(unittest.TestCase):
    def test_order_method_with_fake_locals(self):
        service = OrderService()

        result = fake_vars.set_vars(service.process, status="OK")

        self.assertEqual(result.status, "OK")
        assert service.audit == ["OK"]

    def test_order_function_with_fake_locals(self):
        audit = []

        def process():
            status = "PENDING"
            audit.append(status)
            return status

        result = fake_vars.set_vars(process, status="OK")

        self.assertEqual(result.status, "OK")
        self.assertEqual(audit, ["OK"])
```

```python
from soigia import fake_vars


class OrderService:
    def __init__(self):
        self.qty = 1

    def compute_v1(self):
        a = 1
        b = 1
        c = self.qty + a + b
        return c

    def compute_v2(self):
        a = 1
        a = 2
        b = 1
        c = a + b
        return c

    def compute_v3(self):
        items = ["A", "B", "C"]
        count = len(items)
        return items, count


service = OrderService()

# Override locals that already exist in the frame.
# By default, the override stays active throughout the function.
result_v1 = fake_vars.set_vars(service.compute_v1, a=2, b=3)
print(result_v1)

# Apply transforms to locals.
result_v1_if_equal = fake_vars.set_vars(
    service.compute_v1,
    local={
        "a": fake_vars.if_equal(1, 10),
        "b": fake_vars.if_equal(1, 20),
    },
)
print(result_v1_if_equal)

# Later assignments in the function do not win when lock=True.
result_v2 = fake_vars.set_vars(service.compute_v2, a=20, b=10)
print(result_v2)

# `head` trims list-like locals.
result_v3 = fake_vars.set_vars(service.compute_v3, local={"items": fake_vars.head(1)})
print(result_v3)

# Disable locking if you only want the first injection.
result_v4 = fake_vars.set_vars(service.compute_v2, a=20, b=10, lock=False)
print(result_v4)
```

This is useful when:

- you want to swap one exact local value for a test
- you want to keep the rest of the function logic unchanged
- you want to trim a list-like local value to a smaller sample

Quick rules:

1. You can omit `local` when you only want the original function run
2. Use positional arguments for the function call itself
3. Use `local` for the local variables you want to override, fully or partially
4. You can mix positional arguments with `local` when that is more convenient
5. Use `lock=False` when you only want the first injection to stick
6. Use a transform when you do not want a plain value

Note: `local` is optional on `set_vars`, but the target function still needs enough
arguments for its own signature. If the target takes no required arguments, you can call
`set_vars(target)` directly.

Common transforms:

- `fake_vars.head(1)`: keep only the first row/item of a local value
- `fake_vars.if_equal(expected, replacement)`: replace only when the current value matches `expected`

The helper returns a `SimpleNamespace` snapshot of the target function's locals at return time, and the snapshot is deep-copied for mutable values so it is safer to inspect during debug and tests.

`soigia` is a Python SDK built around four practical workspaces:

- `DataTable` for dataframe-style work with a friendlier API
- `Pipeline` for repeatable business workflows
- `ORM` for model-driven business data with backend adapters
- `UI` for lightweight internal screens on top of Streamlit

## Philosophy

Soi Gia keeps the codebase practical and explicit:

- domain rules stay close to the business language
- public APIs stay small and predictable
- shared concepts live in `common`, not scattered across modules
- each bounded context owns its own entities, services, policies, and value objects
- runtime code does not depend on docs, and docs do not depend on external references
- examples in this README are meant to be copy-paste friendly and match the real API

## Quick Index

- [DataTable](#1-datatable)
- [Pipeline](#2-pipeline)
- [ORM](#3-orm)
- [UI](#4-ui)
- [Fake Data](#fake-data)
- [Odoo Client](#odoo-client)
- [Domain](#5-domain) - module-based bounded contexts for business rules

This README focuses on those four areas plus the domain layer.

## Odoo Client

`soigia.odoo_client` is a practical Python client for working with Odoo 19 from:

- Django applications
- FastAPI or Flask APIs
- background jobs
- Airflow DAGs
- console scripts
- cron jobs
- one-off migration scripts

The design goal is simple:

- create one client once
- reuse it everywhere
- keep the public API readable
- support both low-level Odoo access and a higher-level ORM-like style
- support batch work directly from `pandas.DataFrame`

### Install

```bash
pip install soigia
```

If you use Django:

```bash
pip install "soigia[django]"
```

### Import

```python
from soigia.odoo_client import OdooClient
```

### Odoo Version

This client is built for Odoo 19 and uses:

- `JSON-2` when appropriate
- `XML-RPC` as a compatibility and generic fallback

This means you can use modern Odoo 19 APIs while still keeping broad method coverage.

### Quick Start

Create the client once, then reuse it:

```python
from soigia.odoo_client import OdooClient

odoo = OdooClient(
    "http://127.0.0.1:8070",
    "odoo",
    login="admin",
    password="admin",
)
```

### Local Docker Stack

The repo includes a self-bootstrapping Docker stack in [`compose.yaml`](./compose.yaml).
It starts Odoo and PostgreSQL, initializes the `base` module on first run, and is ready for `soigia.odoo_client`.

```bash
cp .env.example .env
docker compose up -d
```

This starts:

- `soigia-sdk-odoo` on `http://127.0.0.1:8069`
- `soigia-sdk-db` as the PostgreSQL service

The sample runtime also recognizes `soigia-sdk-odoo:8069` from inside Docker.

Then get a model:

```python
Partner = odoo.env["res.partner"]
```

Or define a stricter model with selected fields and batch keys:

```python
Partner = odoo.env.model(
    "res.partner",
    fields=["id", "name", "email", "phone", "is_company"],
    key_fields=["email"],
)
```

### Connection Styles

#### Direct configuration

```python
odoo = OdooClient(
    "http://127.0.0.1:8070",
    "odoo",
    login="admin",
    password="admin",
)
```

#### From environment variables

Supported variables:

- `ODOO_URL`
- `ODOO_DB`
- `ODOO_LOGIN`
- `ODOO_PASSWORD`
- `ODOO_API_KEY`

Example:

```python
odoo = OdooClient.from_env()
```

Example shell:

```bash
export ODOO_URL=http://127.0.0.1:8070
export ODOO_DB=odoo
export ODOO_LOGIN=admin
export ODOO_PASSWORD=admin
```

#### From Django settings

Dictionary style:

```python
ODOO = {
    "URL": "http://127.0.0.1:8070",
    "DB": "odoo",
    "LOGIN": "admin",
    "PASSWORD": "admin",
    "RETRY_COUNT": 2,
    "RETRY_DELAY": 0.5,
}
```

Then:

```python
odoo = OdooClient.from_django_settings()
```

Attribute style is also supported:

```python
ODOO_URL = "http://127.0.0.1:8070"
ODOO_DB = "odoo"
ODOO_LOGIN = "admin"
ODOO_PASSWORD = "admin"
```

### Recommended Usage Pattern

The recommended pattern is:

1. create `OdooClient` once
2. keep it in your app container, service, or module
3. create model objects from `odoo.env`
4. query and write through the model

Example:

```python
from soigia.odoo_client import OdooClient

odoo = OdooClient.from_env()

Partner = odoo.env.model(
    "res.partner",
    fields=["id", "name", "email", "phone", "is_company"],
    key_fields=["email"],
)
```

### ORM-Like Query Style

#### Get one record

```python
partner = Partner.objects.get(id=1)
```

#### Filter records

```python
partners = Partner.objects.filter(is_company=True).all(limit=10)
```

#### First record

```python
first_partner = Partner.objects.filter(name__icontains="soigia").first()
```

#### Count and exists

```python
count = Partner.objects.filter(is_company=True).count()
exists = Partner.objects.filter(email="a@example.com").exists()
```

#### Sorting and paging

```python
rows = (
    Partner.objects
    .filter(is_company=True)
    .order_by("name asc")
    .limit(20)
    .offset(0)
    .all()
)
```

#### Values and values_list

```python
rows = Partner.objects.filter(is_company=True).values("id", "name", limit=10)
names = Partner.objects.filter(is_company=True).values_list("name", flat=True, limit=10)
```

### Record Lifecycle

#### Create

```python
partner = Partner.objects.create(
    name="SoiGia Partner",
    email="hello@example.com",
    phone="+849999999",
)
```

#### Update via object

```python
partner.phone = "+848888888"
partner.save()
```

#### Refresh

```python
partner.refresh()
```

#### Delete

```python
partner.delete()
```

### Browse and RecordSet

```python
recordset = Partner.browse([1, 2, 3])
```

Useful helpers:

- `recordset.ids`
- `recordset.exists()`
- `recordset.read()`
- `recordset.records()`
- `recordset.first()`
- `recordset.write({...})`
- `recordset.delete()`
- `recordset.datatable(...)`

Example:

```python
recordset = Partner.objects.filter(is_company=True).browse()
recordset.write({"phone": "+840000000"})
```

### DataFrame Support

This client is built to work well with `pandas.DataFrame`.

#### Query Odoo into DataFrame

```python
df = Partner.objects.filter(is_company=True).datatable("id", "name", "email", limit=20)
```

`datatable(...)` returns a real `pd.DataFrame`.

You can also convert a recordset:

```python
recordset = Partner.objects.filter(is_company=True).limit(10).browse()
df = recordset.datatable(fields=["id", "name", "email", "phone"])
```

### Batch Write API

The batch API is one of the main reasons to use this client.

Define the model once:

```python
Partner = odoo.env.model(
    "res.partner",
    fields=["id", "name", "email", "phone", "is_company"],
    key_fields=["email"],
)
```

Then work directly with `DataFrame`.

#### Validate before write

```python
validated_df = Partner.validate(df, operation="save")
```

Validation checks:

- required `id` for update and delete
- required `key_fields` for save, upsert, and delete-by-keys
- duplicate keys inside the same batch

#### Insert many rows

```python
inserted = Partner.insert(df_insert)
print(inserted.ids)
```

#### Update many rows

`update()` expects an `id` column.

```python
updated_count = Partner.update(df_update)
```

#### Save many rows

`save()` is the recommended batch write method.

It behaves like batch upsert:

- update when the key already exists
- create when the key does not exist

```python
result = Partner.save(df_save)
```

Return format:

```python
{
    "created_ids": [...],
    "updated_ids": [...],
    "created_count": 1,
    "updated_count": 2,
}
```

#### Upsert many rows

`upsert()` is equivalent to `save()`.

```python
result = Partner.upsert(df_upsert)
```

#### Delete many rows by id

`delete()` expects `id`.

```python
deleted_count = Partner.delete(df_delete)
```

#### Delete many rows by business key

`delete_by_keys()` is useful when your source data does not have Odoo ids.

```python
delete_keys_df = pd.DataFrame(
    [
        {"email": "a@example.com"},
        {"email": "b@example.com"},
    ]
)

deleted_count = Partner.delete_by_keys(delete_keys_df)
```

### Full DataFrame Example

```python
import pandas as pd
from soigia.odoo_client import OdooClient

odoo = OdooClient(
    "http://127.0.0.1:8070",
    "odoo",
    login="admin",
    password="admin",
)

Partner = odoo.env.model(
    "res.partner",
    fields=["id", "name", "email", "phone", "is_company"],
    key_fields=["email"],
)

df_insert = pd.DataFrame(
    [
        {"name": "A", "email": "a@example.com", "phone": "+841"},
        {"name": "B", "email": "b@example.com", "phone": "+842"},
    ]
)

inserted = Partner.insert(df_insert)

df_update = inserted.datatable(fields=["id", "name", "email", "phone"]).copy()
df_update["phone"] = ["+8491", "+8492"]
Partner.update(df_update)

df_save = pd.DataFrame(
    [
        {"email": "a@example.com", "name": "A Updated", "phone": "+8471"},
        {"email": "c@example.com", "name": "C New", "phone": "+8472"},
    ]
)

validated_df = Partner.validate(df_save, operation="save")
result = Partner.save(validated_df)

df_delete = pd.DataFrame(
    [
        {"email": "a@example.com"},
        {"email": "b@example.com"},
        {"email": "c@example.com"},
    ]
)

Partner.delete_by_keys(df_delete)
```

### Calling Odoo Methods

#### Call a model method

```python
result = Partner.call("name_search", "SoiGia", limit=5)
```

#### Call a record method

```python
partner = Partner.objects.get(id=1)
partner.call("write", {"phone": "+849999999"})
```

#### Low-level generic execute

Use this when you want direct Odoo method access.

```python
result = odoo.execute("res.partner", "name_search", "SoiGia", limit=5)
```

This is useful for:

- custom model methods
- less common Odoo operations
- advanced workflows
- exact low-level control

### Transport Control

The client supports:

- `transport="auto"`
- `transport="json2"`
- `transport="xmlrpc"`

Example:

```python
result = odoo.execute("res.partner", "name_search", "A", transport="xmlrpc")
```

Default recommendation:

- use `auto` unless you have a specific reason

### Odoo Metadata

#### Fields

```python
fields = Partner.fields
field_names = Partner.field_names
```

#### Functions and methods

```python
functions = Partner.functions
methods = Partner.methods
```

Use these when you want to inspect the model from Python.

### Authentication Helpers

#### Ensure authenticated session

```python
odoo.ensure_authenticated()
```

#### Create and reuse API key

```python
api_key = odoo.ensure_api_key()
```

#### Revoke API key

```python
odoo.revoke_api_key()
```

### Context Helpers

You can clone the client with a modified Odoo context.

#### Language

```python
vi_odoo = odoo.with_lang("vi_VN")
```

#### Timezone

```python
tz_odoo = odoo.with_timezone("Asia/Bangkok")
```

#### Company

```python
company_odoo = odoo.with_company(1)
```

#### Generic context

```python
custom_odoo = odoo.with_context(active_test=False)
```

### Attachments

#### Upload

```python
attachment_id = odoo.upload_attachment(
    name="hello.txt",
    content=b"hello",
    res_model="res.partner",
    res_id=1,
    mimetype="text/plain",
)
```

#### Download

```python
content = odoo.download_attachment(attachment_id)
```

### Wizards and Reports

#### Create wizard

```python
wizard_id = odoo.create_wizard("res.config.settings", {"group_multi_company": True})
```

#### Run wizard

```python
result = odoo.run_wizard("res.config.settings", "execute", values={"group_multi_company": True})
```

#### Download report

```python
pdf_bytes = odoo.download_report("sale.report_saleorder", [1, 2])
```

### Logging and Timing

Batch methods include timing and logging decorators.

Logged operations include:

- `validate`
- `insert`
- `insert_many`
- `update`
- `update_many`
- `delete`
- `delete_many`
- `delete_by_keys`
- `upsert`
- `upsert_many`
- `save`

You can pass a custom logger:

```python
import logging

logger = logging.getLogger("myapp.odoo")

odoo = OdooClient(
    "http://127.0.0.1:8070",
    "odoo",
    login="admin",
    password="admin",
    logger=logger,
)
```

Log records include:

- `operation`
- `started_at`
- `finished_at`
- `duration_seconds`
- `status`

### Retry Behavior

The client supports retries for low-level network and RPC operations.

```python
odoo = OdooClient(
    "http://127.0.0.1:8070",
    "odoo",
    login="admin",
    password="admin",
    retry_count=2,
    retry_delay=0.5,
)
```

Use retries for:

- unstable local Docker
- VPN or network instability
- API services with intermittent connectivity
- background jobs that should tolerate short failures

### Recommended Patterns

#### Django service

```python
from soigia.odoo_client import OdooClient

odoo = OdooClient.from_django_settings()
Partner = odoo.env.model("res.partner", key_fields=["email"])
```

#### Airflow task

```python
def sync_partner_task():
    odoo = OdooClient.from_env()
    Partner = odoo.env.model("res.partner", key_fields=["email"])
    Partner.save(dataframe)
```

#### Console script

```python
def main():
    odoo = OdooClient.from_env()
    Partner = odoo.env.model("res.partner", key_fields=["email"])
    print(Partner.objects.datatable("id", "name", "email", limit=20))
```

#### API backend

```python
odoo = OdooClient.from_env()

def sync_partners(dataframe):
    Partner = odoo.env.model("res.partner", key_fields=["email"])
    validated = Partner.validate(dataframe, operation="save")
    return Partner.save(validated)
```

### Error Handling Notes

The main exceptions are:

- `OdooError`
- `OdooAuthenticationError`
- `OdooRpcError`
- `OdooJson2Error`
- `OdooHttpError`

Typical recommendation:

```python
from soigia.odoo_client import OdooError

try:
    result = Partner.save(df)
except OdooError as exc:
    print(f"Odoo error: {exc}")
```

### Samples

You can refer to the sample files in the repository:

- `samples/odoo_client_example.py`
- `samples/odoo_console_example.py`
- `samples/odoo_dataframe_sync_example.py`

### Practical Recommendation

For real applications, the best default pattern is:

1. build one `OdooClient`
2. define models from `odoo.env.model(...)`
3. use `Partner.validate(df, operation="save")`
4. use `Partner.save(df)` for main batch sync
5. use `Partner.delete_by_keys(df)` when you only have business keys
6. use `odoo.execute(...)` only for advanced Odoo methods

If your use case is mostly data synchronization, the most important methods are:

- `validate`
- `datatable`
- `insert`
- `update`
- `save`
- `delete`
- `delete_by_keys`

## Quickstart

Install the package:

```bash
pip install soigia
```

Or for Django projects:

```bash
pip install "soigia[django]"
```

Try a small domain example:

```python
from decimal import Decimal
from soigia.domain import common, sales

customer = common.Customer(name="Alice")
product = common.Product(
    sku=common.SKU("COFFEE-001"),
    name="Coffee",
    price=common.Money(Decimal("50000")),
    stock_qty=100,
)

order = sales.SalesOrder(customer=customer)
order.add_line(product=product, quantity=common.Quantity(2))

workflow = sales.SalesWorkflowService()
workflow.confirm(order)

print(order.amount_total)
```

## Install

For local development:

```bash
pip install -e .
```

For editable install with Django extras:

```bash
pip install -e ".[django]"
```

If you only need the runtime package, `pip install soigia` is enough.

## 1. DataTable

`DataTable` is the main table abstraction in `soigia`. It keeps the pandas mental model, but adds a more opinionated, workflow-friendly API for cleaning, filtering, joining, validating, and exporting data.

Use `DataTable` when you want:

- a table object that behaves like a dataframe
- chainable query-style operations
- simple data loading helpers
- schema checks before a dataset moves to the next step
- snapshot and version helpers for auditability

### Typical workflow

1. Load rows from records, CSV, JSON, Excel, Parquet, Feather, pickle, SQL, or Google Sheets.
2. Clean and normalize the data.
3. Filter, sort, deduplicate, or join with another table.
4. Validate the final shape and types.
5. Export or snapshot the result.

### Example

```python
from soigia.datatable import DataTable

orders = DataTable.from_records(
    [
        {"order_id": 1001, "customer": "Alice", "amount": 120.5, "city": "Hanoi"},
        {"order_id": 1002, "customer": "Bob", "amount": 88.0, "city": "Saigon"},
        {"order_id": 1003, "customer": "Carol", "amount": 240.0, "city": "Danang"},
    ]
)

adults = orders.objects.filter(amount__gte=100).order_by("amount")
print(adults.to_rows())
```

### Practical things you can do

- `DataTable.from_records(records)` for raw Python data
- `DataTable.from_csv(path)` for flat files
- `DataTable.from_json(path)` for payloads and exports
- `DataTable.from_excel(path)` for business spreadsheets
- `DataTable.from_parquet(path)` for analytics datasets
- `DataTable.from_sql(...)` for database-backed loading
- `df.objects.filter(...)` for query-like row selection
- `df.objects.exclude(...)` for inverse filtering
- `df.objects.order_by(...)` for sorting
- `df.objects.distinct(...)` for deduplication
- `df.objects.group_by(...)` for grouped summaries
- `df.join(...)` and related helpers for relational work
- `df.validate_schema(...)` for checks before release
- `df.snapshot(...)` and `df.auto_version(...)` for versioned outputs

### When to use it

`DataTable` is a good fit when:

- you need dataframe behavior, but want a stricter workflow layer
- you are passing data between cleaning steps and business logic
- you want tests to assert shape and content more clearly
- you need small, readable transformations instead of scattered pandas calls

### DataTable in practice

`DataTable` is the most direct way to work with structured records in Soi Gia. The main thing to remember is that it keeps the dataframe mental model, but the API is optimized for business workflows rather than generic data science notebooks.

If you are choosing between raw pandas and `DataTable`, use `DataTable` when:

- you want a clearer chainable API for row filtering and transformation
- you need the SDK to keep row-level intent readable in tests
- you want built-in helpers for export, validation, and snapshotting

## 2. Pipeline

`Pipeline` is the workflow layer for business processing. It is designed for repeatable jobs where data comes in, gets normalized, goes through a few stages, and then writes out artifacts and summary files.

Use `Pipeline` when you want:

- a predictable execution order
- stage-based processing
- automatic output files
- rollback hooks for side effects
- config-driven behavior
- a shared model namespace for computed results

### Mental model

1. `load_data()` brings data into memory.
2. Each stage mutates or enriches the working dataset.
3. `self.datatable` is the main working dataset.
4. `self._data` stores extra data loaded from DB tables.
5. `self.envs` gives fast SQLite table access for ad-hoc reads.
6. `self.model` stores computed results.
7. `self.config` exposes YAML values.
8. `save_outputs()` writes final artifacts.
9. `on_rollback(...)` protects side effects when a stage fails.

### Example

```python
from soigia.pipeline import Pipeline
from soigia.datatable import DataTable


class SalesPipeline(Pipeline):
    stages = ["clean", "enrich", "save", "summarize"]

    def load_data(self):
        return DataTable(
            [
                {"id": 1, "amount": 100.0},
                {"id": 2, "amount": 250.0},
            ]
        )

    def clean(self):
        self.datatable = self.datatable.dropna().reset_index(drop=True)

    def enrich(self):
        self.datatable["net_amount"] = self.datatable["amount"] * 0.98

    def save(self):
        self.datatable.save_to_csv("data/sales/clean_sales.csv")
        self.push("sales", self.datatable, mode="replace")

    def summarize(self):
        self.model.total_rows = len(self.datatable)
        self.model.total_amount = float(self.datatable["amount"].sum())


pipeline = SalesPipeline(name="sales")
result = pipeline.run()

print(result.success)
print(result.summary_path)
print(result.csv_path)
```

### What a pipeline usually includes

- input loading from files, databases, or APIs
- normalization and cleaning
- business-specific enrichment
- summary or scoring logic
- CSV, SQLite, and Parquet outputs
- a Markdown summary for traceability
- log files for debugging and audit

### Recommended structure

- keep the pipeline class small and stage-oriented
- put reusable config in `config.yaml`
- use `config.example.yaml` as the checked-in template
- use `pipeline.config.example.yaml` as the pipeline runtime template
- use `self.model` for computed values that later stages need
- keep rollback handlers focused on external effects, not dataframe-only work

### Pipeline DB defaults

`Pipeline` creates a SQLite database automatically on first use, so subclasses can read and write without extra setup.

The main working attribute is `self.datatable`. Extra tables loaded from DB are cached in `self._data`.

`self.datatable` is a `DataTable`, so you can use helpers like `save_to_csv()`, `save_to_parquet()`, `load_from_db()`, and `save_to_db()` directly. For SQLite, `connection_string` is optional; if you omit it, Soi Gia uses the default local DB:

```python
self.datatable.save_to_csv("data/output.csv")
self.datatable.save_to_parquet("data/output.parquet")
self.datatable.save_to_db("sales")
self.datatable = DataTable.load_from_db("select * from sales")
```

For quick reads, `self.envs` exposes table handles:

```python
orders = self.envs.orders.filter(self.envs.orders.id > 1)
```

You can also use keyword lookups:

```python
orders = self.envs.orders.filter(id__gt=1)
```

```python
class DemoPipeline(Pipeline):
    stages = ["seed", "sync", "reload"]

    def load_data(self):
        return DataTable([{"id": 1, "name": "alice"}])

    def seed(self):
        self.db.set("users", self.datatable)

    def sync(self):
        self.db.upsert_rows(
            "users",
            [{"id": 1, "name": "alice-updated"}, {"id": 2, "name": "bob"}],
        )

    def reload(self):
        self.datatable = DataTable.load_from_db("select * from users")
```

Short aliases are also available:

```python
self.push("users", self.datatable)
self.pull("users")
self.push_datatable("users", self.datatable, mode="upsert")
self.pull_datatable("users")
```

Default upsert keys can be configured per table:

```yaml
db_upsert_keys:
  users:
    - id
  orders:
    - order_id
```

### Pipeline in practice

`Pipeline` is the orchestration layer for repeatable business jobs. Treat it as a stage runner with explicit loading, cleaning, enrichment, persistence, and summary steps.

Use `Pipeline` when you need:

- deterministic stage execution
- automatic output artifacts
- rollback hooks for side effects
- config-driven behavior
- a shared `model` object for computed values

## 3. ORM

`ORM` is the model-driven data layer in `soigia`. It keeps the class-based model style of Odoo and Django, while separating storage through adapters like memory, pandas, and SQLite.

Use `ORM` when you want:

- declarative models with fields
- query helpers like `filter()`, `exclude()`, `order_by()`, and `limit()`
- record creation, updates, deletes, and browsing by id
- model inheritance and Odoo-style `_inherit`
- business methods on the model itself
- a backend-agnostic API that can switch storage engines

### Mental model

1. Define a model with `class MyModel(Model)`.
2. Declare fields directly on the class.
3. Bind the model to an `Env`.
4. Use `env["model.name"]` or `Model.objects` to query data.
5. Let the backend handle storage, while the model handles business behavior.

### Example

```python
from soigia.db import Env, MemoryBackend, Model, fields


class Customer(Model):
    _name = "demo.customer"

    name = fields.String(required=True)


env = Env(backend=MemoryBackend())
customer = env["demo.customer"].create({"name": "Alice"})

print(customer.id)
print(env["demo.customer"].search([("name", "=", "Alice")]).ids)
```

### ORM in practice

The ORM layer follows a model-centric style. Define fields on the model, bind it to an environment, and let the backend manage storage while the model keeps business behavior.

Use `ORM` when you need:

- declarative models with field definitions
- query helpers like `filter()`, `exclude()`, `order_by()`, and `limit()`
- record lifecycle methods on the model itself
- a backend-agnostic layer that can switch storage engines

## 4. UI

`soigia.ui` is a declarative Streamlit layer for internal tools. It is meant for dashboards, admin screens, review pages, and lightweight operational views.

Use `UI` when you want:

- a single-file screen definition
- simple page composition
- forms, metrics, tables, and filters
- a fast way to expose operational workflows

### Mental model

1. Define a page with `ui.page()`.
2. Compose layout blocks with `ui.sidebar()`, `ui.columns()`, `ui.tabs()`, and `ui.form()`.
3. Render data with `ui.table()` and `ui.metric()`.
4. Add inputs with `ui.text_input()`, `ui.select()`, and `ui.number_input()`.
5. Run the app with `ui.run()`.

### Example

```python
from soigia.ui import ui

orders = [
    {"id": 1, "customer": "Alice", "amount": 120.0, "status": "paid"},
    {"id": 2, "customer": "Bob", "amount": 88.5, "status": "pending"},
    {"id": 3, "customer": "Carol", "amount": 220.0, "status": "paid"},
]


@ui.page("Orders Dashboard")
def orders_page(ctx):
    ui.markdown("Track orders, filter data, and prepare quick actions.")

    with ui.sidebar():
        keyword = ui.text_input("Search customer")
        status = ui.select("Status", ["all", "paid", "pending"])
        min_amount = ui.number_input("Minimum amount", value=0)

    filtered = [
        row
        for row in orders
        if keyword.value.lower() in row["customer"].lower()
        and (status.value == "all" or row["status"] == status.value)
        and row["amount"] >= min_amount.value
    ]

    ui.metric("Orders", len(filtered))
    ui.metric("Total amount", sum(row["amount"] for row in filtered))
    ui.table(filtered)


ui.run()
```

### Common building blocks

- `ui.page()` for page registration
- `ui.sidebar()` for filters and navigation
- `ui.columns()` for split layouts
- `ui.tabs()` for grouped views
- `ui.form()` for update flows
- `ui.table()` for records and results
- `ui.metric()` for KPI cards
- `ui.text()`, `ui.markdown()`, and input widgets for interaction

### When to use it

`UI` works best for:

- internal admin dashboards
- review screens for ops or business teams
- quick forms and compact workflows
- tools that should stay simple enough to maintain in one file

### UI in practice

`soigia.ui` is a declarative Streamlit layer for internal tools. It is intentionally simple: compose a page, place a few widgets, render data, and keep the screen definition readable.

## Quick Summary

- Use `DataTable` when the problem is mostly table manipulation.
- Use `Pipeline` when the problem is a staged business workflow.
- Use `UI` when the problem is an internal screen or dashboard.

## Fake Data

`soigia.fake` provides Faker-backed demo tables as pandas DataFrames for quick local testing and mock integrations.
`soigia.news_fake` adds a newsroom/news CMS-style dataset, and `soigia.datasets` provides a unified namespace for all fake builders and exporters.

It includes linked datasets for a sales-style core:

- `customer`
- `user`
- `category`
- `product`
- `order`
- `orderline`

It also includes fuller business bundles with:

- `warehouse`
- `supplier`
- `inventory`
- `purchase_order`
- `purchase_orderline`
- `invoice`
- `payment`
- `shipment`

and a news/content bundle with:

- `newsroom`
- `section`
- `author`
- `editor`
- `category`
- `tag`
- `subscriber`
- `article`
- `article_tag`
- `media`
- `comment`
- `engagement`
- `newsletter`
- `newsletter_article`
- `analytics_daily`

Unified access:

```python
from soigia.datasets import build, export

sales = build("sales", seed=42)
news = build("news", seed=42)
export("news", news, "./artifacts/news", prefix="news_")
```

Example:

```python
from soigia.fake import build_fake_system

system = build_fake_system(seed=42)
orders = system.order
payments = system.payment
```

## 5. Domain

`soigia.domain` is the pure business layer of the SDK. It is organized by bounded context, and each context stays flat so the package is easy to scan when published on PyPI.

### Domain map

- `common` contains shared primitives used by more than one business module.
- `anki` contains spaced-repetition behavior for cards, decks, and review scheduling.
- `crm` contains lead, contact, activity, stage, and workflow behavior.
- `sales` contains sales order behavior and sales workflow rules.
- `purchases` contains supplier, purchase order, approval, and purchase workflow rules.
- `obsidian` contains note, vault, task, template, and canvas behavior for knowledge-base style workflows.

### Domain structure

Each bounded context uses the same flat layout:

- `entities.py`
- `policies.py`
- `services.py`
- `value_objects.py` when needed
- `enums.py`
- `exceptions.py`

That layout keeps the package readable without deep folder nesting.

### What belongs where

- `entities.py` holds entities, aggregates, and their internal behavior.
- `value_objects.py` holds immutable business primitives and validation rules.
- `policies.py` holds business decision rules and strategy objects.
- `services.py` holds orchestration that coordinates multiple domain objects.
- `enums.py` holds state and type enums.
- `exceptions.py` holds domain-specific errors.

### Example imports

```python
from soigia.domain import anki, crm, obsidian, purchases, sales

deck = anki.Deck(name=anki.DeckName("Japanese"))
lead = crm.Lead(name="Website inquiry", stage=crm.Stage(name="Qualification", sequence=10))
vault = obsidian.Vault(name=obsidian.VaultName("Main"))
```

### Domain usage rules

- Keep business rules inside the domain.
- Keep I/O, database access, API calls, and UI out of the domain.
- Prefer value objects for validated business data.
- Prefer services for cross-entity orchestration.
- Prefer policies for rule variation and decision logic.
- Keep tests in `tests/domain`, not inside the runtime package.

### Current domain modules

- `common` for shared order, product, money, quantity, and SKU behavior.
- `anki` for review cards, decks, and scheduling.
- `crm` for contacts, leads, activities, and lead scoring.
- `sales` for sales order workflow.
- `purchases` for purchase order workflow.
- `obsidian` for notes, templates, tasks, and canvas structure.

### Domain test coverage

The domain layer is covered by dedicated tests under `tests/domain`. Those tests are the contract for the package and should be treated as the source of truth for behavior changes.

### Install variants

```bash
pip install soigia
pip install "soigia[django]"
pip install -e .
pip install -e ".[django]"
```

Use the plain install if you only need the SDK, and use the Django extra if you need the Django integration layer.

### Domain cookbook

#### Common

```python
from decimal import Decimal
from soigia.domain import common

money = common.Money(Decimal("120000"))
quantity = common.Quantity(3)
sku = common.SKU("SKU-001")
product = common.Product(sku=sku, name="Coffee", price=money, stock_qty=10)
```

#### Anki

```python
from datetime import date
from soigia.domain import anki

deck = anki.Deck(name=anki.DeckName("Japanese"))
card = anki.Card(front=anki.CardText("猫"), back=anki.CardText("cat"))
deck.add_card(card)

service = anki.ReviewService()
service.review(card, anki.ReviewRating.GOOD, reviewed_on=date(2026, 4, 4))
```

#### CRM

```python
from decimal import Decimal
from soigia.domain import crm

stage = crm.Stage(name="Qualification", sequence=10, probability=Decimal("0.40"))
lead = crm.Lead(name="Website inquiry", stage=stage)
workflow = crm.LeadWorkflowService()
workflow.qualify(lead, crm.Stage(name="Proposal", sequence=20, probability=Decimal("0.75")))
```

#### Sales

```python
from decimal import Decimal
from soigia.domain import common, sales

customer = common.Customer(name="Alice")
product = common.Product(
    sku=common.SKU("COFFEE-001"),
    name="Coffee",
    price=common.Money(Decimal("50000")),
    stock_qty=100,
)

order = sales.SalesOrder(customer=customer)
order.add_line(product=product, quantity=common.Quantity(2))
workflow = sales.SalesWorkflowService()
workflow.confirm(order)
```

#### Purchases

```python
from decimal import Decimal
from soigia.domain import common, purchases

supplier = purchases.Supplier(name="ACME Supplies")
product = common.Product(
    sku=common.SKU("PAPER-A4"),
    name="Paper",
    price=common.Money(Decimal("20000")),
    stock_qty=500,
)

order = purchases.PurchaseOrder(supplier=supplier)
order.add_line(product=product, quantity=common.Quantity(10))
workflow = purchases.PurchaseWorkflowService()
workflow.confirm(order)
```

#### Obsidian

```python
from datetime import date
from soigia.domain import obsidian

vault = obsidian.Vault(name=obsidian.VaultName("Main"))
note = obsidian.Note(
    title=obsidian.NoteTitle("Daily Note"),
    path=obsidian.NotePath("Daily/2026-04-04.md"),
    content="# 2026-04-04\n- [ ] Follow up",
)

vault.add_note(note)
obsidian.TaskService().sync_note_tasks(note)
obsidian.VaultService().create_daily_note(vault, on_date=date(2026, 4, 4))
```

### Domain API notes

- `common` is the shared core. Keep pricing, stock, money, quantity, and SKU rules there.
- `anki` is for spaced repetition. `ReviewService` updates the card state and review history.
- `crm` is for leads and opportunities. `LeadWorkflowService` moves stages and refreshes scoring.
- `sales` is for sales orders. It owns sales state transitions and pricing behavior.
- `purchases` is for purchase orders. It owns approval logic and purchase workflow transitions.
- `obsidian` is for note-based knowledge work. It owns note, vault, task, template, and canvas behavior.

### Domain invariants to remember

- Do not put database code, API calls, or UI code inside `soigia.domain`.
- Keep entity methods focused on business state and invariant enforcement.
- Keep services focused on multi-object workflow orchestration.
- Keep policy objects focused on rule selection and transition decisions.
- Keep tests outside the runtime package in `tests/domain`.
