---
title: 'Create domain term endpoints (single + bulk)'
type: 'feature'
created: '2026-04-10'
status: 'done'
baseline_commit: '90da41c'
context:
  - '_planning/docs/services/textiq-doc-extraction/architecture.md'
---

<frozen-after-approval reason="human-owned intent — do not modify unless human renegotiates">

## Intent

**Problem:** The Airflow plugin has a PATCH endpoint for updating domain terms but no way to create them. APA needs to create domain terms — both individually and in bulk — via the plugin API (TEXTIQ-341, TEXTIQ-344).

**Approach:** Add `POST /` (single create) and `POST /bulk` (bulk create with `skip_duplicates` flag) to the existing `domain_terms_app` in `airflow/plugins/api/domain_terms.py`. Follow existing plugin patterns: HMAC auth, RLS, raw SQL via `rls_connection`. Duplicate check scoped to tenant+node within RLS context.

## Boundaries & Constraints

**Always:**
- HMAC authentication + RLS tenant isolation (same as existing PATCH endpoint)
- `tenant_id` and `node_id` come from HMAC context (`request.state.hmac_ctx`), not request body
- Duplicate check: within RLS connection, check `term_original + node_id` uniqueness
- Set `tenant_id` and `node_id` on all created records from HMAC context
- Bulk endpoint: cap at 500 terms per request

**Ask First:**
- Whether to add an Alembic migration changing `term_original` unique constraint from global to composite `(term_original, tenant_id, node_id)` — needed if different tenants/nodes can have the same term

**Never:**
- Modify existing PATCH endpoint behavior
- Import from `src.*` (Airflow plugin restriction)
- Cascade to Milvus or chunks (creation only adds to dictionary table)

## I/O & Edge-Case Matrix

| Scenario | Input / State | Expected Output / Behavior | Error Handling |
|----------|--------------|---------------------------|----------------|
| Single create | Valid term_original + source_language | 201 with created term | N/A |
| Single duplicate | term_original exists for same tenant+node | N/A | 409 DUPLICATE_TERM |
| Bulk happy path | terms array, skip_duplicates=false | 201 with counts + results | N/A |
| Bulk with dupes, skip=true | Some terms exist | 201, skipped count > 0 | N/A |
| Bulk with dupes, skip=false | Some terms exist | N/A | 409 DUPLICATE_TERMS |
| Empty terms array | terms: [] | N/A | 422 validation error |
| Bulk exceeds 500 | 501 terms | N/A | 422 validation error |
| Missing required fields | No term_original | N/A | 422 validation error |

</frozen-after-approval>

## Code Map

- `airflow/plugins/api/domain_terms.py` -- Add POST endpoints + Pydantic schemas (existing file with PATCH)
- `tests/airflow/test_plugins/test_domain_terms.py` -- Add tests using existing _FakeConnection pattern

## Tasks & Acceptance

**Execution:**
- [x] `airflow/plugins/api/domain_terms.py` -- Add Pydantic schemas: DomainTermCreate, DomainTermBulkCreate
- [x] `airflow/plugins/api/domain_terms.py` -- Add `POST /` endpoint: duplicate check within RLS, INSERT with tenant_id+node_id from HMAC, return 201
- [x] `airflow/plugins/api/domain_terms.py` -- Add `POST /bulk` endpoint: iterate terms, duplicate check per term, respect skip_duplicates flag, return counts
- [x] `tests/airflow/test_plugins/test_domain_terms.py` -- Add tests for all I/O matrix scenarios using existing test patterns

**Acceptance Criteria:**
- Given valid HMAC auth, when POST /api/v1/domain-terms with valid body, then term is created with correct tenant_id and node_id from HMAC context and 201 returned
- Given valid HMAC auth, when POST /api/v1/domain-terms/bulk with skip_duplicates=true and some duplicates, then new terms created, duplicates skipped, and response includes accurate counts
- Given a duplicate term_original for same tenant+node, when POST single create, then 409 returned

## Verification

**Commands:**
- `python -m pytest tests/airflow/test_plugins/test_domain_terms.py -v` -- expected: all tests pass

## Suggested Review Order

**Schemas & endpoint logic**

- Pydantic request models: DomainTermCreate (single) and DomainTermBulkCreate (with skip_duplicates)
  [`domain_terms.py:44`](../../airflow/plugins/api/domain_terms.py#L44)

- POST / — single create with RLS duplicate check, IntegrityError fallback to 409
  [`domain_terms.py:111`](../../airflow/plugins/api/domain_terms.py#L111)

- POST /bulk — partition into new/skipped, respect skip_duplicates flag, IntegrityError handling
  [`domain_terms.py:191`](../../airflow/plugins/api/domain_terms.py#L191)

**Tests**

- Single create: happy path, duplicate 409, missing fields 422, custom category
  [`test_domain_terms.py:96`](../../tests/airflow/test_plugins/test_domain_terms.py#L96)

- Bulk create: happy path, skip=true/false, empty array, intra-batch dedup
  [`test_domain_terms.py:162`](../../tests/airflow/test_plugins/test_domain_terms.py#L162)

- IntegrityError handling: single and bulk return 409 on constraint violation
  [`test_domain_terms.py:292`](../../tests/airflow/test_plugins/test_domain_terms.py#L292)

- Validation: 501 terms rejected, missing required fields in bulk
  [`test_domain_terms.py:355`](../../tests/airflow/test_plugins/test_domain_terms.py#L355)

- Auth: POST and POST /bulk return 401 without HMAC
  [`test_domain_terms.py:372`](../../tests/airflow/test_plugins/test_domain_terms.py#L372)