Files
pablan/backend/app/rag/permissions.py
ProfessorNovaandClaude Opus 5 784b76baf7 Pablan, as it stands
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
2026-09-04 09:21:37 +02:00

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)