WEBDIENSTE/lasuite-docs
1
0
Fork 0
mirror of https://github.com/suitenumerique/docs.git synced 2026-09-13 13:17:53 +02:00
Code Issues Packages Projects Releases Wiki Activity

♻️(backend) delegate tree, sharing and permissions to Drive

Browse Source 
Documents live in Drive as pointer items sharing the document id, and
Drive is the source of truth for hierarchy, sharing, link reach and
trash. Keeping a local copy of any of it would mean syncing two
authorities, so the Document model drops its tree structure (treebeard)
and access rows entirely and becomes a thin wrapper: a non-persisted
drive_item, fetched per request on behalf of the user, feeds abilities,
user role, link fields and tree data. Fail closed when Drive is
unreachable.

Lists and trees are proxied from Drive; "locally known" documents (the
ones the user created or visited) only back secondary views. Deleting
from Docs pushes to Drive's trash, and idempotent server-to-server
endpoints let Drive propagate delete, restore and purge for whole
subtrees. Public link reach now applies to anonymous visitors too,
including attachments.

DocumentAccess and Invitation are removed (invitations only existed to
become accesses); sharing endpoints are commented, not deleted, so they
can come back when Drive implements AskForAccess.
This commit is contained in:
Nathan Vasse
2026-07-30 15:49:21 +02:00
parent 6d49898fe3
commit 5f5bf1efa2
25 changed files with 1598 additions and 3458 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/core/admin.py
+1 -63
View File
@@ -8,7 +8,6 @@ from django.db import transaction
from django.shortcuts import redirect
from django.utils.translation import gettext_lazy as _
from treebeard.admin import TreeAdmin
from core import models
from core.tasks.user_reconciliation import user_reconciliation_csv_import_job
@@ -164,16 +163,8 @@ class UserReconciliationAdmin(admin.ModelAdmin):
actions = [process_reconciliation]
class DocumentAccessInline(admin.TabularInline):
"""Inline admin class for document accesses."""
autocomplete_fields = ["user"]
model = models.DocumentAccess
extra = 0
@admin.register(models.Document)
class DocumentAdmin(TreeAdmin):
class DocumentAdmin(admin.ModelAdmin):
"""Document admin interface declaration."""
fieldsets = (
@@ -183,76 +174,23 @@ class DocumentAdmin(TreeAdmin):
"fields": (
"id",
"title",
)
},
),
(
_("Permissions"),
{
"fields": (
"creator",
"link_reach",
"link_role",
)
},
),
(
_("Tree structure"),
{
"fields": (
"path",
"depth",
"numchild",
"duplicated_from",
"attachments",
)
},
),
)
inlines = (DocumentAccessInline,)
list_display = (
"id",
"title",
"link_reach",
"link_role",
"created_at",
"updated_at",
)
readonly_fields = (
"attachments",
"creator",
"depth",
"duplicated_from",
"id",
"numchild",
"path",
)
search_fields = ("id", "title")
@admin.register(models.Invitation)
class InvitationAdmin(admin.ModelAdmin):
"""Admin interface to handle invitations."""
fields = (
"email",
"document",
"role",
"created_at",
"issuer",
)
readonly_fields = (
"created_at",
"is_expired",
"issuer",
)
list_display = (
"email",
"document",
"created_at",
"is_expired",
)
def save_model(self, request, obj, form, change):
obj.issuer = request.user
obj.save()
src/backend/core/api/permissions.py
+116 -94
View File
@@ -7,7 +7,7 @@ from django.http import Http404
from rest_framework import permissions
from core import choices
from core.models import DocumentAccess, RoleChoices, get_trashbin_cutoff
from core.models import RoleChoices, get_trashbin_cutoff # noqa: F401
ACTION_FOR_METHOD_TO_PERMISSION = {
"versions_detail": {"DELETE": "versions_destroy", "GET": "versions_retrieve"},
@@ -66,122 +66,144 @@ class IsOwnedOrPublic(IsAuthenticated):
return False
class CanCreateInvitationPermission(permissions.BasePermission):
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# class CanCreateInvitationPermission(permissions.BasePermission):
# """
# Custom permission class to handle permission checks for managing invitations.
# """
#
# def has_permission(self, request, view):
# user = request.user
#
# # Ensure the user is authenticated
# if not (bool(request.auth) or request.user.is_authenticated):
# return False
#
# # Apply permission checks only for creation (POST requests)
# if view.action != "create":
# return True
#
# # Check if resource_id is passed in the context
# try:
# document_id = view.kwargs["resource_id"]
# except KeyError as exc:
# raise exceptions.ValidationError(
# "You must set a document ID in kwargs to manage document invitations."
# ) from exc
#
# # Check if the user has access to manage invitations (Owner/Admin roles)
# return DocumentAccess.objects.filter(
# Q(user=user) | Q(team__in=user.teams),
# document=document_id,
# role__in=[RoleChoices.OWNER, RoleChoices.ADMIN],
# ).exists()
#
#
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# class ResourceWithAccessPermission(permissions.BasePermission):
# """A permission class for invitations."""
#
# def has_permission(self, request, view):
# """check create permission."""
# return request.user.is_authenticated or view.action != "create"
#
# def has_object_permission(self, request, view, obj):
# """Check permission for a given object."""
# abilities = obj.get_abilities(request.user)
# action = view.action
# return abilities.get(action, False)
class DriveDelegatedPermission(permissions.BasePermission):
"""
Custom permission class to handle permission checks for managing invitations.
Delegate document permissions to Drive: abilities are computed from the
Drive item mirroring the document, fetched server-to-server on behalf of
the current user.
"""
def has_permission(self, request, view):
user = request.user
"""
Let anonymous users through: link reach is owned by Drive, so the
Drive-derived abilities (all False unless the item is public) are the
actual gate, applied in has_object_permission. List-level actions are
safe: DB-backed lists return nothing for anonymous users and Drive
proxy calls fail closed with a 401/403 from Drive.
"""
return True
# Ensure the user is authenticated
if not (bool(request.auth) or request.user.is_authenticated):
return False
def has_object_permission(self, request, view, obj):
"""Check the action against Drive-derived abilities."""
# Import here to avoid a circular import through core.api.serializers
from core.services import drive_client # pylint: disable=import-outside-toplevel
# Apply permission checks only for creation (POST requests)
if view.action != "create":
return True
# Check if resource_id is passed in the context
try:
document_id = view.kwargs["resource_id"]
except KeyError as exc:
raise exceptions.ValidationError(
"You must set a document ID in kwargs to manage document invitations."
) from exc
item = drive_client.get_item(obj.id, request.user)
except drive_client.DriveClientError as exc:
if exc.status_code in (403, 404):
return False
drive_client.raise_as_drf(exc)
# Check if the user has access to manage invitations (Owner/Admin roles)
return DocumentAccess.objects.filter(
Q(user=user) | Q(team__in=user.teams),
document=document_id,
role__in=[RoleChoices.OWNER, RoleChoices.ADMIN],
).exists()
# The document is a wrapper around the Drive item: hydrate it so
# abilities and link data flow from the instance everywhere downstream.
obj.drive_item = item
abilities = drive_client.map_drive_abilities(item.get("abilities"))
class ResourceWithAccessPermission(permissions.BasePermission):
"""A permission class for invitations."""
def has_permission(self, request, view):
"""check create permission."""
return request.user.is_authenticated or view.action != "create"
def has_object_permission(self, request, view, obj):
"""Check permission for a given object."""
abilities = obj.get_abilities(request.user)
action = view.action
return abilities.get(action, False)
class DocumentPermission(permissions.BasePermission):
"""Subclass to handle soft deletion specificities."""
def has_permission(self, request, view):
"""check create permission for documents."""
return request.user.is_authenticated or view.action != "create"
def has_object_permission(self, request, view, obj):
"""
Return a 404 on deleted documents
- for which the trashbin cutoff is past
- for which the current user is not owner of the document or one of its ancestors
"""
if (
deleted_at := obj.ancestors_deleted_at
) and deleted_at < get_trashbin_cutoff():
raise Http404
abilities = obj.get_abilities(request.user)
action = view.action
try:
action = ACTION_FOR_METHOD_TO_PERMISSION[view.action][request.method]
except KeyError:
pass
has_permission = abilities.get(action, False)
if obj.ancestors_deleted_at and not RoleChoices.OWNER in obj.user_roles:
raise Http404
return has_permission
class ResourceAccessPermission(IsAuthenticated):
"""Permission class for document access objects."""
def has_permission(self, request, view):
"""check create permission for accesses in documents tree."""
if super().has_permission(request, view) is False:
return False
if view.action == "create":
role = getattr(view, view.resource_field_name).get_role(request.user)
if role not in choices.PRIVILEGED_ROLES:
raise exceptions.PermissionDenied(
"You are not allowed to manage accesses for this resource."
)
return True
def has_object_permission(self, request, view, obj):
"""Check permission for a given object."""
abilities = obj.get_abilities(request.user)
requested_role = request.data.get("role")
if requested_role and requested_role not in abilities.get("set_role_to", []):
return False
action = view.action
return abilities.get(action, False)
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# class ResourceAccessPermission(IsAuthenticated):
# """Permission class for document access objects."""
#
# def has_permission(self, request, view):
# """check create permission for accesses in documents tree."""
# if super().has_permission(request, view) is False:
# return False
#
# if view.action == "create":
# role = getattr(view, view.resource_field_name).get_role(request.user)
# if role not in choices.PRIVILEGED_ROLES:
# raise exceptions.PermissionDenied(
# "You are not allowed to manage accesses for this resource."
# )
#
# return True
#
# def has_object_permission(self, request, view, obj):
# """Check permission for a given object."""
# abilities = obj.get_abilities(request.user)
#
# requested_role = request.data.get("role")
# if requested_role and requested_role not in abilities.get("set_role_to", []):
# return False
#
# action = view.action
# return abilities.get(action, False)
class CommentPermission(permissions.BasePermission):
"""Permission class for comments."""
"""
Permission class for comments. Abilities are delegated to Drive, which
owns document sharing.
"""
def has_permission(self, request, view):
"""Check permission for a given object."""
if view.action in ["create", "list"]:
document_abilities = view.get_document_or_404().get_abilities(request.user)
return document_abilities["comment"]
# Import here to avoid a circular import through core.api.serializers
from core.services import ( # pylint: disable=import-outside-toplevel
drive_client,
)
document = view.get_document_or_404()
abilities, _role = drive_client.get_doc_context(document.id, request.user)
return abilities["comment"]
return True
src/backend/core/api/serializers.py
+398 -320
View File
@@ -17,14 +17,13 @@ import magic
from rest_framework import serializers
from core import choices, enums, models, validators
from core.services import mime_types
from core.services import drive_client, mime_types
from core.services.ai_services.legacy import AI_ACTIONS
from core.services.converter_services import (
ConversionError,
Converter,
)
from core.utils.analytics import PosthogEventName, posthog_capture
from core.utils.treebeard import create_tree_node_with_retry
class UserSerializer(serializers.ModelSerializer):
@@ -81,11 +80,17 @@ class ListDocumentSerializer(serializers.ModelSerializer):
"""Serialize documents with limited fields for display in lists."""
is_favorite = serializers.BooleanField(read_only=True)
nb_accesses_ancestors = serializers.IntegerField(read_only=True)
nb_accesses_direct = serializers.IntegerField(read_only=True)
nb_accesses_ancestors = serializers.SerializerMethodField(read_only=True)
nb_accesses_direct = serializers.SerializerMethodField(read_only=True)
user_role = serializers.SerializerMethodField(read_only=True)
abilities = serializers.SerializerMethodField(read_only=True)
deleted_at = serializers.SerializerMethodField(read_only=True)
# The hierarchy and sharing are owned by Drive: these keys are kept for
# the payload shape and sourced from the document's drive_item.
path = serializers.SerializerMethodField(read_only=True)
depth = serializers.SerializerMethodField(read_only=True)
numchild = serializers.SerializerMethodField(read_only=True)
excerpt = serializers.SerializerMethodField(read_only=True)
class Meta:
model = models.Document
@@ -136,19 +141,61 @@ class ListDocumentSerializer(serializers.ModelSerializer):
]
def to_representation(self, instance):
"""Precompute once per instance"""
paths_links_mapping = self.context.get("paths_links_mapping")
"""
Hydrate the document's drive_item once so every field (abilities, link
reach/role properties, hierarchy data) reads from it.
if paths_links_mapping is not None:
links = paths_links_mapping.get(instance.path[: -instance.steplen], [])
instance.ancestors_link_definition = choices.get_equivalent_link_definition(
links
)
Note (POC): DB-backed lists (favorites, search) trigger one Drive
fetch per row, amortized by drive_client's short per-user cache.
"""
request = self.context.get("request")
if request is not None:
try:
instance.get_drive_item(request.user)
except drive_client.DriveClientError:
pass # fields fall back to safe defaults
return super().to_representation(instance)
def _get_drive_item(self, instance):
"""Return the document's hydrated Drive item, if available."""
return instance.drive_item
def get_path(self, instance):
"""Return the Drive item path, or the document id as a degenerate path."""
item = self._get_drive_item(instance)
return item["path"] if item and item.get("path") else str(instance.pk)
def get_depth(self, instance):
"""Return the depth in the Drive tree, defaulting to root depth."""
item = self._get_drive_item(instance)
if item and item.get("path"):
return len(str(item["path"]).split("."))
return 1
def get_numchild(self, instance):
"""Return the number of children as known by Drive."""
item = self._get_drive_item(instance)
return item.get("numchild", 0) if item else 0
def get_excerpt(self, _instance):
"""Excerpts are not supported anymore."""
return None
def get_nb_accesses_direct(self, instance):
"""Number of accesses, as known by Drive."""
item = self._get_drive_item(instance)
return item.get("nb_accesses", 0) if item else 0
def get_nb_accesses_ancestors(self, instance):
"""Number of accesses including ancestors, as known by Drive."""
return self.get_nb_accesses_direct(instance)
def get_abilities(self, instance) -> dict:
"""Return abilities of the logged-in user on the instance."""
"""
Return abilities of the logged-in user on the instance, as delegated
to Drive (which owns the document tree and sharing).
"""
request = self.context.get("request")
if not request:
return {}
@@ -157,25 +204,36 @@ class ListDocumentSerializer(serializers.ModelSerializer):
def get_user_role(self, instance):
"""
Return roles of the logged-in user for the current document,
taking into account ancestors.
Return the role of the logged-in user for the current document, as
known by Drive (which owns sharing).
"""
request = self.context.get("request")
return instance.get_role(request.user) if request else None
item = self._get_drive_item(instance)
return item.get("user_role") if item else None
def get_deleted_at(self, instance):
"""Return the deleted_at of the current document."""
return instance.ancestors_deleted_at
return instance.deleted_at
class DocumentLightSerializer(serializers.ModelSerializer):
"""Minial document serializer for nesting in document accesses."""
path = serializers.SerializerMethodField(read_only=True)
depth = serializers.SerializerMethodField(read_only=True)
class Meta:
model = models.Document
fields = ["id", "path", "depth"]
read_only_fields = ["id", "path", "depth"]
def get_path(self, instance):
"""The hierarchy is owned by Drive: degenerate single-node path."""
return str(instance.pk)
def get_depth(self, _instance):
"""The hierarchy is owned by Drive: all local documents are roots."""
return 1
class DocumentSerializer(ListDocumentSerializer):
"""Serialize documents with all fields for display in detail views."""
@@ -336,97 +394,99 @@ class DocumentContentSerializer(serializers.Serializer):
raise NotImplementedError("Create is not supported for this serializer.")
class DocumentAccessSerializer(serializers.ModelSerializer):
"""Serialize document accesses."""
document = DocumentLightSerializer(read_only=True)
user_id = serializers.PrimaryKeyRelatedField(
queryset=models.User.objects.all(),
write_only=True,
source="user",
required=False,
allow_null=True,
)
user = UserSerializer(read_only=True)
team = serializers.CharField(required=False, allow_blank=True)
abilities = serializers.SerializerMethodField(read_only=True)
max_ancestors_role = serializers.SerializerMethodField(read_only=True)
max_role = serializers.SerializerMethodField(read_only=True)
class Meta:
model = models.DocumentAccess
resource_field_name = "document"
fields = [
"id",
"document",
"user",
"user_id",
"team",
"role",
"abilities",
"max_ancestors_role",
"max_role",
]
read_only_fields = [
"id",
"document",
"abilities",
"max_ancestors_role",
"max_role",
]
def get_abilities(self, instance) -> dict:
"""Return abilities of the logged-in user on the instance."""
request = self.context.get("request")
if request:
return instance.get_abilities(request.user)
return {}
def get_max_ancestors_role(self, instance):
"""Return max_ancestors_role if annotated; else None."""
return getattr(instance, "max_ancestors_role", None)
def get_max_role(self, instance):
"""Return max_ancestors_role if annotated; else None."""
return choices.RoleChoices.max(
getattr(instance, "max_ancestors_role", None),
instance.role,
)
def update(self, instance, validated_data):
"""Make "user" field readonly but only on update."""
validated_data.pop("team", None)
validated_data.pop("user", None)
return super().update(instance, validated_data)
class DocumentAccessLightSerializer(DocumentAccessSerializer):
"""Serialize document accesses with limited fields."""
user = UserLightSerializer(read_only=True)
class Meta:
model = models.DocumentAccess
resource_field_name = "document"
fields = [
"id",
"document",
"user",
"team",
"role",
"abilities",
"max_ancestors_role",
"max_role",
]
read_only_fields = [
"id",
"document",
"team",
"role",
"abilities",
"max_ancestors_role",
"max_role",
]
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# class DocumentAccessSerializer(serializers.ModelSerializer):
# """Serialize document accesses."""
#
# document = DocumentLightSerializer(read_only=True)
# user_id = serializers.PrimaryKeyRelatedField(
# queryset=models.User.objects.all(),
# write_only=True,
# source="user",
# required=False,
# allow_null=True,
# )
# user = UserSerializer(read_only=True)
# team = serializers.CharField(required=False, allow_blank=True)
# abilities = serializers.SerializerMethodField(read_only=True)
# max_ancestors_role = serializers.SerializerMethodField(read_only=True)
# max_role = serializers.SerializerMethodField(read_only=True)
#
# class Meta:
# model = models.DocumentAccess
# resource_field_name = "document"
# fields = [
# "id",
# "document",
# "user",
# "user_id",
# "team",
# "role",
# "abilities",
# "max_ancestors_role",
# "max_role",
# ]
# read_only_fields = [
# "id",
# "document",
# "abilities",
# "max_ancestors_role",
# "max_role",
# ]
#
# def get_abilities(self, instance) -> dict:
# """Return abilities of the logged-in user on the instance."""
# request = self.context.get("request")
# if request:
# return instance.get_abilities(request.user)
# return {}
#
# def get_max_ancestors_role(self, instance):
# """Return max_ancestors_role if annotated; else None."""
# return getattr(instance, "max_ancestors_role", None)
#
# def get_max_role(self, instance):
# """Return max_ancestors_role if annotated; else None."""
# return choices.RoleChoices.max(
# getattr(instance, "max_ancestors_role", None),
# instance.role,
# )
#
# def update(self, instance, validated_data):
# """Make "user" field readonly but only on update."""
# validated_data.pop("team", None)
# validated_data.pop("user", None)
# return super().update(instance, validated_data)
#
#
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# class DocumentAccessLightSerializer(DocumentAccessSerializer):
# """Serialize document accesses with limited fields."""
#
# user = UserLightSerializer(read_only=True)
#
# class Meta:
# model = models.DocumentAccess
# resource_field_name = "document"
# fields = [
# "id",
# "document",
# "user",
# "team",
# "role",
# "abilities",
# "max_ancestors_role",
# "max_role",
# ]
# read_only_fields = [
# "id",
# "document",
# "team",
# "role",
# "abilities",
# "max_ancestors_role",
# "max_role",
# ]
class ServerCreateDocumentSerializer(serializers.Serializer):
@@ -443,8 +503,11 @@ class ServerCreateDocumentSerializer(serializers.Serializer):
"""
# Document
# Optional client-supplied id, used by Drive to make the document reuse
# the id of the item pointing to it.
id = serializers.UUIDField(required=False)
title = serializers.CharField(required=True)
content = serializers.CharField(required=True)
content = serializers.CharField(required=False, allow_blank=True, default="")
# User
sub = serializers.CharField(
required=True, validators=[validators.sub_validator], max_length=255
@@ -457,6 +520,14 @@ class ServerCreateDocumentSerializer(serializers.Serializer):
message = serializers.CharField(required=False)
subject = serializers.CharField(required=False)
def validate_id(self, value):
"""Ensure the provided ID is not already taken."""
if models.Document.objects.filter(id=value).exists():
raise serializers.ValidationError(
"A document with this ID already exists. You cannot override it."
)
return value
def create(self, validated_data):
"""Create the document and associate it with the user or send an invitation."""
language = validated_data.get("language", settings.LANGUAGE_CODE)
@@ -475,20 +546,25 @@ class ServerCreateDocumentSerializer(serializers.Serializer):
email = user.email
language = user.language or language
try:
document_content = Converter().convert(
validated_data["content"], mime_types.MARKDOWN, mime_types.YJS
)
except ConversionError as err:
raise serializers.ValidationError(
{"content": ["Could not convert content"]}
) from err
document_content = None
if validated_data.get("content"):
try:
document_content = Converter().convert(
validated_data["content"], mime_types.MARKDOWN, mime_types.YJS
)
except ConversionError as err:
raise serializers.ValidationError(
{"content": ["Could not convert content"]}
) from err
document = create_tree_node_with_retry(
lambda: models.Document.add_root(
title=validated_data["title"],
creator=user,
)
extra_document_fields = {}
if validated_data.get("id"):
extra_document_fields["id"] = validated_data["id"]
document = models.Document.objects.create(
title=validated_data["title"],
creator=user,
**extra_document_fields,
)
posthog_capture(PosthogEventName.DOC_CREATED, user, {}, document=document)
@@ -502,25 +578,18 @@ class ServerCreateDocumentSerializer(serializers.Serializer):
document=document,
)
if user:
# Associate the document with the pre-existing user
models.DocumentAccess.objects.create(
document=document,
role=models.RoleChoices.OWNER,
user=user,
)
else:
# The user doesn't exist in our database: we need to invite him/her
models.Invitation.objects.create(
document=document,
email=email,
role=models.RoleChoices.OWNER,
)
# Sharing is owned by Drive: no local access row is created. When the
# user is not known locally yet (creator=None), Drive still grants
# access and DriveDelegatedPermission lets them in on first visit.
document.content = document_content
document.save()
if document_content is not None:
document.content = document_content
document.save()
self._send_email_notification(document, validated_data, email, language)
# Back-channel creations from Drive (id provided) are silent: the user
# initiated the creation themselves from the Drive UI.
if not validated_data.get("id"):
self._send_email_notification(document, validated_data, email, language)
return document
def _send_email_notification(self, document, validated_data, email, language):
@@ -542,74 +611,79 @@ class ServerCreateDocumentSerializer(serializers.Serializer):
raise NotImplementedError("Update is not supported for this serializer.")
class LinkDocumentSerializer(serializers.ModelSerializer):
"""
Serialize link configuration for documents.
We expose it separately from document in order to simplify and secure access control.
"""
link_reach = serializers.ChoiceField(
choices=models.LinkReachChoices.choices, required=True
)
class Meta:
model = models.Document
fields = [
"link_role",
"link_reach",
]
def validate(self, attrs):
"""Validate that link_role and link_reach are compatible using get_select_options."""
link_reach = attrs.get("link_reach")
link_role = attrs.get("link_role")
if not link_reach:
raise serializers.ValidationError(
{"link_reach": _("This field is required.")}
)
# Get available options based on ancestors' link definition
available_options = models.LinkReachChoices.get_select_options(
**self.instance.ancestors_link_definition
)
# Validate link_reach is allowed
if link_reach not in available_options:
msg = _(
"Link reach '%(link_reach)s' is not allowed based on parent document configuration."
)
raise serializers.ValidationError(
{"link_reach": msg % {"link_reach": link_reach}}
)
# Validate link_role is compatible with link_reach
allowed_roles = available_options[link_reach]
# Restricted reach: link_role must be None
if link_reach == models.LinkReachChoices.RESTRICTED:
if link_role is not None:
raise serializers.ValidationError(
{
"link_role": (
"Cannot set link_role when link_reach is 'restricted'. "
"Link role must be null for restricted reach."
)
}
)
return attrs
# Non-restricted: link_role must be in allowed roles
if link_role not in allowed_roles:
allowed_roles_str = ", ".join(allowed_roles) if allowed_roles else "none"
raise serializers.ValidationError(
{
"link_role": (
f"Link role '{link_role}' is not allowed for link reach '{link_reach}'. "
f"Allowed roles: {allowed_roles_str}"
)
}
)
return attrs
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# class LinkDocumentSerializer(serializers.ModelSerializer):
# """
# Serialize link configuration for documents.
# We expose it separately from document in order to simplify and secure access control.
# """
#
# link_reach = serializers.ChoiceField(
# choices=models.LinkReachChoices.choices, required=True
# )
# # Link configuration is owned by Drive; kept for payload compatibility only.
# link_role = serializers.ChoiceField(
# choices=models.LinkRoleChoices.choices, required=False, allow_null=True
# )
#
# class Meta:
# model = models.Document
# fields = [
# "link_role",
# "link_reach",
# ]
#
# def validate(self, attrs):
# """Validate that link_role and link_reach are compatible using get_select_options."""
# link_reach = attrs.get("link_reach")
# link_role = attrs.get("link_role")
#
# if not link_reach:
# raise serializers.ValidationError(
# {"link_reach": _("This field is required.")}
# )
#
# # Get available options based on ancestors' link definition
# available_options = models.LinkReachChoices.get_select_options(
# **self.instance.ancestors_link_definition
# )
#
# # Validate link_reach is allowed
# if link_reach not in available_options:
# msg = _(
# "Link reach '%(link_reach)s' is not allowed based on parent document configuration."
# )
# raise serializers.ValidationError(
# {"link_reach": msg % {"link_reach": link_reach}}
# )
#
# # Validate link_role is compatible with link_reach
# allowed_roles = available_options[link_reach]
#
# # Restricted reach: link_role must be None
# if link_reach == models.LinkReachChoices.RESTRICTED:
# if link_role is not None:
# raise serializers.ValidationError(
# {
# "link_role": (
# "Cannot set link_role when link_reach is 'restricted'. "
# "Link role must be null for restricted reach."
# )
# }
# )
# return attrs
# # Non-restricted: link_role must be in allowed roles
# if link_role not in allowed_roles:
# allowed_roles_str = ", ".join(allowed_roles) if allowed_roles else "none"
# raise serializers.ValidationError(
# {
# "link_role": (
# f"Link role '{link_role}' is not allowed for link reach '{link_reach}'. "
# f"Allowed roles: {allowed_roles_str}"
# )
# }
# )
# return attrs
class DocumentDuplicationSerializer(serializers.Serializer):
@@ -695,119 +769,123 @@ class FileUploadSerializer(serializers.Serializer):
return attrs
class InvitationSerializer(serializers.ModelSerializer):
"""Serialize invitations."""
abilities = serializers.SerializerMethodField(read_only=True)
class Meta:
model = models.Invitation
fields = [
"id",
"abilities",
"created_at",
"email",
"document",
"role",
"issuer",
"is_expired",
]
read_only_fields = [
"id",
"abilities",
"created_at",
"document",
"issuer",
"is_expired",
]
def get_abilities(self, invitation) -> dict:
"""Return abilities of the logged-in user on the instance."""
request = self.context.get("request")
if request:
return invitation.get_abilities(request.user)
return {}
def validate(self, attrs):
"""Validate invitation data."""
request = self.context.get("request")
user = getattr(request, "user", None)
attrs["document_id"] = self.context["resource_id"]
# Only set the issuer if the instance is being created
if self.instance is None:
attrs["issuer"] = user
if attrs.get("email"):
attrs["email"] = attrs["email"].lower()
return attrs
def validate_role(self, role):
"""Custom validation for the role field."""
request = self.context.get("request")
user = getattr(request, "user", None)
document_id = self.context["resource_id"]
# If the role is OWNER, check if the user has OWNER access
if role == models.RoleChoices.OWNER:
if not models.DocumentAccess.objects.filter(
Q(user=user) | Q(team__in=user.teams),
document=document_id,
role=models.RoleChoices.OWNER,
).exists():
raise serializers.ValidationError(
"Only owners of a document can invite other users as owners."
)
return role
class RoleSerializer(serializers.Serializer):
"""Serializer validating role choices."""
role = serializers.ChoiceField(
choices=models.RoleChoices.choices, required=False, allow_null=True
)
class DocumentAskForAccessCreateSerializer(serializers.Serializer):
"""Serializer for creating a document ask for access."""
role = serializers.ChoiceField(
choices=[
role for role in choices.RoleChoices if role != models.RoleChoices.OWNER
],
required=False,
default=models.RoleChoices.READER,
)
class DocumentAskForAccessSerializer(serializers.ModelSerializer):
"""Serializer for document ask for access model"""
abilities = serializers.SerializerMethodField(read_only=True)
user = UserSerializer(read_only=True)
class Meta:
model = models.DocumentAskForAccess
fields = [
"id",
"document",
"user",
"role",
"created_at",
"abilities",
]
read_only_fields = ["id", "document", "user", "role", "created_at", "abilities"]
def get_abilities(self, instance) -> dict:
"""Return abilities of the logged-in user on the instance."""
request = self.context.get("request")
if request:
return instance.get_abilities(request.user)
return {}
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# class InvitationSerializer(serializers.ModelSerializer):
# """Serialize invitations."""
#
# abilities = serializers.SerializerMethodField(read_only=True)
#
# class Meta:
# model = models.Invitation
# fields = [
# "id",
# "abilities",
# "created_at",
# "email",
# "document",
# "role",
# "issuer",
# "is_expired",
# ]
# read_only_fields = [
# "id",
# "abilities",
# "created_at",
# "document",
# "issuer",
# "is_expired",
# ]
#
# def get_abilities(self, invitation) -> dict:
# """Return abilities of the logged-in user on the instance."""
# request = self.context.get("request")
# if request:
# return invitation.get_abilities(request.user)
# return {}
#
# def validate(self, attrs):
# """Validate invitation data."""
# request = self.context.get("request")
# user = getattr(request, "user", None)
#
# attrs["document_id"] = self.context["resource_id"]
#
# # Only set the issuer if the instance is being created
# if self.instance is None:
# attrs["issuer"] = user
#
# if attrs.get("email"):
# attrs["email"] = attrs["email"].lower()
#
# return attrs
#
# def validate_role(self, role):
# """Custom validation for the role field."""
# request = self.context.get("request")
# user = getattr(request, "user", None)
# document_id = self.context["resource_id"]
#
# # If the role is OWNER, check if the user has OWNER access
# if role == models.RoleChoices.OWNER:
# if not models.DocumentAccess.objects.filter(
# Q(user=user) | Q(team__in=user.teams),
# document=document_id,
# role=models.RoleChoices.OWNER,
# ).exists():
# raise serializers.ValidationError(
# "Only owners of a document can invite other users as owners."
# )
#
# return role
#
#
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# class RoleSerializer(serializers.Serializer):
# """Serializer validating role choices."""
#
# role = serializers.ChoiceField(
# choices=models.RoleChoices.choices, required=False, allow_null=True
# )
#
#
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# class DocumentAskForAccessCreateSerializer(serializers.Serializer):
# """Serializer for creating a document ask for access."""
#
# role = serializers.ChoiceField(
# choices=[
# role for role in choices.RoleChoices if role != models.RoleChoices.OWNER
# ],
# required=False,
# default=models.RoleChoices.READER,
# )
#
#
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# class DocumentAskForAccessSerializer(serializers.ModelSerializer):
# """Serializer for document ask for access model"""
#
# abilities = serializers.SerializerMethodField(read_only=True)
# user = UserSerializer(read_only=True)
#
# class Meta:
# model = models.DocumentAskForAccess
# fields = [
# "id",
# "document",
# "user",
# "role",
# "created_at",
# "abilities",
# ]
# read_only_fields = ["id", "document", "user", "role", "created_at", "abilities"]
#
# def get_abilities(self, instance) -> dict:
# """Return abilities of the logged-in user on the instance."""
# request = self.context.get("request")
# if request:
# return instance.get_abilities(request.user)
# return {}
class VersionFilterSerializer(serializers.Serializer):
src/backend/core/api/utils.py
-52
View File
@@ -14,58 +14,6 @@ from lasuite.oidc_login.decorators import refresh_oidc_access_token
from rest_framework.throttling import BaseThrottle
def nest_tree(flat_list, steplen):
"""
Convert a flat list of serialized documents into a nested tree making advantage
of the`path` field and its step length.
"""
node_dict = {}
roots = []
# Sort the flat list by path to ensure parent nodes are processed first
flat_list.sort(key=lambda x: x["path"])
for node in flat_list:
node["children"] = [] # Initialize children list
node_dict[node["path"]] = node
# Determine parent path
parent_path = node["path"][:-steplen]
if parent_path in node_dict:
node_dict[parent_path]["children"].append(node)
else:
roots.append(node) # Collect root nodes
if len(roots) > 1:
raise ValueError("More than one root element detected.")
return roots[0] if roots else None
def filter_root_paths(paths, skip_sorting=False):
"""
Filters root paths from a list of paths representing a tree structure.
A root path is defined as a path that is not a prefix of any other path.
Args:
paths (list of str): The list of paths.
Returns:
list of str: The filtered list of root paths.
"""
if not skip_sorting:
paths.sort()
root_paths = []
for path in paths:
# If the current path is not a prefix of the last added root path, add it
if not root_paths or not path.startswith(root_paths[-1]):
root_paths.append(path)
return root_paths
def generate_s3_authorization_headers(key):
"""
Generate authorization headers for an s3 object.
src/backend/core/api/viewsets.py
+706 -1222
View File
File diff suppressed because it is too large Load Diff
src/backend/core/external_api/viewsets.py
+33 -35
View File
@@ -5,15 +5,11 @@ from django.conf import settings
from lasuite.oidc_resource_server.authentication import ResourceServerAuthentication
from core.api.permissions import (
CanCreateInvitationPermission,
DocumentPermission,
DriveDelegatedPermission,
IsSelf,
ResourceAccessPermission,
)
from core.api.viewsets import (
DocumentAccessViewSet,
DocumentViewSet,
InvitationViewset,
UserViewSet,
)
from core.external_api.permissions import ResourceServerClientPermission
@@ -34,11 +30,11 @@ class ResourceServerRestrictionMixin:
class ResourceServerDocumentViewSet(ResourceServerRestrictionMixin, DocumentViewSet):
"""Resource Server Viewset for Documents."""
"""Resource Server Viewset for Documents, abilities delegated to Drive."""
authentication_classes = [ResourceServerAuthentication]
permission_classes = [ResourceServerClientPermission & DocumentPermission] # type: ignore
permission_classes = [ResourceServerClientPermission & DriveDelegatedPermission] # type: ignore
@property
def resource_server_actions(self):
@@ -46,36 +42,38 @@ class ResourceServerDocumentViewSet(ResourceServerRestrictionMixin, DocumentView
return self._get_resource_server_actions("documents")
class ResourceServerDocumentAccessViewSet(
ResourceServerRestrictionMixin, DocumentAccessViewSet
):
"""Resource Server Viewset for DocumentAccess."""
authentication_classes = [ResourceServerAuthentication]
permission_classes = [ResourceServerClientPermission & ResourceAccessPermission] # type: ignore
@property
def resource_server_actions(self):
"""Get resource_server_actions from settings."""
return self._get_resource_server_actions("document_access")
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# class ResourceServerDocumentAccessViewSet(
# ResourceServerRestrictionMixin, DocumentAccessViewSet
# ):
# """Resource Server Viewset for DocumentAccess."""
#
# authentication_classes = [ResourceServerAuthentication]
#
# permission_classes = [ResourceServerClientPermission & ResourceAccessPermission]
#
# @property
# def resource_server_actions(self):
# """Get resource_server_actions from settings."""
# return self._get_resource_server_actions("document_access")
class ResourceServerInvitationViewSet(
ResourceServerRestrictionMixin, InvitationViewset
):
"""Resource Server Viewset for Invitations."""
authentication_classes = [ResourceServerAuthentication]
permission_classes = [
ResourceServerClientPermission & CanCreateInvitationPermission
]
@property
def resource_server_actions(self):
"""Get resource_server_actions from settings."""
return self._get_resource_server_actions("document_invitation")
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# class ResourceServerInvitationViewSet(
# ResourceServerRestrictionMixin, InvitationViewset
# ):
# """Resource Server Viewset for Invitations."""
#
# authentication_classes = [ResourceServerAuthentication]
#
# permission_classes = [
# ResourceServerClientPermission & CanCreateInvitationPermission
# ]
#
# @property
# def resource_server_actions(self):
# """Get resource_server_actions from settings."""
# return self._get_resource_server_actions("document_invitation")
class ResourceServerUserViewSet(ResourceServerRestrictionMixin, UserViewSet):
src/backend/core/factories.py
+3 -91
View File
@@ -44,16 +44,6 @@ class UserFactory(factory.django.DjangoModelFactory):
language = factory.fuzzy.FuzzyChoice([lang[0] for lang in settings.LANGUAGES])
password = make_password("password")
@factory.post_generation
def with_owned_document(self, create, extracted, **kwargs):
"""
Create a document for which the user is owner to check
that there is no interference
"""
if create and (extracted is True):
UserDocumentAccessFactory(user=self, role="owner")
class ParentNodeFactory(factory.declarations.ParameteredAttribute):
"""Custom factory attribute for setting the parent node."""
@@ -82,59 +72,15 @@ class DocumentFactory(factory.django.DjangoModelFactory):
parent = ParentNodeFactory()
title = factory.Sequence(lambda n: f"document{n}")
excerpt = factory.Sequence(lambda n: f"excerpt{n}")
content = YDOC_HELLO_WORLD_BASE64
creator = factory.SubFactory(UserFactory)
deleted_at = None
link_reach = factory.fuzzy.FuzzyChoice(
[a[0] for a in models.LinkReachChoices.choices]
)
link_role = factory.fuzzy.FuzzyChoice(
[r[0] for r in models.LinkRoleChoices.choices]
)
@classmethod
def _create(cls, model_class, *args, **kwargs):
"""
Custom creation logic for the factory: creates a document as a child node if
a parent is provided; otherwise, creates it as a root node.
"""
parent = kwargs.pop("parent", None)
if parent:
# Add as a child node
kwargs["ancestors_deleted_at"] = (
kwargs.get("ancestors_deleted_at") or parent.ancestors_deleted_at
)
return parent.add_child(instance=model_class(**kwargs))
# Add as a root node
return model_class.add_root(instance=model_class(**kwargs))
@factory.lazy_attribute
def ancestors_deleted_at(self):
"""Should always be set when "deleted_at" is set."""
return self.deleted_at
@factory.post_generation
def users(self, create, extracted, **kwargs):
"""Add users to document from a given list of users with or without roles."""
if create and extracted:
for item in extracted:
if isinstance(item, models.User):
UserDocumentAccessFactory(document=self, user=item)
else:
UserDocumentAccessFactory(document=self, user=item[0], role=item[1])
@factory.post_generation
def teams(self, create, extracted, **kwargs):
"""Add teams to document from a given list of teams with or without roles."""
if create and extracted:
for item in extracted:
if isinstance(item, str):
TeamDocumentAccessFactory(document=self, team=item)
else:
TeamDocumentAccessFactory(document=self, team=item[0], role=item[1])
"""The hierarchy is owned by Drive: documents are plain rows."""
kwargs.pop("parent", None)
return model_class.objects.create(**kwargs)
@factory.post_generation
def link_traces(self, create, extracted, **kwargs):
@@ -151,28 +97,6 @@ class DocumentFactory(factory.django.DjangoModelFactory):
models.DocumentFavorite.objects.create(document=self, user=item)
class UserDocumentAccessFactory(factory.django.DjangoModelFactory):
"""Create fake document user accesses for testing."""
class Meta:
model = models.DocumentAccess
document = factory.SubFactory(DocumentFactory)
user = factory.SubFactory(UserFactory)
role = factory.fuzzy.FuzzyChoice([r[0] for r in models.RoleChoices.choices])
class TeamDocumentAccessFactory(factory.django.DjangoModelFactory):
"""Create fake document team accesses for testing."""
class Meta:
model = models.DocumentAccess
document = factory.SubFactory(DocumentFactory)
team = factory.Sequence(lambda n: f"team{n}")
role = factory.fuzzy.FuzzyChoice([r[0] for r in models.RoleChoices.choices])
class DocumentAskForAccessFactory(factory.django.DjangoModelFactory):
"""Create fake document ask for access for testing."""
@@ -184,18 +108,6 @@ class DocumentAskForAccessFactory(factory.django.DjangoModelFactory):
role = factory.fuzzy.FuzzyChoice([r[0] for r in models.RoleChoices.choices])
class InvitationFactory(factory.django.DjangoModelFactory):
"""A factory to create invitations for a user"""
class Meta:
model = models.Invitation
email = factory.Faker("email")
document = factory.SubFactory(DocumentFactory)
role = factory.fuzzy.FuzzyChoice([role[0] for role in models.RoleChoices.choices])
issuer = factory.SubFactory(UserFactory)
class ThreadFactory(factory.django.DjangoModelFactory):
"""A factory to create threads for a document"""
src/backend/core/management/commands/clean_document.py
+5 -15
View File
@@ -13,10 +13,8 @@ from botocore.exceptions import ClientError
from core.choices import LinkReachChoices, LinkRoleChoices, RoleChoices
from core.models import (
Document,
DocumentAccess,
DocumentAskForAccess,
DocumentFavorite,
Invitation,
LinkTrace,
Thread,
)
@@ -80,9 +78,10 @@ class Command(BaseCommand):
except (Document.DoesNotExist, ValueError) as err:
raise CommandError(f"Document {document_id} does not exist.") from err
descendants = list(document.get_descendants())
descendant_ids = [doc.id for doc in descendants]
all_documents = [document, *descendants]
# The hierarchy is owned by Drive: no local descendants anymore.
descendants = []
descendant_ids = []
all_documents = [document]
# Collect all attachment keys before the transaction clears them
all_attachment_keys = []
@@ -101,11 +100,7 @@ class Command(BaseCommand):
# update() so the post_save signal fires (search re-indexation) and
# `updated_at` is refreshed. All descendants are about to be deleted,
# so the document can no longer have any deleted child either.
document.excerpt = None
document.link_reach = options["link_reach"]
document.link_role = options["link_role"]
document.attachments = []
document.has_deleted_children = False
if options["title"] is not None:
document.title = options["title"]
document.save()
@@ -150,13 +145,8 @@ class Command(BaseCommand):
owners), invitations, threads, favorites, link traces and pending
access requests.
"""
access_count, _ = DocumentAccess.objects.filter(
Q(document_id=document.id) & ~Q(role=RoleChoices.OWNER)
).delete()
self.stdout.write(f"Deleted {access_count} access(es) on root document.")
# Sharing is owned by Drive: no local accesses/invitations to clean.
for model, label in (
(Invitation, "invitation"),
(Thread, "thread"),
(DocumentFavorite, "favorite"),
(LinkTrace, "link trace"),
src/backend/core/migrations/0033_alter_document_options_and_more.py
+57
View File
@@ -0,0 +1,57 @@
# Generated by Django 5.2.14 on 2026-07-24 20:45
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('core', '0032_remove_linktrace_is_masked'),
]
operations = [
migrations.AlterModelOptions(
name='document',
options={'ordering': ('-created_at',), 'verbose_name': 'Document', 'verbose_name_plural': 'Documents'},
),
migrations.RemoveConstraint(
model_name='document',
name='check_deleted_at_matches_ancestors_deleted_at_when_set',
),
migrations.RemoveField(
model_name='document',
name='ancestors_deleted_at',
),
migrations.RemoveField(
model_name='document',
name='depth',
),
migrations.RemoveField(
model_name='document',
name='duplicated_from',
),
migrations.RemoveField(
model_name='document',
name='excerpt',
),
migrations.RemoveField(
model_name='document',
name='has_deleted_children',
),
migrations.RemoveField(
model_name='document',
name='link_reach',
),
migrations.RemoveField(
model_name='document',
name='link_role',
),
migrations.RemoveField(
model_name='document',
name='numchild',
),
migrations.RemoveField(
model_name='document',
name='path',
),
]
src/backend/core/migrations/0034_remove_invitation_document_remove_invitation_issuer_and_more.py
+27
View File
@@ -0,0 +1,27 @@
# Generated by Django 5.2.14 on 2026-07-27 13:30
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('core', '0033_alter_document_options_and_more'),
]
operations = [
migrations.RemoveField(
model_name='invitation',
name='document',
),
migrations.RemoveField(
model_name='invitation',
name='issuer',
),
migrations.DeleteModel(
name='DocumentAccess',
),
migrations.DeleteModel(
name='Invitation',
),
]
src/backend/core/models.py
+152 -791
View File
File diff suppressed because it is too large Load Diff
src/backend/core/services/collaboration_services.py
+8 -11
View File
@@ -22,8 +22,8 @@ class CollaborationService:
def reset_connections(self, document_id, user_id=None):
"""
Reset the connections of a document and all its descendants in the
collaboration server.
Reset the connections of a document in the collaboration server. The
hierarchy is owned by Drive, so there is no descendant fan-out.
Resetting a connection means that the user will be disconnected and will
have to reconnect to the collaboration server, with updated rights.
@@ -34,15 +34,12 @@ class CollaborationService:
logger.error("Document %s does not exists anymore", document_id)
return
documents = models.Document.objects.filter(
path__startswith=document.path, depth__gte=document.depth
).order_by("path")
for doc in documents:
try:
self._reset_connection(doc.id, user_id)
except requests.HTTPError:
logger.error("impossible to reset connections for document %s", doc.id)
try:
self._reset_connection(document.id, user_id)
except requests.HTTPError:
logger.error(
"impossible to reset connections for document %s", document.id
)
def _reset_connection(self, room, user_id=None):
"""
src/backend/core/services/search_indexers.py
+34 -37
View File
@@ -15,7 +15,6 @@ import requests
from core import models
from core.enums import SearchType
from core.utils.dicts import get_value_by_pattern
from core.utils.paths import get_ancestor_to_descendants_map
from core.utils.yjs import base64_yjs_to_text
logger = logging.getLogger(__name__)
@@ -42,34 +41,33 @@ def get_document_indexer():
return None
def get_batch_accesses_by_users_and_teams(paths):
def get_batch_accesses_by_users_and_teams(document_ids):
"""
Get accesses related to a list of document paths,
grouped by users and teams, including all ancestor paths.
Get read access candidates for a list of document ids, grouped by users
and teams. Sharing is owned by Drive: locally we only know the creator
and the users who already visited the document (LinkTrace). Drive still
gates actual document opens.
"""
ancestor_map = get_ancestor_to_descendants_map(
paths, steplen=models.Document.steplen
)
ancestor_paths = list(ancestor_map.keys())
access_by_document_id = defaultdict(lambda: {"users": set(), "teams": set()})
access_qs = models.DocumentAccess.objects.filter(
document__path__in=ancestor_paths
).values("document__path", "user__sub", "team")
creators_qs = models.Document.objects.filter(
id__in=document_ids, creator__isnull=False
).values("id", "creator__sub")
for entry in creators_qs:
if entry["creator__sub"]:
access_by_document_id[str(entry["id"])]["users"].add(
str(entry["creator__sub"])
)
access_by_document_path = defaultdict(lambda: {"users": set(), "teams": set()})
traces_qs = models.LinkTrace.objects.filter(
document_id__in=document_ids, user__sub__isnull=False
).values("document_id", "user__sub")
for entry in traces_qs:
access_by_document_id[str(entry["document_id"])]["users"].add(
str(entry["user__sub"])
)
for access in access_qs:
ancestor_path = access["document__path"]
user_sub = access["user__sub"]
team = access["team"]
for descendant_path in ancestor_map.get(ancestor_path, []):
if user_sub:
access_by_document_path[descendant_path]["users"].add(str(user_sub))
if team:
access_by_document_path[descendant_path]["teams"].add(team)
return dict(access_by_document_path)
return dict(access_by_document_id)
def get_visited_document_ids_of(queryset, user) -> tuple[str, ...]:
@@ -86,10 +84,9 @@ def get_visited_document_ids_of(queryset, user) -> tuple[str, ...]:
)
docs = (
queryset.exclude(accesses__user=user)
queryset.exclude(creator=user)
.filter(
deleted_at__isnull=True,
ancestors_deleted_at__isnull=True,
)
.filter(pk__in=visited_ids)
.order_by("pk")
@@ -153,12 +150,12 @@ class BaseDocumentIndexer(ABC):
if not documents_batch:
break
doc_paths = [doc.path for doc in documents_batch]
doc_ids = [doc.id for doc in documents_batch]
last_id = documents_batch[-1].id
accesses_by_document_path = get_batch_accesses_by_users_and_teams(doc_paths)
accesses_by_document_id = get_batch_accesses_by_users_and_teams(doc_ids)
serialized_batch = [
self.serialize_document(document, accesses_by_document_path)
self.serialize_document(document, accesses_by_document_id)
for document in documents_batch
if document.content or document.title
]
@@ -319,24 +316,24 @@ class FindDocumentIndexer(BaseDocumentIndexer):
Returns:
dict: A JSON-serializable dictionary.
"""
doc_path = document.path
doc_id = str(document.id)
doc_content = document.content
text_content = base64_yjs_to_text(doc_content) if doc_content else ""
return {
"id": str(document.id),
"id": doc_id,
"title": document.title or "",
"content": text_content,
"depth": document.depth,
"path": document.path,
"numchild": document.numchild,
"depth": 1,
"path": doc_id,
"numchild": 0,
"created_at": document.created_at.isoformat(),
"updated_at": document.updated_at.isoformat(),
"users": list(accesses.get(doc_path, {}).get("users", set())),
"groups": list(accesses.get(doc_path, {}).get("teams", set())),
"users": list(accesses.get(doc_id, {}).get("users", set())),
"groups": list(accesses.get(doc_id, {}).get("teams", set())),
"reach": document.computed_link_reach,
"size": len(text_content.encode("utf-8")),
"is_active": not bool(document.ancestors_deleted_at),
"is_active": document.deleted_at is None,
}
def search_query(self, data, token) -> requests.Response:
src/backend/core/signals.py
-27
View File
@@ -4,14 +4,12 @@ Declare and configure the signals for the impress core application
from functools import partial
from django.core.cache import cache
from django.db import transaction
from django.db.models import signals
from django.dispatch import receiver
from core import models
from core.tasks.search import trigger_batch_document_indexer
from core.utils.users import get_users_sharing_documents_with_cache_key
@receiver(signals.post_save, sender=models.Document)
@@ -22,28 +20,3 @@ def document_post_save(sender, instance, **kwargs): # pylint: disable=unused-ar
error.
"""
transaction.on_commit(partial(trigger_batch_document_indexer, instance))
@receiver(signals.post_save, sender=models.DocumentAccess)
def document_access_post_save(sender, instance, created, **kwargs): # pylint: disable=unused-argument
"""
Asynchronous call to the document indexer at the end of the transaction.
Clear cache for the affected user.
"""
if not created:
transaction.on_commit(
partial(trigger_batch_document_indexer, instance.document)
)
# Invalidate cache for the user
cache_key = get_users_sharing_documents_with_cache_key(instance.user_id)
cache.delete(cache_key)
@receiver(signals.post_delete, sender=models.DocumentAccess)
def document_access_post_delete(sender, instance, **kwargs): # pylint: disable=unused-argument
"""
Clear cache for the affected user when document access is deleted.
"""
cache_key = get_users_sharing_documents_with_cache_key(instance.user_id)
cache.delete(cache_key)
src/backend/core/tasks/mail.py
+16 -15
View File
@@ -7,18 +7,19 @@ from core import models
from impress.celery_app import app
@app.task
def send_ask_for_access_mail(ask_for_access_id):
"""Send mail using celery task."""
# Send email to document owners/admins
ask_for_access = models.DocumentAskForAccess.objects.get(id=ask_for_access_id)
owner_admin_accesses = models.DocumentAccess.objects.filter(
document=ask_for_access.document, role__in=models.PRIVILEGED_ROLES
).select_related("user")
for access in owner_admin_accesses:
if access.user and access.user.email:
ask_for_access.send_ask_for_access_email(
access.user.email,
access.user.language or settings.LANGUAGE_CODE,
)
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# @app.task
# def send_ask_for_access_mail(ask_for_access_id):
# """Send mail using celery task."""
# # Send email to document owners/admins
# ask_for_access = models.DocumentAskForAccess.objects.get(id=ask_for_access_id)
# owner_admin_accesses = models.DocumentAccess.objects.filter(
# document=ask_for_access.document, role__in=models.PRIVILEGED_ROLES
# ).select_related("user")
#
# for access in owner_admin_accesses:
# if access.user and access.user.email:
# ask_for_access.send_ask_for_access_email(
# access.user.email,
# access.user.language or settings.LANGUAGE_CODE,
# )
src/backend/core/tasks/search.py
+1 -3
View File
@@ -54,9 +54,7 @@ def batch_document_indexer_task(timestamp):
if indexer:
queryset = models.Document.objects.filter(
Q(updated_at__gte=timestamp)
| Q(deleted_at__gte=timestamp)
| Q(ancestors_deleted_at__gte=timestamp)
Q(updated_at__gte=timestamp) | Q(deleted_at__gte=timestamp)
)
count = indexer.index(queryset)
src/backend/core/tests/test_api_utils_filter_root_paths.py
-94
View File
@@ -1,94 +0,0 @@
"""
Unit tests for the filter_root_paths utility function.
"""
from core.api.utils import filter_root_paths
def test_api_utils_filter_root_paths_success():
"""
The `filter_root_paths` function should correctly identify root paths
from a given list of paths.
This test uses a list of paths with missing intermediate paths to ensure that
only the minimal set of root paths is returned.
"""
paths = [
"0001",
"00010001",
"000100010001",
"000100010002",
# missing 00010002
"000100020001",
"000100020002",
"0002",
"00020001",
"00020002",
# missing 0003
"00030001",
"000300010001",
"00030002",
# missing 0004
# missing 00040001
# missing 000400010001
# missing 000400010002
"000400010003",
"0004000100030001",
"000400010004",
]
filtered_paths = filter_root_paths(paths, skip_sorting=True)
assert filtered_paths == [
"0001",
"0002",
"00030001",
"00030002",
"000400010003",
"000400010004",
]
def test_api_utils_filter_root_paths_sorting():
"""
The `filter_root_paths` function should fail is sorting is skipped and paths are not sorted.
This test verifies that when sorting is skipped, the function respects the input order, and
when sorting is enabled, the result is correctly ordered and minimal.
"""
paths = [
"0001",
"00010001",
"000100010001",
"000100020002",
"000100010002",
"000100020001",
"00020001",
"0002",
"00020002",
"000300010001",
"00030001",
"00030002",
"0004000100030001",
"000400010003",
"000400010004",
]
filtered_paths = filter_root_paths(paths, skip_sorting=True)
assert filtered_paths == [
"0001",
"00020001",
"0002",
"000300010001",
"00030001",
"00030002",
"0004000100030001",
"000400010003",
"000400010004",
]
filtered_paths = filter_root_paths(paths)
assert filtered_paths == [
"0001",
"0002",
"00030001",
"00030002",
"000400010003",
"000400010004",
]
src/backend/core/tests/test_utils.py
-26
View File
@@ -10,7 +10,6 @@ import pytest
from core import factories
from core.utils.dicts import get_value_by_pattern
from core.utils.paths import get_ancestor_to_descendants_map
from core.utils.users import (
get_users_sharing_documents_with_cache_key,
users_sharing_documents_with,
@@ -93,31 +92,6 @@ def test_utils_extract_attachments():
assert extract_attachments(base64_string) == [image_key1, image_key3]
def test_utils_get_ancestor_to_descendants_map_single_path():
"""Test ancestor mapping of a single path."""
paths = ["000100020005"]
result = get_ancestor_to_descendants_map(paths, steplen=4)
assert result == {
"0001": {"000100020005"},
"00010002": {"000100020005"},
"000100020005": {"000100020005"},
}
def test_utils_get_ancestor_to_descendants_map_multiple_paths():
"""Test ancestor mapping of multiple paths with shared prefixes."""
paths = ["000100020005", "00010003"]
result = get_ancestor_to_descendants_map(paths, steplen=4)
assert result == {
"0001": {"000100020005", "00010003"},
"00010002": {"000100020005"},
"000100020005": {"000100020005"},
"00010003": {"00010003"},
}
def test_utils_users_sharing_documents_with_cache_miss():
"""Test cache miss: should query database and cache result."""
user1 = factories.UserFactory()
src/backend/core/tests/test_utils_create_tree_node_with_retry.py
-89
View File
@@ -1,89 +0,0 @@
"""Tests for the create_tree_node_with_retry utils."""
from unittest import mock
from django.core.exceptions import ValidationError as DjangoValidationError
from django.db import IntegrityError
import pytest
from core.factories import UserFactory
from core.models import Document
from core.utils.treebeard import _is_tree_path_collision, create_tree_node_with_retry
pytestmark = pytest.mark.django_db
@pytest.mark.parametrize(
"exc",
[
DjangoValidationError({"path": "not unique"}),
IntegrityError("impress_document_path_key"),
],
)
def test_utils_create_tree_node_with_retry_exceed_max_attempts(settings, exc):
"""Test exceeding the max attempts should reraise the exception."""
settings.TREEBEARD_PATH_COMPUTE_RETRY_MAX_ATTEMPTS = 2
create_fn = mock.MagicMock()
create_fn.side_effect = exc
with (
pytest.raises(exc.__class__),
mock.patch(
"core.utils.treebeard._is_tree_path_collision"
) as mock__is_tree_path_collision,
):
mock__is_tree_path_collision.side_effect = _is_tree_path_collision
create_tree_node_with_retry(create_fn)
mock__is_tree_path_collision.assert_called()
assert mock__is_tree_path_collision.call_count == 2
assert create_fn.call_count == 2
@pytest.mark.parametrize(
"exc",
[
DjangoValidationError({"foo": "bar"}),
IntegrityError("not handled"),
],
)
def test_utils_create_tree_node_with_retry_exceed_exception_not_handled(settings, exc):
"""Test with an exception not handled should return reraise it immediately."""
settings.TREEBEARD_PATH_COMPUTE_RETRY_MAX_ATTEMPTS = 2
create_fn = mock.MagicMock()
create_fn.side_effect = exc
with (
pytest.raises(exc.__class__),
mock.patch(
"core.utils.treebeard._is_tree_path_collision"
) as mock__is_tree_path_collision,
):
mock__is_tree_path_collision.side_effect = _is_tree_path_collision
create_tree_node_with_retry(create_fn)
mock__is_tree_path_collision.assert_called()
assert mock__is_tree_path_collision.call_count == 1
assert create_fn.call_count == 1
def test_utils_create_tree_node_with_retry_success():
"""Test executing successfully the create_fn callback."""
user = UserFactory()
document = create_tree_node_with_retry(
lambda: Document.add_root(
creator=user,
title="success",
)
)
assert isinstance(document, Document)
assert document.title == "success"
assert document.path is not None
src/backend/core/tests/test_utils_filter_descendants.py
-163
View File
@@ -1,163 +0,0 @@
"""
Unit tests for the filter_root_paths utility function.
"""
from core.utils.paths import filter_descendants
def test_utils_filter_descendants_success():
"""
The `filter_descendants` function should correctly identify descendant paths
from a given list of paths and root paths.
This test verifies that the function returns only the paths that have a prefix
matching one of the root paths.
"""
paths = [
"0001",
"00010001",
"000100010001",
"000100010002",
"000100020001",
"000100020002",
"0002",
"00020001",
"00020002",
"00030001",
"000300010001",
"00030002",
"0004",
"000400010003",
"0004000100030001",
"000400010004",
]
root_paths = [
"0001",
"0002",
"000400010003",
]
filtered_paths = filter_descendants(paths, root_paths, skip_sorting=True)
assert filtered_paths == [
"0001",
"00010001",
"000100010001",
"000100010002",
"000100020001",
"000100020002",
"0002",
"00020001",
"00020002",
"000400010003",
"0004000100030001",
]
def test_utils_filter_descendants_sorting():
"""
The `filter_descendants` function should handle unsorted input when sorting is enabled.
This test verifies that the function sorts the input if sorting is not skipped
and still correctly identifies accessible descendant paths.
"""
paths = [
"000300010001",
"000100010002",
"0001",
"00010001",
"000100010001",
"000100020002",
"000100020001",
"0002",
"00020001",
"00020002",
"00030001",
"00030002",
"0004000100030001",
"0004",
"000400010003",
"000400010004",
]
root_paths = [
"0002",
"000400010003",
"0001",
]
filtered_paths = filter_descendants(paths, root_paths)
assert filtered_paths == [
"0001",
"00010001",
"000100010001",
"000100010002",
"000100020001",
"000100020002",
"0002",
"00020001",
"00020002",
"000400010003",
"0004000100030001",
]
filtered_paths = filter_descendants(paths, root_paths, skip_sorting=True)
assert filtered_paths == [
"0001",
"00010001",
"000100010001",
"000100010002",
"000100020001",
"000100020002",
"0002",
"00020001",
"00020002",
"000400010003",
"0004000100030001",
]
def test_utils_filter_descendants_empty():
"""
The function should return an empty list if one or both inputs are empty.
"""
assert not filter_descendants([], ["0001"])
assert not filter_descendants(["0001"], [])
assert not filter_descendants([], [])
def test_utils_filter_descendants_no_match():
"""
The function should return an empty list if no path starts with any root path.
"""
paths = ["0001", "0002", "0003"]
root_paths = ["0004", "0005"]
assert not filter_descendants(paths, root_paths, skip_sorting=True)
def test_utils_filter_descendants_exact_match():
"""
The function should include paths that exactly match a root path.
"""
paths = ["0001", "0002", "0003"]
root_paths = ["0001", "0002"]
assert filter_descendants(paths, root_paths, skip_sorting=True) == ["0001", "0002"]
def test_utils_filter_descendants_single_root_matches_all():
"""
A single root path should match all its descendants.
"""
paths = ["0001", "00010001", "000100010001", "00010002"]
root_paths = ["0001"]
assert filter_descendants(paths, root_paths) == [
"0001",
"00010001",
"000100010001",
"00010002",
]
def test_utils_filter_descendants_path_shorter_than_root():
"""
A path shorter than any root path should not match.
"""
paths = ["0001", "0002"]
root_paths = ["00010001"]
assert not filter_descendants(paths, root_paths)
src/backend/core/urls.py
+33 -30
View File
@@ -17,26 +17,28 @@ router.register("users", viewsets.UserViewSet, basename="users")
# - Routes nested under a document
document_related_router = DefaultRouter()
document_related_router.register(
"accesses",
viewsets.DocumentAccessViewSet,
basename="document_accesses",
)
document_related_router.register(
"invitations",
viewsets.InvitationViewset,
basename="invitations",
)
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# document_related_router.register(
# "accesses",
# viewsets.DocumentAccessViewSet,
# basename="document_accesses",
# )
# document_related_router.register(
# "invitations",
# viewsets.InvitationViewset,
# basename="invitations",
# )
document_related_router.register(
"threads",
viewsets.ThreadViewSet,
basename="threads",
)
document_related_router.register(
"ask-for-access",
viewsets.DocumentAskForAccessViewSet,
basename="ask_for_access",
)
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# document_related_router.register(
# "ask-for-access",
# viewsets.DocumentAskForAccessViewSet,
# basename="ask_for_access",
# )
thread_related_router = DefaultRouter()
thread_related_router.register(
@@ -88,21 +90,22 @@ if settings.OIDC_RESOURCE_SERVER_ENABLED:
# - Routes nested under a document in external API
external_api_document_related_router = DefaultRouter()
document_access_config = settings.EXTERNAL_API.get("document_access", {})
if document_access_config.get("enabled", False):
external_api_document_related_router.register(
"accesses",
external_api_viewsets.ResourceServerDocumentAccessViewSet,
basename="resource_server_document_accesses",
)
document_invitation_config = settings.EXTERNAL_API.get("document_invitation", {})
if document_invitation_config.get("enabled", False):
external_api_document_related_router.register(
"invitations",
external_api_viewsets.ResourceServerInvitationViewSet,
basename="resource_server_document_invitations",
)
# POC-DRIVE-SHARING: disabled until Drive implements AskForAccess — do not delete
# document_access_config = settings.EXTERNAL_API.get("document_access", {})
# if document_access_config.get("enabled", False):
# external_api_document_related_router.register(
# "accesses",
# external_api_viewsets.ResourceServerDocumentAccessViewSet,
# basename="resource_server_document_accesses",
# )
#
# document_invitation_config = settings.EXTERNAL_API.get("document_invitation", {})
# if document_invitation_config.get("enabled", False):
# external_api_document_related_router.register(
# "invitations",
# external_api_viewsets.ResourceServerInvitationViewSet,
# basename="resource_server_document_invitations",
# )
urlpatterns.append(
path(
src/backend/core/utils/paths.py
-63
View File
@@ -1,63 +0,0 @@
"""Path and tree structure utilities."""
from collections import defaultdict
def get_ancestor_to_descendants_map(paths, steplen):
"""
Given a list of document paths, return a mapping of ancestor_path -> set of descendant_paths.
Each path is assumed to use materialized path format with fixed-length segments.
Args:
paths (list of str): List of full document paths.
steplen (int): Length of each path segment.
Returns:
dict[str, set[str]]: Mapping from ancestor path to its descendant paths (including itself).
"""
ancestor_map = defaultdict(set)
for path in paths:
for i in range(steplen, len(path) + 1, steplen):
ancestor = path[:i]
ancestor_map[ancestor].add(path)
return ancestor_map
def filter_descendants(paths, root_paths, skip_sorting=False):
"""
Filters paths to keep only those that are descendants of any path in root_paths.
A path is considered a descendant of a root path if it starts with the root path.
If `skip_sorting` is not set to True, the function will sort both lists before
processing because both `paths` and `root_paths` need to be in lexicographic order
before going through the algorithm.
Args:
paths (iterable of str): List of paths to be filtered.
root_paths (iterable of str): List of paths to check as potential prefixes.
skip_sorting (bool): If True, assumes both `paths` and `root_paths` are already sorted.
Returns:
list of str: A list of sorted paths that are descendants of any path in `root_paths`.
"""
results = []
i = 0
n = len(root_paths)
if not skip_sorting:
paths.sort()
root_paths.sort()
for path in paths:
# Try to find a matching prefix in the sorted accessible paths
while i < n:
if path.startswith(root_paths[i]):
results.append(path)
break
if root_paths[i] < path:
i += 1
else:
# If paths[i] > path, no need to keep searching
break
return results
src/backend/core/utils/treebeard.py
-62
View File
@@ -1,62 +0,0 @@
"""Treebeard path collision handling utilities."""
import logging
import time
from django.conf import settings
from django.core.exceptions import ValidationError as DjangoValidationError
from django.db import IntegrityError, transaction
logger = logging.getLogger(__name__)
def _is_tree_path_collision(exc):
"""Return True when `exc` is caused by a Document.path uniqueness conflict.
Treebeard computes the materialized path by reading the current siblings;
under concurrency two callers may compute the same value. Depending on
timing this surfaces either as:
- `django.core.exceptions.ValidationError` raised by `full_clean()` /
`validate_unique()` before the INSERT (BaseModel.save calls full_clean),
with this message `{'path': ['Document with this Path already exists.']}`
- or `IntegrityError` from the database unique index when the validate
step misses the conflict. With this message:
duplicate key value violates unique constraint "impress_document_path_key"
DETAIL: Key (path)=(0000001) already exists.
"""
if isinstance(exc, DjangoValidationError):
message_dict = getattr(exc, "message_dict", None)
if message_dict is not None:
return "path" in message_dict
return "path" in str(exc).lower()
# search in the IntegrityError exception
return "impress_document_path_key" in str(exc).lower()
def create_tree_node_with_retry(create_fn):
"""Run `create_fn` in a fresh atomic block, retrying on path collisions.
The Document.path field carries a unique constraint, which is the source of
truth that prevents duplicate paths. On collision we let the failed
transaction roll back, and call `create_fn` again so treebeard recomputes
the path from the latest state.
"""
max_attempts = settings.TREEBEARD_PATH_COMPUTE_RETRY_MAX_ATTEMPTS
for attempt in range(max_attempts):
try:
with transaction.atomic():
return create_fn()
except (IntegrityError, DjangoValidationError) as exc:
if not _is_tree_path_collision(exc) or attempt == max_attempts - 1:
raise
logger.info(
"tree path collision on attempt %d/%d, retrying",
attempt + 1,
max_attempts,
)
time.sleep(attempt * 0.1)
raise RuntimeError("create_tree_node_with_retry exited without result")
src/backend/core/utils/users.py
-55
View File
@@ -1,55 +0,0 @@
"""User sharing cache utilities."""
import logging
import time
from django.core.cache import cache
from django.db import models as db
from django.db.models import Subquery
from core import models
logger = logging.getLogger(__name__)
def get_users_sharing_documents_with_cache_key(user_id):
"""Generate a unique cache key for each user."""
return f"users_sharing_documents_with_{user_id}"
def users_sharing_documents_with(user_id):
"""
Returns a map of users sharing documents with the given user,
sorted by last shared date.
"""
start_time = time.time()
cache_key = get_users_sharing_documents_with_cache_key(user_id)
cached_result = cache.get(cache_key)
if cached_result is not None:
elapsed = time.time() - start_time
logger.info(
"users_sharing_documents_with cache hit for user %s (took %.3fs)",
user_id,
elapsed,
)
return cached_result
user_docs_qs = models.DocumentAccess.objects.filter(user__id=user_id).values_list(
"document_id", flat=True
)
shared_qs = (
models.DocumentAccess.objects.filter(document__id__in=Subquery(user_docs_qs))
.exclude(user__id=user_id)
.values("user")
.annotate(last_shared=db.Max("created_at"))
)
result = {item["user"]: item["last_shared"] for item in shared_qs}
cache.set(cache_key, result, 86400) # Cache for 1 day
elapsed = time.time() - start_time
logger.info(
"users_sharing_documents_with cache miss for user %s (took %.3fs)",
user_id,
elapsed,
)
return result
src/backend/demo/management/commands/create_demo.py
+8 -100
View File
@@ -118,106 +118,14 @@ class Timeit:
def create_demo(stdout):
"""
Create a database with demo data for developers to work in a realistic environment.
The code is engineered to create a huge number of objects fast.
"""
queue = BulkQueue(stdout)
with Timeit(stdout, "Creating users"):
name_size = int(math.sqrt(defaults.NB_OBJECTS["users"]))
first_names = [fake.first_name() for _ in range(name_size)]
last_names = [fake.last_name() for _ in range(name_size)]
for i in range(defaults.NB_OBJECTS["users"]):
first_name = random.choice(first_names)
queue.push(
models.User(
admin_email=f"user.test{i:d}@example.com",
email=f"user.test{i:d}@example.com",
password="!",
is_superuser=False,
is_active=True,
is_first_connection=False,
is_staff=False,
short_name=first_name,
full_name=f"{first_name:s} {random.choice(last_names):s}",
language=random.choice(languages),
)
)
queue.flush()
users_ids = list(models.User.objects.values_list("id", flat=True))
with Timeit(stdout, "Creating documents"):
for i in range(defaults.NB_OBJECTS["docs"]):
# pylint: disable=protected-access
key = models.Document._int2str(i) # noqa: SLF001
padding = models.Document.alphabet[0] * (models.Document.steplen - len(key))
title = fake.sentence(nb_words=4)
document = models.Document(
id=uuid4(),
depth=1,
path=f"{padding}{key}",
creator_id=random.choice(users_ids),
title=title,
link_reach=models.LinkReachChoices.AUTHENTICATED
if random_true_with_probability(0.5)
else random.choice(models.LinkReachChoices.values),
)
document.save_content(get_ydoc_for_text(f"Content for {title:s}"))
queue.push(document)
queue.flush()
with Timeit(stdout, "Creating docs accesses"):
docs_ids = list(models.Document.objects.values_list("id", flat=True))
for doc_id in docs_ids:
for user_id in random.sample(
users_ids,
random.randint(1, defaults.NB_OBJECTS["max_users_per_document"]),
):
role = random.choice(models.RoleChoices.choices)
queue.push(
models.DocumentAccess(
document_id=doc_id, user_id=user_id, role=role[0]
)
)
queue.flush()
with Timeit(stdout, "Creating development users"):
for dev_user in defaults.DEV_USERS:
queue.push(
models.User(
admin_email=dev_user["email"],
email=dev_user["email"],
sub=dev_user["email"],
password="!",
is_superuser=False,
is_active=True,
is_first_connection=False,
is_staff=False,
language=dev_user["language"] or random.choice(languages),
)
)
queue.flush()
with Timeit(stdout, "Creating docs accesses on development users"):
for dev_user in defaults.DEV_USERS:
docs_ids = list(models.Document.objects.values_list("id", flat=True))
user_id = models.User.objects.get(email=dev_user["email"]).id
for doc_id in docs_ids:
role = random.choice(models.RoleChoices.choices)
queue.push(
models.DocumentAccess(
document_id=doc_id, user_id=user_id, role=role[0]
)
)
queue.flush()
"""Not supported with the Drive integration."""
# The document tree, sharing and listing are owned by Drive: demo documents
# created locally would have no Drive item and would be invisible in the
# app. Seed data through the Drive app instead.
raise CommandError(
"create_demo is not supported with the Drive integration: documents "
"must be created through Drive."
)
class Command(BaseCommand):
"""A management command to create a demo database."""