luplo.core.links

CRUD operations for the links table.

Links are typed, weighted edges between items — e.g. excludes, synergizes_with, interacts_with. The link_type is free-form text (not an enum) so new relationship types can appear as systems grow.

Functions

create_link(→ luplo.core.models.Link)	Create a typed edge between two items.
get_links(→ list[luplo.core.models.Link])	Query links connected to an item.
delete_link(→ bool)	Delete a link by its composite primary key.

Module Contents

async luplo.core.links.create_link(conn: psycopg.AsyncConnection[Any], *, from_item_id: str, to_item_id: str, link_type: str, strength: int = 5, note: str | None = None, actor_id: str | None = None) → luplo.core.models.Link

Create a typed edge between two items.

Parameters:

• conn – Async psycopg connection.

• from_item_id – Source item.

• to_item_id – Target item.

• link_type – Relationship type (e.g. "excludes").

• strength – Weight 1-10, default 5.

• note – Optional annotation.

• actor_id – Who created this link.

Returns:

The newly created Link.

Raises:

psycopg.errors.UniqueViolation – If the exact (from_item_id, to_item_id, link_type) triple already exists.

async luplo.core.links.get_links(conn: psycopg.AsyncConnection[Any], item_id: str, *, direction: str = 'from', link_type: str | None = None) → list[luplo.core.models.Link]

Query links connected to an item.

Parameters:

• conn – Async psycopg connection.

• item_id – The item to query links for.

• direction – "from" (outgoing), "to" (incoming), or "both".

• link_type – Optional filter by relationship type.

Returns:

List of matching Link objects ordered by created_at DESC.

async luplo.core.links.delete_link(conn: psycopg.AsyncConnection[Any], from_item_id: str, to_item_id: str, link_type: str) → bool

Delete a link by its composite primary key.

Returns:

True if the link existed and was deleted, False otherwise.