Zio.Ingestion()

The ingestion pipeline directly handles the original data (ingestion_data table) that it fills. Anything that comes in via mail, DB, spreadsheet, web, or API is accumulated here as raw_data. Querying, counting, creating, updating, and deleting (CRUD) are performed in the same way as with Zio.entity.

Do not guess field names

The ingestion schema varies completely by instance and industry (food, construction, fashion…). Before you try to guess what schemas exist or which fields they contain, first check them with Zio.ingestion.schemas(). Guessing field names quietly yields empty results.

What can you ask

The placeholders <스키마명> · <필드명> below are just stand‑ins. Use the actual names obtained from schemas().

When you receive this request	Do this
What is being collected? / What fields exist?	Zio.ingestion.schemas()
Show me the most recent data for OO	Zio.ingestion("<스키마명>").order_by("-ingested_at").fetch()
Show only those that match this condition	.where(**{"<필드명>": "<값>"}) .fetch()
How many match this condition?	.where(**{"<필드명>": "<값>"}) .count()
Match any of several fields (OR)	.where_or(status__contains="x", sender_email__contains="y").fetch() groups multiple conditions on the same schema with OR (different from other .where() which are AND). Within one group, top‑level columns and raw_data can be mixed. Calling it multiple times creates separate OR groups; the suffix is the same as .where(). It only applies to reads (fetch, count) — applying OR to update or delete would affect more rows than intended, so that is not allowed.
Only those not yet processed / errored	.where(status="error").fetch() · process_state bit condition
Fix a value / change a status	.where(...).update({...})
Delete these	.where(...).delete()
Insert one manually	Zio.ingestion("<스키마명>").create({...})

Relationships and group aggregations are not here

Questions like “who traded with whom” (relationship queries) or “how many per field value” (group‑by aggregates) must be looked at in the knowledge graph after analysis is finished → Zio.node. Here (Zio.ingestion) only deals with the original data, performing conditional lookups and .count() only.

Fluent CRUD API for Ingested Data

Method	Description
Zio.ingestion.schemas()	Retrieves a list of ingestion schemas. Each item is {name, group, description, count, columns[]} — name is the unique identifier (used as an address), group is the pipeline type (e.g., odbc, shared by many), and columns contains the actual field names · types of that schema. Call this before querying.
Zio.ingestion(schema)	The starting point for Fluent chaining on a single ingestion schema. Put the name returned by schemas() into schema — name is uniquely enforced, so it becomes the official address. If you provide a group (pipeline type) name instead of an address, it returns the schema names within that group; choose one from those. (The numeric schema_id is also accepted, but since names are now unique, it is rarely needed.)
.where(**filters)	Apply conditions. Works on fields inside raw_data as well as top-level columns (id, status, process_state, source_identifier, ingested_at). Suffixes: __contains, __gte, __lte, __in, etc. If a field name contains Korean or spaces, pass it as .where(**{"필드명": value}).
.order_by(*fields)	Sorting. Prefix a field with - for descending order (e.g., "-ingested_at").
.fetch(limit=100, skip=0)	Retrieves data matching the conditions as a GraphResult. Flattens raw_data and includes id, status, and process_state.
.count()	Counts rows that match the conditions. (No group aggregation — see above.)
.create(props)	Inserts a single row of data. props are fields for raw_data. Defaults to status="pending" and process_state=1 (collection complete) before subsequent processing. Usually performed by the pipeline itself; rarely used manually.
.where(...).update(props)	Updates rows that match the conditions. Can modify raw_data fields as well as status and process_state. Rejects if no where condition is provided.
.where(...).delete()	Deletes rows that match the conditions. Also removes related items (analysis, deep analysis, attachments, inference queue) and attached files on disk. Irreversible; rejects if no condition is given.

Process History & Status Values — process_state and status

The extent of processing a row has reached is indicated by process_state (bitwise accumulation), while the engine’s handling decision is governed by status. Changing these values via update causes cron workers to automatically reprocess—the SDK itself does not trigger re-execution.

process_state	Integer Value	Meaning
Bit 0	1	Collection complete
Bit 1	2	Data enrichment (AI omission correction)
Bit 2	4	Category classification
Bit 3	8	Aggregation / summarization
Bit 4	16	Agent (Dynamic) · Knowledge graph processing
Bit 5	32	Final inference

status	Engine	Meaning
pending	Running	Awaiting processing (new or manually triggered reprocessing)
processing	Locked	Worker in progress (deduplication)
done	Stopped	Current assigned process completed
error	Halted	Stopped due to exception (needs review)
paused	Halted	Manually paused by user

Reverting collection stages (Bits 0–3) automatically resets subsequent bits (4 and 5) to 0, preserving freshness.

Example (actual schema and field names can be verified with schemas() — placeholders below)

from zio_ontology import Zio

# 1. 무엇을 수집하고 있나 — 필드명은 여기서 확인한다 (인스턴스마다 다르다)
for s in Zio.ingestion.schemas():
    print(s["name"], s["count"], [c["name"] for c in s["columns"]])

# 2. 조건에 맞는 건수 세기
#    <스키마명> · <필드명> 은 위에서 확인한 실제 이름으로 바꾼다
n = Zio.ingestion("<스키마명>").where(**{"<필드명>": "<값>"}).count()

# 3. 최근 것부터 100건 조회
rows = Zio.ingestion("<스키마명>").order_by("-ingested_at").fetch(limit=100).data

# 4. 조건에 맞는 값 고치기 (where 없으면 거절)
Zio.ingestion("<스키마명>").where(**{"<필드명>": "<값>"}).update({"<필드명>": "<새값>"})

# 5. 상태를 바꿔 재처리 흐르게 하기 (cron 이 알아서 다시 처리)
Zio.ingestion("<스키마명>").where(id=123).update({"status": "pending", "process_state": 1})

# 6. 조건에 맞는 것 지우기 (딸린 것·첨부 파일까지 함께, 되돌릴 수 없음)
Zio.ingestion("<스키마명>").where(**{"<필드명>": "<값>"}).delete()

---

Modifying the source does not propagate derived results

Even if you change raw_data via update, knowledge graphs and analysis results already created will not automatically update. To reflect changes in analysis outcomes, either reset the relevant bit of process_state (e.g., keep only the collection bit and set others to 0) so that cron reprocesses, or directly modify the graph using Zio.node.