Self-hosted knowledge management for SMEs: a split-screen Markdown editor whose sections an LLM refines while you write, and RAG question answering over the documents that result. FastAPI + Postgres/pgvector on the back, SvelteKit on the front, everything OpenAI-compatible and self-hostable. Squashed into a single commit; the development history stays local. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CA43ZJda8Rbp2hKXNy8f6b
110 lines
3.9 KiB
Python
110 lines
3.9 KiB
Python
"""Single source of truth for who may read which document.
|
|
|
|
Used by BOTH the documents API and retrieval, so the permission filter that
|
|
runs before the LLM can never drift from what the API
|
|
exposes. Always evaluated against the live documents table — never against
|
|
denormalized chunk meta, which can be stale between edits and reindexing.
|
|
"""
|
|
|
|
from sqlalchemy import ColumnElement, and_, exists, or_, select, true
|
|
|
|
from app.models import (
|
|
DocPermission,
|
|
Document,
|
|
DocumentStatus,
|
|
DocumentVisibility,
|
|
ReviewRequest,
|
|
User,
|
|
)
|
|
|
|
|
|
def searchable_documents_filter(user: User) -> ColumnElement[bool]:
|
|
"""Published documents the user may read.
|
|
|
|
Rules: public to everyone; department to members of the owning
|
|
department; restricted only via doc_permissions grants. Authors always
|
|
see their own documents.
|
|
"""
|
|
clauses: list[ColumnElement[bool]] = [
|
|
Document.visibility == DocumentVisibility.public,
|
|
Document.author_id == user.id,
|
|
]
|
|
if user.department_id is not None:
|
|
clauses.append(
|
|
and_(
|
|
Document.visibility == DocumentVisibility.department,
|
|
Document.department_id == user.department_id,
|
|
)
|
|
)
|
|
clauses.append(
|
|
exists(
|
|
select(DocPermission.document_id).where(
|
|
DocPermission.document_id == Document.id,
|
|
DocPermission.department_id == user.department_id,
|
|
)
|
|
)
|
|
)
|
|
return and_(Document.status == DocumentStatus.published, or_(*clauses))
|
|
|
|
|
|
def readable_documents_filter(user: User) -> ColumnElement[bool]:
|
|
"""Searchable documents plus the unpublished ones this user owns or was
|
|
asked to check.
|
|
|
|
Being asked IS the grant: a reviewer must be able to open the draft they
|
|
were pointed at. The clause lives here only, never in
|
|
`searchable_documents_filter`, so an unpublished document still never
|
|
reaches chat/search retrieval."""
|
|
return or_(
|
|
searchable_documents_filter(user),
|
|
Document.author_id == user.id,
|
|
open_review_for(user),
|
|
)
|
|
|
|
|
|
def open_review_for(user: User) -> ColumnElement[bool]:
|
|
"""This user has an unanswered request to check the document."""
|
|
return exists(
|
|
select(ReviewRequest.document_id).where(
|
|
ReviewRequest.document_id == Document.id,
|
|
ReviewRequest.reviewer_id == user.id,
|
|
ReviewRequest.resolved_at.is_(None),
|
|
)
|
|
)
|
|
|
|
|
|
def has_open_review() -> ColumnElement[bool]:
|
|
"""Anyone has an unanswered question about the document — what marks it as
|
|
"may not be right yet" wherever it is shown, including chat sources."""
|
|
return exists(
|
|
select(ReviewRequest.document_id).where(
|
|
ReviewRequest.document_id == Document.id,
|
|
ReviewRequest.resolved_at.is_(None),
|
|
)
|
|
)
|
|
|
|
|
|
def document_reader_filter(document: Document) -> ColumnElement[bool]:
|
|
"""A `User`-table filter for who may read `document` AS IF it were
|
|
published — the candidate set for assigning a reviewer. Inverts the read
|
|
rules of `searchable_documents_filter` (author, public, owning department,
|
|
granted departments), ignoring status so a still-pending document can be
|
|
handed to a reviewer who will then be able to see it (the reviewer clause
|
|
in `readable_documents_filter`)."""
|
|
clauses: list[ColumnElement[bool]] = [User.id == document.author_id]
|
|
if document.visibility == DocumentVisibility.public:
|
|
clauses.append(true())
|
|
elif (
|
|
document.visibility == DocumentVisibility.department
|
|
and document.department_id is not None
|
|
):
|
|
clauses.append(User.department_id == document.department_id)
|
|
clauses.append(
|
|
User.department_id.in_(
|
|
select(DocPermission.department_id).where(
|
|
DocPermission.document_id == document.id
|
|
)
|
|
)
|
|
)
|
|
return or_(*clauses)
|