luplo.core.items

CRUD operations for the items table.

Items are the central entity in luplo — decisions, knowledge, policies, and documents all live here. Edits create new rows via supersedes_id (the old row is never mutated). Deletes are soft (deleted_at is set, row stays).

Attributes

ITEM_COLUMNS

Functions

row_to_item(→ luplo.core.models.Item)	Convert a dict-row into an Item, dropping unknown columns and NULL-safe defaults.
create_item(→ luplo.core.models.Item)	Insert a new item and return it.
get_item(→ luplo.core.models.Item | None)	Fetch a single item by ID or hex prefix.
get_item_including_deleted(→ luplo.core.models.Item | None)	Fetch a single item by ID or prefix, even if soft-deleted.
list_items(→ list[luplo.core.models.Item])	List items for a project with optional filters.
delete_item(→ bool)	Soft-delete an item by setting deleted_at.
get_supersedes_chain(→ list[luplo.core.models.Item])	Walk the supersedes_id chain from item_id back to the original.

Module Contents

luplo.core.items.ITEM_COLUMNS = ('id', 'project_id', 'item_type', 'title', 'body', 'source_url', 'parent_item_id',...

luplo.core.items.row_to_item(row: dict[str, Any]) → luplo.core.models.Item

Convert a dict-row into an Item, dropping unknown columns and NULL-safe defaults.

async luplo.core.items.create_item(conn: psycopg.AsyncConnection[Any], data: luplo.core.models.ItemCreate) → luplo.core.models.Item

Insert a new item and return it.

• Generates a UUID4 id automatically.

• Populates search_tsv from title + body + rationale.

• If data.supersedes_id is set the new row is treated as an edit of the referenced item (the old row is untouched).

Parameters:

• conn – An async psycopg connection (inside an open transaction).

• data – Fields for the new item.

Returns:

The fully-populated Item as stored in the database.

async luplo.core.items.get_item(conn: psycopg.AsyncConnection[Any], item_id: str, *, project_id: str | None = None) → luplo.core.models.Item | None

Fetch a single item by ID or hex prefix.

Parameters:

• conn – Open async connection.

• item_id – Either a full UUID or a hex prefix of at least luplo.core.id_resolve.MIN_PREFIX_LENGTH characters.

• project_id – Optional project scope for prefix lookups. Strongly recommended whenever the caller knows it; without it, prefix collisions are evaluated across every project in the database.

Returns:

The matching item, or None when no row exists / the row is soft-deleted.

Raises:

• AmbiguousIdError – If the prefix matches multiple rows.

• IdTooShortError – If a hex prefix shorter than the minimum is supplied.

• InvalidIdFormatError – If the input is not a UUID or hex prefix.

async luplo.core.items.get_item_including_deleted(conn: psycopg.AsyncConnection[Any], item_id: str, *, project_id: str | None = None) → luplo.core.models.Item | None

Fetch a single item by ID or prefix, even if soft-deleted.

See get_item() for argument and exception semantics. The only difference is that soft-deleted rows are still returned.

async luplo.core.items.list_items(conn: psycopg.AsyncConnection[Any], project_id: str, *, item_type: str | None = None, system_id: str | None = None, work_unit_id: str | None = None, include_deleted: bool = False, limit: int = 100, offset: int = 0) → list[luplo.core.models.Item]

List items for a project with optional filters.

Parameters:

• conn – Async psycopg connection.

• project_id – Required project scope.

• item_type – Filter by item_type (e.g. "decision").

• system_id – Filter items whose system_ids array contains this value.

• work_unit_id – Filter by work_unit_id.

• include_deleted – If True, include soft-deleted items.

• limit – Maximum rows returned (default 100).

• offset – Pagination offset.

Returns:

List of Item ordered by created_at DESC.

async luplo.core.items.delete_item(conn: psycopg.AsyncConnection[Any], item_id: str, *, actor_id: str) → bool

Soft-delete an item by setting deleted_at.

The row is never removed. get_item will return None for it, but list_items(include_deleted=True) will still find it.

Parameters:

• conn – Async psycopg connection.

• item_id – ID of the item to delete.

• actor_id – Who performed the deletion (for audit trail).

Returns:

True if the item existed and was deleted, False if it was already deleted or not found.

async luplo.core.items.get_supersedes_chain(conn: psycopg.AsyncConnection[Any], item_id: str) → list[luplo.core.models.Item]

Walk the supersedes_id chain from item_id back to the original.

Returns the full chain ordered oldest-first (the original item is at index 0, the most recent version — item_id itself — is last).

If item_id does not exist, returns an empty list.