bvandeusenandClaude Opus 5
cca40affe4
CI & Build / Python lint (push) Successful in 2s
CI & Build / TypeScript typecheck (push) Successful in 12s
CI & Build / integration (push) Successful in 25s
CI & Build / Python tests (push) Failing after 29s
CI & Build / Build & push image (push) Has been skipped
Milestone #232 step 1 (task #2081). Takes the enabler first rather than the write-path trigger: reverse lookup, drift checks and the duplicate finder all need to QUERY structured fields, and building them on body-regex first means writing them twice. #227 deferred this bag "unless body-convention ergonomics prove insufficient" — answering "which snippets live in this file?" by scanning every snippet and regexing its body is that condition being met. Migration 0070 adds `notes.data` (nullable JSONB) + a GIN index. The body is UNCHANGED and still what gets embedded and read by humans; `data` mirrors the same facts in a shape Postgres can index. Code is deliberately not copied into it — the body holds it, and duplicating a blob into the column we index around would be waste. - compose_data() builds the mirror, omitting empties so the column stays sparse - snippet_fields() prefers `data`, falling back to parsing the body. Rows written before 0070 have no `data` and are never backfilled, so a hand-edited body stays authoritative for them with no conversion deadline - create / update / merge all write body and mirror from the same merged field set, so the two can't drift; merge in particular has to grow the mirror with the survivor's location set or a merged snippet would be unfindable at the very call sites the merge just recorded Named `data`, not `metadata`, because that collides with SQLAlchemy's declarative Base.metadata — which is why the pre-0069 model had to map an awkward `entity_metadata` attribute. Not a revival of the column 0069 dropped: different name, different purpose, nothing reads the old shape. Two test fakes needed an explicit `data = None`: snippet_fields prefers `data` when truthy and an auto-MagicMock attribute is truthy, so every parsed field would have come back a MagicMock. Checked every fake reaching snippet code this time rather than waiting for CI (note 2109). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RLwAaV4DQEmVyn496HnEvt
630 lines
24 KiB
Python
630 lines
24 KiB
Python
"""Snippet service — reusable functions/components recorded for recall.
|
|
|
|
A *snippet* is a Note with ``note_type='snippet'``: a named, reusable function or
|
|
component recorded once so a later session can recall it before writing a
|
|
one-off. It carries a name, language, signature, canonical location
|
|
(repo · path · symbol), a one-line "when to reach for it", and the code itself.
|
|
|
|
Structured fields are stored as a **body-convention** — no dedicated column, so
|
|
a snippet inherits everything a note has (embeddings, ACL, project/System
|
|
association, dedup) and, crucially, becomes eligible for semantic recall the
|
|
moment it's embedded:
|
|
|
|
- ``title`` = ``"{name} — {when_to_use}"``. The title is exactly what the
|
|
title-first auto-inject surfaces, so this one line self-describes the snippet
|
|
in a recall menu.
|
|
- ``tags`` = ``[language, "snippet", *caller_tags]``.
|
|
- ``body`` = templated markdown (When to use / Signature / Location, then a
|
|
fenced code block).
|
|
|
|
The public field API (name/language/signature/location/when_to_use/code) lives
|
|
here, so the underlying storage could later move to a structured column without
|
|
changing any caller (MCP tool, REST route, UI).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import re
|
|
|
|
from scribe.services import knowledge as knowledge_svc
|
|
from scribe.services import notes as notes_svc
|
|
|
|
SNIPPET_NOTE_TYPE = "snippet"
|
|
SNIPPET_TAG = "snippet"
|
|
|
|
# Sentinel for "argument not supplied" on update, so None stays available as a
|
|
# real value meaning "clear this". Needed for project_id, where 0 is not a valid
|
|
# id and None is the clear — the two can't share one default.
|
|
UNSET: object = object()
|
|
|
|
|
|
def _embed_snippet(note) -> None:
|
|
"""Fire-and-forget embedding refresh for a snippet.
|
|
|
|
A snippet's whole value is *immediate* recall — it must join the semantic /
|
|
auto-inject pool the moment it's recorded, not wait for the startup backfill.
|
|
Unlike a plain note (embedded at the REST-route boundary only, so its MCP
|
|
create path defers to restart-backfill), a snippet is recorded primarily via
|
|
MCP, so we embed here in the service — covering BOTH the MCP tool and the
|
|
REST route by construction. Mirrors the route pattern: fire-and-forget,
|
|
text = title + body. Import lazily so the pure serialize/parse helpers can be
|
|
imported without pulling in the embedding model.
|
|
"""
|
|
text = f"{note.title}\n{note.body}".strip() if note.body else (note.title or "")
|
|
if not text:
|
|
return
|
|
from scribe.services.embeddings import upsert_note_embedding
|
|
asyncio.create_task(upsert_note_embedding(note.id, note.user_id, text))
|
|
|
|
|
|
# --- serialize: structured fields -> note (title/body/tags) ------------------
|
|
|
|
def compose_title(name: str, when_to_use: str = "") -> str:
|
|
"""`name — when to use` (or just `name` when no usage note is given)."""
|
|
name = (name or "").strip()
|
|
when = (when_to_use or "").strip()
|
|
return f"{name} — {when}" if when else name
|
|
|
|
|
|
def compose_tags(language: str = "", tags: list[str] | None = None) -> list[str]:
|
|
"""Language (lowercased) first, then the `snippet` marker, then caller tags —
|
|
de-duplicated, order preserved."""
|
|
out: list[str] = []
|
|
lang = (language or "").strip().lower()
|
|
if lang:
|
|
out.append(lang)
|
|
out.append(SNIPPET_TAG)
|
|
for t in tags or []:
|
|
t = (t or "").strip()
|
|
if t and t not in out:
|
|
out.append(t)
|
|
return out
|
|
|
|
|
|
def _normalize_locations(locations: list[dict] | None) -> list[dict]:
|
|
"""Clean a list of {repo,path,symbol} locations: strip fields, drop wholly
|
|
empty entries, de-duplicate identical ones (order preserved). A merged
|
|
snippet carries several locations (one per call site); a fresh one carries
|
|
at most one."""
|
|
out: list[dict] = []
|
|
seen: set[tuple[str, str, str]] = set()
|
|
for loc in locations or []:
|
|
repo = (loc.get("repo") or "").strip()
|
|
path = (loc.get("path") or "").strip()
|
|
symbol = (loc.get("symbol") or "").strip()
|
|
if not (repo or path or symbol):
|
|
continue
|
|
key = (repo, path, symbol)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append({"repo": repo, "path": path, "symbol": symbol})
|
|
return out
|
|
|
|
|
|
def _location_str(loc: dict) -> str:
|
|
"""`repo` · `path` · `symbol` — only the non-empty parts."""
|
|
parts = [(loc.get(k) or "").strip() for k in ("repo", "path", "symbol")]
|
|
return " · ".join(f"`{p}`" for p in parts if p)
|
|
|
|
|
|
def _render_location_block(locations: list[dict]) -> str | None:
|
|
"""One `**Location:**` line for a single location; a `**Locations:**` bullet
|
|
list for several. None when there are none."""
|
|
locs = [loc for loc in locations if _location_str(loc)]
|
|
if not locs:
|
|
return None
|
|
if len(locs) == 1:
|
|
return f"**Location:** {_location_str(locs[0])}"
|
|
lines = "\n".join(f"- {_location_str(loc)}" for loc in locs)
|
|
return f"**Locations:**\n{lines}"
|
|
|
|
|
|
def compose_body(
|
|
*,
|
|
code: str,
|
|
language: str = "",
|
|
signature: str = "",
|
|
when_to_use: str = "",
|
|
repo: str = "",
|
|
path: str = "",
|
|
symbol: str = "",
|
|
locations: list[dict] | None = None,
|
|
) -> str:
|
|
"""Render structured fields into the snippet body markdown. Empty fields are
|
|
omitted so the body stays clean.
|
|
|
|
Locations: pass ``locations`` (a list of {repo,path,symbol}) for the general
|
|
multi-location case; the single ``repo``/``path``/``symbol`` params remain as
|
|
a back-compat shorthand for one location and are used only when ``locations``
|
|
is not given.
|
|
"""
|
|
if locations is None:
|
|
locations = [{"repo": repo, "path": path, "symbol": symbol}]
|
|
locs = _normalize_locations(locations)
|
|
|
|
header: list[str] = []
|
|
if (when_to_use or "").strip():
|
|
header.append(f"**When to use:** {when_to_use.strip()}")
|
|
if (signature or "").strip():
|
|
header.append(f"**Signature:** `{signature.strip()}`")
|
|
loc_block = _render_location_block(locs)
|
|
if loc_block:
|
|
header.append(loc_block)
|
|
fence_lang = (language or "").strip().lower()
|
|
code_block = f"```{fence_lang}\n{(code or '').rstrip()}\n```"
|
|
if header:
|
|
return "\n\n".join(header) + "\n\n" + code_block + "\n"
|
|
return code_block + "\n"
|
|
|
|
|
|
# --- parse: note -> structured fields (best-effort, never raises) ------------
|
|
|
|
_WHEN_RE = re.compile(r"^\*\*When to use:\*\*\s*(.+?)\s*$", re.MULTILINE)
|
|
_SIG_RE = re.compile(r"^\*\*Signature:\*\*\s*(.+?)\s*$", re.MULTILINE)
|
|
_LOC_RE = re.compile(r"^\*\*Location:\*\*\s*(.+?)\s*$", re.MULTILINE)
|
|
_LOCS_RE = re.compile(
|
|
r"^\*\*Locations:\*\*[ \t]*\n((?:[ \t]*-[ \t]*.+\n?)+)", re.MULTILINE
|
|
)
|
|
_CODE_RE = re.compile(r"```([\w+.#-]*)\n(.*?)\n```", re.DOTALL)
|
|
|
|
|
|
def _parse_location_str(s: str) -> dict | None:
|
|
"""Parse a `repo` · `path` · `symbol` fragment back into a location dict, or
|
|
None if it's empty."""
|
|
raw = [p.strip().strip("`").strip() for p in (s or "").split("·")]
|
|
repo = raw[0] if len(raw) > 0 else ""
|
|
path = raw[1] if len(raw) > 1 else ""
|
|
symbol = raw[2] if len(raw) > 2 else ""
|
|
if not (repo or path or symbol):
|
|
return None
|
|
return {"repo": repo, "path": path, "symbol": symbol}
|
|
|
|
|
|
def parse_snippet_fields(
|
|
title: str, body: str, tags: list[str] | None = None
|
|
) -> dict:
|
|
"""Recover structured fields from a snippet note. Tolerant by design: a field
|
|
that isn't present comes back empty and this never raises, so a hand-edited
|
|
body can't break the edit form.
|
|
|
|
``locations`` is a list of {repo,path,symbol}; ``repo``/``path``/``symbol``
|
|
mirror the FIRST location for back-compat with the single-location callers."""
|
|
title = title or ""
|
|
body = body or ""
|
|
name, _, when_from_title = title.partition(" — ")
|
|
fields = {
|
|
"name": name.strip(),
|
|
"when_to_use": when_from_title.strip(),
|
|
"signature": "",
|
|
"language": "",
|
|
"repo": "",
|
|
"path": "",
|
|
"symbol": "",
|
|
"locations": [],
|
|
"code": "",
|
|
}
|
|
|
|
m = _WHEN_RE.search(body)
|
|
if m:
|
|
fields["when_to_use"] = m.group(1).strip()
|
|
m = _SIG_RE.search(body)
|
|
if m:
|
|
fields["signature"] = m.group(1).strip().strip("`").strip()
|
|
|
|
# Locations: prefer the multi-location `**Locations:**` bullet list, else the
|
|
# legacy single `**Location:**` line.
|
|
locations: list[dict] = []
|
|
m = _LOCS_RE.search(body)
|
|
if m:
|
|
for line in m.group(1).splitlines():
|
|
line = line.strip()
|
|
if line.startswith("-"):
|
|
loc = _parse_location_str(line[1:])
|
|
if loc:
|
|
locations.append(loc)
|
|
else:
|
|
m = _LOC_RE.search(body)
|
|
if m:
|
|
loc = _parse_location_str(m.group(1))
|
|
if loc:
|
|
locations.append(loc)
|
|
fields["locations"] = locations
|
|
if locations:
|
|
fields["repo"] = locations[0]["repo"]
|
|
fields["path"] = locations[0]["path"]
|
|
fields["symbol"] = locations[0]["symbol"]
|
|
|
|
m = _CODE_RE.search(body)
|
|
if m:
|
|
fields["language"] = m.group(1).strip()
|
|
fields["code"] = m.group(2)
|
|
|
|
# Language fallback for a body whose code fence lost its language. Only the
|
|
# FIRST tag can be trusted: compose_tags emits [language, "snippet", *caller],
|
|
# so a leading tag that isn't the marker is the language — while a leading
|
|
# marker means no language was recorded. Scanning for "first tag that isn't
|
|
# the marker" instead would promote a caller's plain tag to the language.
|
|
if not fields["language"] and tags and tags[0] != SNIPPET_TAG:
|
|
fields["language"] = tags[0]
|
|
return fields
|
|
|
|
|
|
# --- the queryable mirror (notes.data, migration 0070) -----------------------
|
|
|
|
# Fields kept in `data`. Code is deliberately absent: the body already holds it,
|
|
# and copying a blob into the column we index *around* would be pure weight.
|
|
_DATA_FIELDS = ("name", "when_to_use", "signature", "language", "locations")
|
|
|
|
|
|
def compose_data(
|
|
*,
|
|
name: str = "",
|
|
when_to_use: str = "",
|
|
signature: str = "",
|
|
language: str = "",
|
|
locations: list[dict] | None = None,
|
|
) -> dict:
|
|
"""Build the `notes.data` mirror of a snippet's structured fields.
|
|
|
|
Same facts as the body convention, in a shape Postgres can index — so
|
|
"which snippets live in this path?" is a containment query rather than a
|
|
regex over every body. Empty values are omitted so the column stays sparse
|
|
and containment matches don't trip over blanks.
|
|
"""
|
|
out: dict = {}
|
|
for key, value in (
|
|
("name", (name or "").strip()),
|
|
("when_to_use", (when_to_use or "").strip()),
|
|
("signature", (signature or "").strip()),
|
|
("language", (language or "").strip().lower()),
|
|
):
|
|
if value:
|
|
out[key] = value
|
|
locs = _normalize_locations(locations)
|
|
if locs:
|
|
out["locations"] = locs
|
|
return out
|
|
|
|
|
|
def snippet_fields(note) -> dict:
|
|
"""Structured fields for a snippet, preferring the indexed `data` column and
|
|
falling back to parsing the body.
|
|
|
|
Both paths must agree, because rows written before migration 0070 have no
|
|
`data` and are never backfilled — a hand-edited body is the authority for
|
|
those, and there is no deadline by which they must be converted. `code` only
|
|
ever comes from the body, since `data` doesn't carry it.
|
|
"""
|
|
parsed = parse_snippet_fields(note.title, note.body, note.tags)
|
|
stored = getattr(note, "data", None)
|
|
if not stored:
|
|
return parsed
|
|
merged = dict(parsed)
|
|
for key in _DATA_FIELDS:
|
|
if stored.get(key):
|
|
merged[key] = stored[key]
|
|
# Keep the single-location back-compat mirror consistent with whichever
|
|
# location list won.
|
|
locs = merged.get("locations") or []
|
|
merged["repo"] = locs[0]["repo"] if locs else ""
|
|
merged["path"] = locs[0]["path"] if locs else ""
|
|
merged["symbol"] = locs[0]["symbol"] if locs else ""
|
|
return merged
|
|
|
|
|
|
def snippet_to_dict(note) -> dict:
|
|
"""Note serialization plus a parsed ``snippet`` sub-object of structured
|
|
fields, so callers get both the raw record and the typed view.
|
|
|
|
The raw `data` column is intentionally NOT exposed: it mirrors what
|
|
``snippet`` already reports, and shipping both would give API consumers two
|
|
sources of truth for the same facts."""
|
|
data = note.to_dict()
|
|
data["snippet"] = snippet_fields(note)
|
|
return data
|
|
|
|
|
|
# --- service wrappers over notes_svc ----------------------------------------
|
|
|
|
async def create_snippet(
|
|
user_id: int,
|
|
*,
|
|
name: str,
|
|
code: str,
|
|
language: str = "",
|
|
signature: str = "",
|
|
when_to_use: str = "",
|
|
repo: str = "",
|
|
path: str = "",
|
|
symbol: str = "",
|
|
locations: list[dict] | None = None,
|
|
tags: list[str] | None = None,
|
|
project_id: int | None = None,
|
|
):
|
|
"""Create a snippet note (embedded on create for immediate recall). Returns
|
|
the created Note. Pass ``locations`` for the multi-location case; the single
|
|
``repo``/``path``/``symbol`` are the one-location shorthand."""
|
|
if locations is None:
|
|
locations = [{"repo": repo, "path": path, "symbol": symbol}]
|
|
note = await notes_svc.create_note(
|
|
user_id,
|
|
title=compose_title(name, when_to_use),
|
|
body=compose_body(
|
|
code=code, language=language, signature=signature,
|
|
when_to_use=when_to_use, locations=locations,
|
|
),
|
|
note_type=SNIPPET_NOTE_TYPE,
|
|
tags=compose_tags(language, tags),
|
|
project_id=project_id,
|
|
# The indexed mirror of the same fields (0070). Written together with the
|
|
# body so the two can never describe different things.
|
|
data=compose_data(
|
|
name=name, when_to_use=when_to_use, signature=signature,
|
|
language=language, locations=locations,
|
|
),
|
|
)
|
|
_embed_snippet(note)
|
|
return note
|
|
|
|
|
|
async def get_snippet(user_id: int, snippet_id: int):
|
|
"""Fetch a snippet by id, or None if it doesn't exist / isn't a snippet /
|
|
isn't readable by this user.
|
|
|
|
Share-aware (rule #78): a fetch by id is an explicit act, so it resolves the
|
|
caller's full read scope rather than ownership alone. Without this, a snippet
|
|
that a search legitimately surfaced could not then be opened — see #2093."""
|
|
result = await notes_svc.get_note_for_user(user_id, snippet_id)
|
|
if result is None:
|
|
return None
|
|
note, _permission = result
|
|
if note.note_type != SNIPPET_NOTE_TYPE or note.deleted_at is not None:
|
|
return None
|
|
return note
|
|
|
|
|
|
async def list_snippets(
|
|
user_id: int,
|
|
*,
|
|
q: str | None = None,
|
|
tag: str = "",
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
project_id: int | None = None,
|
|
) -> tuple[list[dict], int]:
|
|
"""List snippets (id/title/tags/preview dicts), most-recently-updated first.
|
|
|
|
``project_id`` narrows to one project; omit it to reach across every project
|
|
— which is the point when the thing you're about to write was already solved
|
|
somewhere else."""
|
|
return await knowledge_svc.query_knowledge(
|
|
user_id=user_id,
|
|
note_type=SNIPPET_NOTE_TYPE,
|
|
tags=[tag] if tag else [],
|
|
sort="modified",
|
|
q=q,
|
|
limit=max(1, min(limit, 100)),
|
|
offset=max(0, offset),
|
|
project_id=project_id,
|
|
)
|
|
|
|
|
|
async def update_snippet(
|
|
user_id: int,
|
|
snippet_id: int,
|
|
*,
|
|
name: str | None = None,
|
|
code: str | None = None,
|
|
language: str | None = None,
|
|
signature: str | None = None,
|
|
when_to_use: str | None = None,
|
|
repo: str | None = None,
|
|
path: str | None = None,
|
|
symbol: str | None = None,
|
|
locations: list[dict] | None = None,
|
|
tags: list[str] | None = None,
|
|
project_id: int | None | object = UNSET,
|
|
):
|
|
"""Partial update: only fields passed (not None) change. Re-serializes the
|
|
merged field set back into title/body/tags. Returns the Note, or None if the
|
|
id isn't a snippet the caller can see.
|
|
|
|
Share-aware (rule #47/#78): resolves the read scope, then requires WRITE —
|
|
so an editor/admin grant lets the holder edit, and a viewer grant does not.
|
|
Raises PermissionError when the caller can read but not write, because "not
|
|
found" would be a lie about a record they can plainly open. The write itself
|
|
is performed as the OWNER, mirroring routes/snippets.py, since the underlying
|
|
note update is owner-scoped.
|
|
|
|
``project_id``: omit to leave unchanged, pass None to detach from its
|
|
project, pass an id to move it.
|
|
|
|
Locations: ``locations`` replaces the whole set; else a legacy single
|
|
``repo``/``path``/``symbol`` overlays onto the first existing location; else
|
|
the existing locations are kept."""
|
|
note = await get_snippet(user_id, snippet_id)
|
|
if note is None:
|
|
return None
|
|
from scribe.services.access import can_write_note
|
|
if not await can_write_note(user_id, snippet_id):
|
|
raise PermissionError(
|
|
f"snippet {snippet_id} is shared with you read-only — ask its owner "
|
|
f"for edit access, or record your own version"
|
|
)
|
|
|
|
cur = snippet_fields(note)
|
|
overlay = {
|
|
"name": name, "code": code, "language": language,
|
|
"signature": signature, "when_to_use": when_to_use,
|
|
}
|
|
merged = {**cur, **{k: v for k, v in overlay.items() if v is not None}}
|
|
|
|
if locations is not None:
|
|
merged_locations = _normalize_locations(locations)
|
|
elif repo is not None or path is not None or symbol is not None:
|
|
base = cur["locations"][0] if cur["locations"] else {"repo": "", "path": "", "symbol": ""}
|
|
merged_locations = _normalize_locations([{
|
|
"repo": repo if repo is not None else base["repo"],
|
|
"path": path if path is not None else base["path"],
|
|
"symbol": symbol if symbol is not None else base["symbol"],
|
|
}])
|
|
else:
|
|
merged_locations = cur["locations"]
|
|
|
|
fields: dict = {
|
|
"title": compose_title(merged["name"], merged["when_to_use"]),
|
|
"body": compose_body(
|
|
code=merged["code"], language=merged["language"],
|
|
signature=merged["signature"], when_to_use=merged["when_to_use"],
|
|
locations=merged_locations,
|
|
),
|
|
# Re-derived from the same merged field set as the body, so an edit can't
|
|
# leave the indexed mirror describing the previous version.
|
|
"data": compose_data(
|
|
name=merged["name"], when_to_use=merged["when_to_use"],
|
|
signature=merged["signature"], language=merged["language"],
|
|
locations=merged_locations,
|
|
),
|
|
}
|
|
# Recompute tags: keep any non-language, non-marker tags the note already had
|
|
# (or the caller's replacement set), then re-derive language + marker.
|
|
existing_extra = [
|
|
t for t in (note.tags or []) if t not in (SNIPPET_TAG, cur.get("language", ""))
|
|
]
|
|
fields["tags"] = compose_tags(
|
|
merged["language"], tags if tags is not None else existing_extra
|
|
)
|
|
if project_id is not UNSET:
|
|
fields["project_id"] = project_id
|
|
|
|
# As the OWNER: update_note is owner-scoped, so a shared editor's own id
|
|
# would find nothing. The write was authorised by can_write_note above.
|
|
updated = await notes_svc.update_note(note.user_id, snippet_id, **fields)
|
|
if updated is not None:
|
|
# Title/body changed → refresh the embedding so recall reflects the edit.
|
|
_embed_snippet(updated)
|
|
return updated
|
|
|
|
|
|
async def delete_snippet(user_id: int, snippet_id: int) -> bool:
|
|
"""Retire a snippet to the trash (recoverable). Returns False if the id isn't
|
|
a snippet this user may WRITE.
|
|
|
|
Recall makes this corrective, not merely tidy: a wrong or obsolete snippet
|
|
doesn't sit quietly — it keeps being offered as prior art. Removing it has to
|
|
be reachable from wherever it was recorded.
|
|
|
|
Note the explicit write check: `get_snippet` resolves the READ scope, which
|
|
now includes snippets merely shared with this user — being able to see one
|
|
must not imply being able to bin it.
|
|
"""
|
|
note = await get_snippet(user_id, snippet_id)
|
|
if note is None:
|
|
return False
|
|
from scribe.services.access import can_write_note
|
|
if not await can_write_note(user_id, snippet_id):
|
|
return False
|
|
from scribe.services.trash import delete as trash_delete
|
|
return await trash_delete(note.user_id, "note", snippet_id) is not None
|
|
|
|
|
|
# --- merge: unify found one-offs into one canonical snippet ------------------
|
|
|
|
def _extra_tags(tags: list[str] | None, language: str = "") -> list[str]:
|
|
"""A snippet's caller tags — everything except the language + `snippet`
|
|
markers that compose_tags re-derives."""
|
|
lang = (language or "").strip().lower()
|
|
return [t for t in (tags or []) if t and t != SNIPPET_TAG and t != lang]
|
|
|
|
|
|
def merge_snippet_fields(
|
|
target_fields: dict, target_tags: list[str] | None, sources: list[tuple[dict, list]]
|
|
) -> tuple[list[dict], list[str]]:
|
|
"""Pure merge: union locations (target's first, then each source in order)
|
|
and union extra tags. The target's scalar fields (name/when_to_use/signature/
|
|
language/code) win — only locations and tags accumulate. ``sources`` is a
|
|
list of (parsed_fields, tags). Returns (locations, extra_tags)."""
|
|
locations = list(target_fields.get("locations") or [])
|
|
extra = _extra_tags(target_tags, target_fields.get("language", ""))
|
|
for sfields, stags in sources:
|
|
locations.extend(sfields.get("locations") or [])
|
|
for t in _extra_tags(stags, sfields.get("language", "")):
|
|
if t not in extra:
|
|
extra.append(t)
|
|
return _normalize_locations(locations), extra
|
|
|
|
|
|
async def merge_snippets(user_id: int, target_id: int, source_ids: list[int]):
|
|
"""Unify source snippets INTO the target: union their locations + tags onto
|
|
the canonical target, keep the target's scalar fields, trash the sources
|
|
(recoverable), re-embed the survivor.
|
|
|
|
Share-aware and write-gated, like update: an editor/admin grant is enough, a
|
|
viewer grant is not. Raises PermissionError when the caller can read the
|
|
target but not write it. Every source must share the TARGET'S OWNER —
|
|
cross-owner merge stays out of scope (#231) — and must itself be writable;
|
|
sources failing either test are skipped rather than silently half-merged.
|
|
|
|
Returns (merged_target_note, merged_source_ids), or None if the target isn't
|
|
a snippet the caller can see."""
|
|
from scribe.services.access import can_write_note
|
|
target = await get_snippet(user_id, target_id)
|
|
if target is None:
|
|
return None
|
|
if not await can_write_note(user_id, target_id):
|
|
raise PermissionError(
|
|
f"snippet {target_id} is shared with you read-only — you can't merge "
|
|
f"into a record you can't edit"
|
|
)
|
|
owner_id = target.user_id
|
|
|
|
sources = []
|
|
for sid in source_ids:
|
|
if sid == target_id:
|
|
continue
|
|
s = await get_snippet(user_id, sid)
|
|
# Same owner as the target, and writable by this caller. Merging trashes
|
|
# the source, so read access is not enough.
|
|
if s is None or s.user_id != owner_id:
|
|
continue
|
|
if not await can_write_note(user_id, sid):
|
|
continue
|
|
sources.append(s)
|
|
|
|
tgt_fields = snippet_fields(target)
|
|
parsed_sources = [(snippet_fields(s), s.tags) for s in sources]
|
|
locations, extra_tags = merge_snippet_fields(tgt_fields, target.tags, parsed_sources)
|
|
|
|
# Owner-scoped write, authorised above — same reason as update_snippet.
|
|
updated = await notes_svc.update_note(
|
|
owner_id, target_id,
|
|
body=compose_body(
|
|
code=tgt_fields["code"], language=tgt_fields["language"],
|
|
signature=tgt_fields["signature"], when_to_use=tgt_fields["when_to_use"],
|
|
locations=locations,
|
|
),
|
|
tags=compose_tags(tgt_fields["language"], extra_tags),
|
|
# The survivor's location set grew, so its mirror has to grow with it —
|
|
# otherwise a merged snippet would be unfindable at the very call sites
|
|
# the merge just recorded.
|
|
data=compose_data(
|
|
name=tgt_fields["name"], when_to_use=tgt_fields["when_to_use"],
|
|
signature=tgt_fields["signature"], language=tgt_fields["language"],
|
|
locations=locations,
|
|
),
|
|
)
|
|
if updated is None:
|
|
return None
|
|
|
|
# Retire the merged-in sources to the trash (recoverable).
|
|
from scribe.services.trash import delete as trash_delete
|
|
merged_ids: list[int] = []
|
|
for s in sources:
|
|
batch = await trash_delete(owner_id, "note", s.id)
|
|
if batch is not None:
|
|
merged_ids.append(s.id)
|
|
|
|
_embed_snippet(updated)
|
|
return updated, merged_ids
|