luplo.core.backend.local

LocalBackend — implements the Backend Protocol via direct PG access.

All core module functions take a raw AsyncConnection. This class acquires a connection from the pool for each operation, adds cross-cutting concerns (audit logging, history recording on supersede), and delegates to the corresponding core function.

Classes

LocalBackend

Backend that talks directly to PostgreSQL via a connection pool.

Module Contents

class luplo.core.backend.local.LocalBackend(pool: psycopg_pool.AsyncConnectionPool, embedding: luplo.core.embedding.EmbeddingBackend | None = None)

Backend that talks directly to PostgreSQL via a connection pool.

This is the primary backend for lp init --local (single-user mode) and for the FastAPI server’s internal use.

Parameters:

  • pool – An open AsyncConnectionPool.

  • embedding – Embedding backend for search reranking. Defaults to NullEmbedding (no vectors).

pool
async create_project(*, id: str, name: str, description: str | None = None) → luplo.core.models.Project
async get_project(id: str) → luplo.core.models.Project | None
async list_projects() → list[luplo.core.models.Project]
async create_actor(*, id: str, name: str, email: str | None = None, role: str | None = None, external_ids: dict[str, str] | None = None) → luplo.core.models.Actor
async get_actor(id: str) → luplo.core.models.Actor | None
async list_item_types() → list[luplo.core.models.ItemType]
async get_item_type(key: str) → luplo.core.models.ItemType | None
async create_item_type(*, key: str, display_name: str, schema: dict[str, Any], owner: str = 'user') → luplo.core.models.ItemType
async get_actor_by_email(email: str) → luplo.core.models.Actor | None
async open_work_unit(*, id: str, project_id: str, title: str, description: str | None = None, system_ids: list[str] | None = None, created_by: str | None = None, context: dict[str, Any] | None = None) → luplo.core.models.WorkUnit
async get_work_unit(id: str, *, project_id: str | None = None) → luplo.core.models.WorkUnit | None
async list_work_units(project_id: str, *, status: str | None = None) → list[luplo.core.models.WorkUnit]
async close_work_unit(id: str, *, actor_id: str, force: bool = False) → luplo.core.models.WorkUnit | None
async archive_work_unit(*, id: str, archived_by: str, replaced_by_wu_id: str) → luplo.core.models.WorkUnit
async find_existing_import_wu(*, project_id: str, content_hash_set: tuple[str, Ellipsis]) → luplo.core.models.WorkUnit | None
async create_system(*, id: str, project_id: str, name: str, description: str | None = None, depends_on_system_ids: list[str] | None = None) → luplo.core.models.System
async get_system(id: str, *, project_id: str | None = None) → luplo.core.models.System | None
async list_systems(project_id: str) → list[luplo.core.models.System]
async update_system(id: str, **kwargs: Any) → luplo.core.models.System | None
async create_item(data: luplo.core.models.ItemCreate) → luplo.core.models.Item
async get_item(id: str, *, project_id: str | None = None) → luplo.core.models.Item | None
async list_items(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]
async delete_item(id: str, *, actor_id: str) → None
async get_supersedes_chain(id: str) → list[luplo.core.models.Item]
async impact(item_id: str, project_id: str, *, depth: int = 5) → luplo.core.impact.ImpactResult
async run_checks(project_id: str, *, rule_names: list[str] | None = None, disabled: tuple[str, Ellipsis] = ()) → list[luplo.core.checks.Finding]
async search(query: str, project_id: str, *, item_types: list[str] | None = None, system_ids: list[str] | None = None, limit: int = 10, tsquery: str | None = None) → list[luplo.core.models.SearchResult]
async add_capture(*, text: str, created_by: str | None = None, summary: str | None = None, sensitivity_hint: str = 'none', signals: dict[str, Any] | None = None) → luplo.core.models.Capture
async list_captures(*, review_state: str | None = None, include_discarded: bool = False, include_redacted: bool = False, since: datetime.datetime | None = None, until: datetime.datetime | None = None, limit: int = 100) → list[luplo.core.models.Capture]
async get_capture(capture_id: str) → luplo.core.models.Capture | None
async search_captures(*, query: str | None = None, review_state: str | None = None, include_discarded: bool = False, include_redacted: bool = False, since: datetime.datetime | None = None, until: datetime.datetime | None = None, limit: int = 50) → list[luplo.core.models.Capture]
async set_capture_state(capture_id: str, *, review_state: str, actor_id: str | None = None) → luplo.core.models.Capture
async discard_capture(capture_id: str, *, actor_id: str | None = None) → luplo.core.models.Capture
async redact_capture(capture_id: str, *, redacted_by: str | None = None) → luplo.core.models.Capture
async annotate_capture(capture_id: str, *, summary: str | None = None, sensitivity_hint: str | None = None, signals: dict[str, Any] | None = None) → luplo.core.models.Capture
async promote_capture_to_item(capture_id: str, data: luplo.core.models.ItemCreate) → tuple[luplo.core.models.Capture, luplo.core.models.Item]
async create_glossary_group(*, id: str, project_id: str, canonical: str, definition: str | None = None, scope: str = 'project', scope_id: str | None = None, created_by: str | None = None) → luplo.core.models.GlossaryGroup
async get_glossary_group(id: str, *, project_id: str | None = None) → luplo.core.models.GlossaryGroup | None
async list_glossary_groups(project_id: str, *, needs_review: bool = False, limit: int = 100, offset: int = 0) → list[luplo.core.models.GlossaryGroup]
async create_glossary_term(*, id: str, group_id: str | None, surface: str, normalized: str, is_protected: bool = False, status: str = 'pending', source_item_id: str | None = None, context_snippet: str | None = None) → luplo.core.models.GlossaryTerm
async list_pending_terms(project_id: str, *, limit: int = 50) → list[luplo.core.models.GlossaryTerm]
async create_glossary_group_with_canonical(*, project_id: str, canonical: str, definition: str | None = None, actor_id: str | None = None) → tuple[luplo.core.models.GlossaryGroup, luplo.core.models.GlossaryTerm]
async add_term_to_group(group_id: str, *, surface: str, actor_id: str, as_canonical: bool = False) → luplo.core.models.GlossaryTerm
async delete_glossary_term(term_id: str, *, actor_id: str) → bool
async approve_term(term_id: str, *, group_id: str, actor_id: str, as_canonical: bool = False) → luplo.core.models.GlossaryTerm | None
async reject_term(term_id: str, *, actor_id: str, reason: str | None = None) → luplo.core.models.GlossaryRejection | None
async merge_groups(source_group_id: str, target_group_id: str, *, actor_id: str) → luplo.core.models.GlossaryGroup | None
async split_term(term_id: str, *, new_canonical: str, actor_id: str) → luplo.core.models.GlossaryGroup | None
async expand_query(query: str, project_id: str) → str
async record_history(*, item_id: str, version: int, changed_by: str, content_before: str | None = None, content_after: str | None = None, content_hash_before: str | None = None, content_hash_after: str | None = None, diff_summary: str | None = None, semantic_impact: str | None = None, source_event_id: str | None = None) → luplo.core.models.HistoryEntry
async query_history(*, project_id: str | None = None, item_id: str | None = None, since: datetime.datetime | None = None, semantic_impacts: list[str] | None = None, limit: int = 50) → list[luplo.core.models.HistoryEntry]
async record_audit(*, actor_id: str, action: str, target_type: str | None = None, target_id: str | None = None, metadata: dict[str, Any] | None = None) → None
async create_task(*, project_id: str, work_unit_id: str, title: str, actor_id: str, sort_order: int | None = None, systems: list[str] | None = None, body: str | None = None, context_extra: dict[str, Any] | None = None) → luplo.core.models.Item
async get_task(task_id: str, *, project_id: str | None = None) → luplo.core.models.Item | None
async list_tasks(work_unit_id: str, *, status: str | None = None) → list[luplo.core.models.Item]
async get_in_progress_task(work_unit_id: str) → luplo.core.models.Item | None
async start_task(task_id: str, *, actor_id: str, project_id: str | None = None) → luplo.core.models.Item
async complete_task(task_id: str, *, actor_id: str, summary: str | None = None, project_id: str | None = None) → luplo.core.models.Item
async block_task(task_id: str, *, actor_id: str, reason: str, project_id: str | None = None) → luplo.core.models.Item
async skip_task(task_id: str, *, actor_id: str, reason: str | None = None, project_id: str | None = None) → luplo.core.models.Item
async reorder_tasks(work_unit_id: str, task_ids: list[str], *, actor_id: str, project_id: str | None = None) → list[luplo.core.models.Item]
async suggest_decision_from_task(task_id: str, *, project_id: str | None = None) → luplo.core.models.ItemCreate | None
async edit_task(task_id: str, *, actor_id: str, title: str | None = None, body: str | None = None, sort_order: int | None = None, project_id: str | None = None) → luplo.core.models.Item
async add_idea(*, project_id: str, work_unit_id: str, text: str, created_by: str | None = None) → luplo.core.models.Idea
async list_ideas(*, work_unit_id: str, project_id: str | None = None, limit: int = 100, include_redacted: bool = False) → list[luplo.core.models.Idea]
async search_ideas(*, project_id: str, query: str | None = None, tsquery: str | None = None, work_unit_id: str | None = None, author: str | None = None, since: datetime.datetime | None = None, until: datetime.datetime | None = None, include_redacted: bool = False, limit: int = 50) → list[luplo.core.models.Idea]
async get_idea(idea_id: str, *, project_id: str | None = None) → luplo.core.models.Idea | None
async redact_idea(*, idea_id: str, redacted_by: str, project_id: str | None = None) → tuple[luplo.core.models.Idea, bool]
async create_qa(*, project_id: str, title: str, actor_id: str, coverage: str, areas: list[str] | None = None, target_item_ids: list[str] | None = None, target_task_ids: list[str] | None = None, work_unit_id: str | None = None, body: str | None = None, context_extra: dict[str, Any] | None = None) → luplo.core.models.Item
async get_qa(qa_id: str, *, project_id: str | None = None) → luplo.core.models.Item | None
async list_qa(project_id: str, *, status: str | None = None, work_unit_id: str | None = None) → list[luplo.core.models.Item]
async list_pending_qa_for_task(task_id: str) → list[luplo.core.models.Item]
async list_pending_qa_for_item(item_id: str) → list[luplo.core.models.Item]
async list_pending_qa_for_wu(work_unit_id: str) → list[luplo.core.models.Item]
async start_qa(qa_id: str, *, actor_id: str, project_id: str | None = None) → luplo.core.models.Item
async pass_qa(qa_id: str, *, actor_id: str, evidence: str | None = None, project_id: str | None = None) → luplo.core.models.Item
async fail_qa(qa_id: str, *, actor_id: str, reason: str, project_id: str | None = None) → luplo.core.models.Item
async block_qa(qa_id: str, *, actor_id: str, reason: str, project_id: str | None = None) → luplo.core.models.Item
async skip_qa(qa_id: str, *, actor_id: str, project_id: str | None = None) → luplo.core.models.Item
async assign_qa(qa_id: str, *, actor_id: str, assignee_actor_id: str, project_id: str | None = None) → luplo.core.models.Item
async enqueue_sync(*, source_type: str, source_page_id: str, payload: str | None = None, source_event_id: str | None = None, debounce_seconds: int = 300) → luplo.core.models.SyncJob
async get_ready_sync_jobs(*, limit: int = 1) → list[luplo.core.models.SyncJob]
async complete_sync_job(job_id: int) → None
async fail_sync_job(job_id: int, *, error: str) → None