mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-08-03 17:42:20 +00:00
stumpylog
5d8287f120
_handle_profile unconditionally called seed_benchmark_dataset every run, colliding with an IntegrityError against data seeded by a prior seed/profile call. Look up the existing perf_target user instead, matching _handle_run's already-correct pattern, and raise a CommandError pointing at `seed` if no dataset exists. Scenario.run/queryset_for_explain now take a User directly instead of a SeededData, since only data.users[0] (perf_target) was ever used. profile no longer seeds, so it drops the --tier dependency: the printed line and history entry no longer include a tier/(tier=...) suffix, and --tier's help text now only mentions `seed`.
111 lines
3.0 KiB
Python
111 lines
3.0 KiB
Python
# src/paperless_benchmark/scenarios.py
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import TYPE_CHECKING
|
|
from typing import Any
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Callable
|
|
|
|
from django.contrib.auth.models import User
|
|
from django.db.models import QuerySet
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Scenario:
|
|
name: str
|
|
describe: str
|
|
run: Callable[[User], Any]
|
|
queryset_for_explain: Callable[[User], QuerySet] | None = None
|
|
|
|
|
|
_SCENARIOS: dict[str, Scenario] = {}
|
|
|
|
|
|
def register(scenario: Scenario) -> None:
|
|
_SCENARIOS[scenario.name] = scenario
|
|
|
|
|
|
def get(name: str) -> Scenario:
|
|
from django.core.management.base import CommandError
|
|
|
|
try:
|
|
return _SCENARIOS[name]
|
|
except KeyError:
|
|
available = ", ".join(sorted(_SCENARIOS)) or "(none registered)"
|
|
raise CommandError(
|
|
f"Unknown benchmark scenario {name!r}. Available: {available}",
|
|
) from None
|
|
|
|
|
|
def all_scenarios() -> tuple[Scenario, ...]:
|
|
return tuple(_SCENARIOS.values())
|
|
|
|
|
|
def _guardian_visibility_query_run(user: User) -> list[int]:
|
|
from documents.models import Document
|
|
from documents.permissions import get_objects_for_user_owner_aware
|
|
|
|
return list(
|
|
get_objects_for_user_owner_aware(
|
|
user,
|
|
"documents.view_document",
|
|
Document,
|
|
).values_list("id", flat=True),
|
|
)
|
|
|
|
|
|
def _guardian_visibility_query_queryset(user: User) -> QuerySet:
|
|
from documents.models import Document
|
|
from documents.permissions import get_objects_for_user_owner_aware
|
|
|
|
return get_objects_for_user_owner_aware(user, "documents.view_document", Document)
|
|
|
|
|
|
register(
|
|
Scenario(
|
|
name="guardian_visibility_query",
|
|
describe=(
|
|
"Document-visibility queryset for a user with mixed owned/shared "
|
|
"documents -- exercises documents.permissions."
|
|
"get_objects_for_user_owner_aware's guardian permission join."
|
|
),
|
|
run=_guardian_visibility_query_run,
|
|
queryset_for_explain=_guardian_visibility_query_queryset,
|
|
),
|
|
)
|
|
|
|
|
|
def _permitted_document_ids_run(user: User) -> list[int]:
|
|
from documents.models import Document
|
|
from documents.permissions import permitted_document_ids
|
|
|
|
return list(
|
|
Document.objects.filter(id__in=permitted_document_ids(user)).values_list(
|
|
"id",
|
|
flat=True,
|
|
),
|
|
)
|
|
|
|
|
|
def _permitted_document_ids_queryset(user: User) -> QuerySet:
|
|
from documents.models import Document
|
|
from documents.permissions import permitted_document_ids
|
|
|
|
return Document.objects.filter(id__in=permitted_document_ids(user))
|
|
|
|
|
|
register(
|
|
Scenario(
|
|
name="permitted_document_ids",
|
|
describe=(
|
|
"Document-visibility query built from documents.permissions."
|
|
"permitted_document_ids -- the resolved-ID-set alternative to "
|
|
"guardian_visibility_query, for side-by-side comparison."
|
|
),
|
|
run=_permitted_document_ids_run,
|
|
queryset_for_explain=_permitted_document_ids_queryset,
|
|
),
|
|
)
|