soigia 0.1.21

Python SDK for Soi Gia

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.

How to use:

from soigia import fake_vars

actual = fake_vars.run_with_fake_locals(
    target,
    a=2,
    b=3,
    data=fake_vars.head(1),
    status=fake_vars.if_equal("processing", "test"),
    count=1,
)

This is useful when:

• data is too large and you only want the first row/item
• you want to swap one exact local value for a test
• you want to keep the rest of the function logic unchanged

Quick rules:

1. Match by local variable name
2. Pass one rule per variable
3. Use a transform when you do not want a plain value

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.

Pytest example

from soigia import fake_vars


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

    def process(self):
        qty = 10
        price = 5
        discount = 2
        subtotal = qty * price
        total = subtotal - discount
        status = "OK" if total > 0 else "BAD"
        self.audit.append(status)
        qty = 2000


def test_order_method_with_fake_locals():
    service = OrderService()

    result = fake_vars.run_with_fake_locals(
        service.process,
        qty=100,
        price=3,
        discount=1,
    )

    assert result.qty == 2000
    assert result.price == 3
    assert result.discount == 1
    assert result.subtotal == 300
    assert result.total == 299
    assert result.status == "OK"
    assert service.audit == ["OK"]

Mixed locals example

import pandas as pd

from soigia import fake_vars


def test_process_orders_with_mixed_locals():
    def process_orders():
        data = pd.DataFrame(
            {
                "order_id": [1, 2, 3],
                "amount": [100, 200, 300],
                "status": ["new", "new", "paid"],
            }
        )
        items = ["A", "B", "C"]
        config = {"debug": True, "limit": 100}
        status = "processing"
        count = 999

        total_amount = data["amount"].sum()
        return data, items, config, status, count, total_amount

    actual = fake_vars.run_with_fake_locals(
        process_orders,
        data=fake_vars.head(1),
        items=fake_vars.head(1),
        config=fake_vars.if_equal({"debug": True, "limit": 100}, {"debug": False, "limit": 1}),
        status=fake_vars.if_equal("processing", "test"),
        count=1,
    )

    assert len(actual.data) == 1
    assert actual.data.iloc[0]["order_id"] == 1
    assert actual.items == ["A"]
    assert actual.config == {"debug": False, "limit": 1}
    assert actual.status == "test"
    assert actual.count == 1
    assert actual.total_amount == 100

unittest example

import unittest

from soigia import fake_vars


class TestOrderCalc(unittest.TestCase):
    def test_order_function_with_fake_locals(self):
        audit = []

        def process():
            qty = 10
            price = 5
            discount = 2
            subtotal = qty * price
            total = subtotal - discount
            status = "OK" if total > 0 else "BAD"
            audit.append(status)
            qty = 2000

        result = fake_vars.run_with_fake_locals(process, qty=100, price=3, discount=1)

        self.assertEqual(result.qty, 2000)
        self.assertEqual(result.price, 3)
        self.assertEqual(result.discount, 1)
        self.assertEqual(result.subtotal, 300)
        self.assertEqual(result.total, 299)
        self.assertEqual(result.status, "OK")
        self.assertEqual(audit, ["OK"])

Typical usage:

• data=fake_vars.head(1) when data is a large DataFrame or list-like value
• status=fake_vars.if_equal("processing", "test") when you want to swap one exact value
• count=1 when you want to force a simple scalar

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
• Pipeline
• ORM
• UI
• Fake Data
• Odoo Client
• 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

pip install soigia

If you use Django:

pip install "soigia[django]"

Import

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:

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. It starts Odoo and PostgreSQL, initializes the base module on first run, and is ready for soigia.odoo_client.

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:

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

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

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

Connection Styles

Direct configuration

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:

odoo = OdooClient.from_env()

Example shell:

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:

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

Then:

odoo = OdooClient.from_django_settings()

Attribute style is also supported:

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:

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

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

Filter records

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

First record

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

Count and exists

count = Partner.objects.filter(is_company=True).count()
exists = Partner.objects.filter(email="[e-mail]").exists()

Sorting and paging

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

Values and values_list

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

partner = Partner.objects.create(
    name="SoiGia Partner",
    email="[e-mail]",
    phone="[phone]",
)

Update via object

partner.phone = "[phone]"
partner.save()

Refresh

partner.refresh()

Delete

partner.delete()

Browse and RecordSet

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:

recordset = Partner.objects.filter(is_company=True).browse()
recordset.write({"phone": "[phone]"})

DataFrame Support

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

Query Odoo into DataFrame

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:

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:

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

Then work directly with DataFrame.

Validate before write

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

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

Update many rows

update() expects an id column.

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

result = Partner.save(df_save)

Return format:

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

Upsert many rows

upsert() is equivalent to save().

result = Partner.upsert(df_upsert)

Delete many rows by id

delete() expects id.

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.

delete_keys_df = pd.DataFrame(
    [
        {"email": "[e-mail]"},
        {"email": "[e-mail]"},
    ]
)

deleted_count = Partner.delete_by_keys(delete_keys_df)

Full DataFrame Example

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": "[e-mail]", "phone": "+841"},
        {"name": "B", "email": "[e-mail]", "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": "[e-mail]", "name": "A Updated", "phone": "+8471"},
        {"email": "[e-mail]", "name": "C New", "phone": "+8472"},
    ]
)

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

df_delete = pd.DataFrame(
    [
        {"email": "[e-mail]"},
        {"email": "[e-mail]"},
        {"email": "[e-mail]"},
    ]
)

Partner.delete_by_keys(df_delete)

Calling Odoo Methods

Call a model method

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

Call a record method

partner = Partner.objects.get(id=1)
partner.call("write", {"phone": "[phone]"})

Low-level generic execute

Use this when you want direct Odoo method access.

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:

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

Default recommendation:

• use auto unless you have a specific reason

Odoo Metadata

Fields

fields = Partner.fields
field_names = Partner.field_names

Functions and methods

functions = Partner.functions
methods = Partner.methods

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

Authentication Helpers

Ensure authenticated session

odoo.ensure_authenticated()

Create and reuse API key

api_key = odoo.ensure_api_key()

Revoke API key

odoo.revoke_api_key()

Context Helpers

You can clone the client with a modified Odoo context.

Language

vi_odoo = odoo.with_lang("vi_VN")

Timezone

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

Company

company_odoo = odoo.with_company(1)

Generic context

custom_odoo = odoo.with_context(active_test=False)

Attachments

Upload

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

Download

content = odoo.download_attachment(attachment_id)

Wizards and Reports

Create wizard

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

Run wizard

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

Download report

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:

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.

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

from soigia.odoo_client import OdooClient

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

Airflow task

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

Console script

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

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:

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:

pip install soigia

Or for Django projects:

pip install "soigia[django]"

Try a small domain example:

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:

pip install -e .

For editable install with Django extras:

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

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

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:

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:

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

You can also use keyword lookups:

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

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:

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:

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

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

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:

from soigia.datasets import build, export

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

Example:

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

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

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

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

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

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

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

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

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.