Coverage for src/paperap/models/abstract/queryset.py: 73%
315 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-03-22 16:02 -0400
1"""
2----------------------------------------------------------------------------
4 METADATA:
6 File: queryset.py
7 Project: paperap
8 Created: 2025-03-04
9 Version: 0.0.9
10 Author: Jess Mann
11 Email: jess@jmann.me
12 Copyright (c) 2025 Jess Mann
14----------------------------------------------------------------------------
16 LAST MODIFIED:
18 2025-03-04 By Jess Mann
20"""
22from __future__ import annotations
24import copy
25import logging
26from datetime import datetime
27from string import Template
28from typing import TYPE_CHECKING, Any, Final, Generic, Iterable, Iterator, Self, TypeAlias, Union, override
30from pydantic import HttpUrl
31from typing_extensions import TypeVar
33from paperap.exceptions import FilterDisabledError, MultipleObjectsFoundError, ObjectNotFoundError
35if TYPE_CHECKING:
36 from paperap.models.abstract.model import BaseModel, StandardModel
37 from paperap.resources.base import BaseResource, StandardResource
39logger = logging.getLogger(__name__)
41# _BaseResource = TypeVar("_BaseResource", bound="BaseResource", default="BaseResource")
43type ClientResponse = dict[str, Any] | list[dict[str, Any]] | None
46class BaseQuerySet[_Model: BaseModel](Iterable[_Model]):
47 """
48 A lazy-loaded, chainable query interface for Paperless NGX resources.
50 BaseQuerySet provides pagination, filtering, and caching functionality similar to Django's QuerySet.
51 It's designed to be lazy - only fetching data when it's actually needed.
53 Args:
54 resource: The BaseResource instance.
55 filters: Initial filter parameters.
56 _cache: Optional internal result cache.
57 _fetch_all: Whether all results have been fetched.
58 _next_url: URL for the next page of results.
59 _last_response: Optional last response from the API.
60 _iter: Optional iterator for the results.
62 Returns:
63 A new instance of BaseQuerySet.
65 Examples:
66 # Create a QuerySet for documents
67 >>> docs = client.documents()
68 >>> for doc in docs:
69 ... print(doc.id)
70 1
71 2
72 3
74 """
76 resource: "BaseResource[_Model, Self]"
77 filters: dict[str, Any]
78 _last_response: ClientResponse | None = None
79 _result_cache: list[_Model] = []
80 _fetch_all: bool = False
81 _next_url: str | None = None
82 _urls_fetched: list[str] = []
83 _iter: Iterator[_Model] | None
85 def __init__(
86 self,
87 resource: "BaseResource[_Model, Self]",
88 filters: dict[str, Any] | None = None,
89 _cache: list[_Model] | None = None,
90 _fetch_all: bool = False,
91 _next_url: str | None = None,
92 _last_response: ClientResponse = None,
93 _iter: Iterator[_Model] | None = None,
94 _urls_fetched: list[str] | None = None,
95 ) -> None:
96 self.resource = resource
97 self.filters = filters or {}
98 self._result_cache = _cache or []
99 self._fetch_all = _fetch_all
100 self._next_url = _next_url
101 self._urls_fetched = _urls_fetched or []
102 self._last_response = _last_response
103 self._iter = _iter
105 super().__init__()
107 @property
108 def _model(self) -> type[_Model]:
109 """
110 Return the model class associated with the resource.
112 Returns:
113 The model class
115 Examples:
116 # Create a model instance
117 >>> model = queryset._model(**params)
119 """
120 return self.resource.model_class
122 @property
123 def _meta(self) -> "BaseModel.Meta[Any]":
124 """
125 Return the model's metadata.
127 Returns:
128 The model's metadata
130 Examples:
131 # Get the model's metadata
132 >>> queryset._meta.read_only_fields
133 {'id', 'added', 'modified'}
135 """
136 return self._model._meta # pyright: ignore[reportPrivateUsage] # pylint: disable=protected-access
138 def _reset(self) -> None:
139 """
140 Reset the QuerySet to its initial state.
142 This clears the result cache and resets the fetch state.
143 """
144 self._result_cache = []
145 self._fetch_all = False
146 self._next_url = None
147 self._urls_fetched = []
148 self._last_response = None
149 self._iter = None
151 def _update_filters(self, values: dict[str, Any]) -> None:
152 """
153 Update the current filters with new values.
155 This updates the current queryset instance. It does not return a new instance. For that reason,
156 do not call this directly. Call filter() or exclude() instead.
158 Args:
159 values: New filter values to add
161 Raises:
162 FilterDisabledError: If a filter is not allowed by the resource
164 Examples:
165 # Update filters with new values
166 queryset._update_filters({"correspondent": 1})
168 # Update filters with multiple values
169 queryset._update_filters({"correspondent": 1, "document_type": 2})
171 """
172 for key, _value in values.items():
173 if not self._meta.filter_allowed(key):
174 raise FilterDisabledError(f"Filtering by {key} for {self.resource.name} does not appear to be supported by the API.")
176 if values:
177 # Reset the cache if filters change
178 self._reset()
179 self.filters.update(**values)
181 def filter(self, **kwargs: Any) -> Self:
182 """
183 Return a new QuerySet with the given filters applied.
185 Args:
186 **kwargs: Filters to apply, where keys are field names and values are desired values.
187 Supports Django-style lookups like field__contains, field__in, etc.
189 Returns:
190 A new QuerySet with the additional filters applied
192 Examples:
193 # Get documents with specific correspondent
194 docs = client.documents.filter(correspondent=1)
196 # Get documents with specific correspondent and document type
197 docs = client.documents.filter(correspondent=1, document_type=2)
199 # Get documents with title containing "invoice"
200 docs = client.documents.filter(title__contains="invoice")
202 # Get documents with IDs in a list
203 docs = client.documents.filter(id__in=[1, 2, 3])
205 """
206 processed_filters = {}
208 for key, value in kwargs.items():
209 # Handle list values for __in lookups
210 if isinstance(value, (list, set, tuple)):
211 # Convert list to comma-separated string for the API
212 processed_value = ",".join(str(item) for item in value)
213 processed_filters[key] = processed_value
214 # Handle boolean values
215 elif isinstance(value, bool):
216 processed_filters[key] = str(value).lower()
217 # Handle normal values
218 else:
219 processed_filters[key] = value
221 return self._chain(filters={**self.filters, **processed_filters})
223 def exclude(self, **kwargs: Any) -> Self:
224 """
225 Return a new QuerySet excluding objects with the given filters.
227 Args:
228 **kwargs: Filters to exclude, where keys are field names and values are excluded values
230 Returns:
231 A new QuerySet excluding objects that match the filters
233 Examples:
234 # Get documents with any correspondent except ID 1
235 docs = client.documents.exclude(correspondent=1)
237 """
238 # Transform each key to its "not" equivalent
239 exclude_filters = {}
240 for key, value in kwargs.items():
241 if "__" in key:
242 field, lookup = key.split("__", 1)
243 # If it already has a "not" prefix, remove it
244 if lookup.startswith("not_"):
245 exclude_filters[f"{field}__{lookup[4:]}"] = value
246 else:
247 exclude_filters[f"{field}__not_{lookup}"] = value
248 else:
249 exclude_filters[f"{key}__not"] = value
251 return self._chain(filters={**self.filters, **exclude_filters})
253 def get(self, pk: Any) -> _Model:
254 """
255 Retrieve a single object from the API.
257 Raises NotImplementedError. Subclasses may implement this.
259 Args:
260 pk: The primary key (e.g. the id) of the object to retrieve
262 Returns:
263 A single object matching the query
265 Raises:
266 ObjectNotFoundError: If no object or multiple objects are found
267 NotImplementedError: If the method is not implemented by the subclass
269 Examples:
270 # Get document with ID 123
271 doc = client.documents.get(123)
273 """
274 raise NotImplementedError("Getting a single resource is not defined by BaseModels without an id.")
276 def _get_last_count(self) -> int | None:
277 if self._last_response is None:
278 return None
279 if isinstance(self._last_response, list):
280 return len(self._last_response)
281 return self._last_response.get("count")
283 def count(self) -> int:
284 """
285 Return the total number of objects in the queryset.
287 Returns:
288 The total count of objects matching the filters
290 Raises:
291 NotImplementedError: If the response does not have a count attribute
293 """
294 # If we have a last response, we can use the "count" field
295 if (count := self._get_last_count()) is not None:
296 return count
298 # Get one page of results, to populate last response
299 _iter = self._request_iter(params=self.filters)
301 # TODO Hack
302 for _ in _iter:
303 break
305 if not self._last_response:
306 # I don't think this should ever occur, but just in case.
307 raise NotImplementedError("Requested iter, but no last response")
309 if (count := self._get_last_count()) is not None:
310 return count
312 # I don't think this should ever occur, but just in case.
313 raise NotImplementedError(f"Unexpected Error: Could not determine count of objects. Last response: {self._last_response}")
315 def count_this_page(self) -> int:
316 """
317 Return the number of objects on the current page.
319 Returns:
320 The count of objects on the current page
322 Raises:
323 NotImplementedError: If _last_response is not set
325 """
326 # If we have a last response, we can count it without a new request
327 if self._last_response:
328 if isinstance(self._last_response, list):
329 return len(self._last_response)
330 results = self._last_response.get("results", [])
331 return len(results)
333 # Get one page of results, to populate last response
334 _iter = self._request_iter(params=self.filters)
336 # TODO Hack
337 for _ in _iter:
338 break
340 if not self._last_response:
341 # I don't think this should ever occur, but just in case.
342 raise NotImplementedError("Requested iter, but no last response")
344 if isinstance(self._last_response, list):
345 return len(self._last_response)
346 results = self._last_response.get("results", [])
347 return len(results)
349 def all(self) -> Self:
350 """
351 Return a new QuerySet that copies the current one.
353 Returns:
354 A copy of the current BaseQuerySet
356 """
357 return self._chain()
359 def order_by(self, *fields: str) -> Self:
360 """
361 Return a new QuerySet ordered by the specified fields.
363 Args:
364 *fields: Field names to order by. Prefix with '-' for descending order.
366 Returns:
367 A new QuerySet with the ordering applied
369 Examples:
370 # Order documents by title ascending
371 docs = client.documents.order_by('title')
373 # Order documents by added date descending
374 docs = client.documents.order_by('-added')
376 """
377 if not fields:
378 return self
380 # Combine with existing ordering if any
381 ordering = self.filters.get("ordering", [])
382 if isinstance(ordering, str):
383 ordering = [ordering]
384 elif not isinstance(ordering, list):
385 ordering = list(ordering)
387 # Add new ordering fields
388 new_ordering = ordering + list(fields)
390 # Join with commas for API
391 ordering_param = ",".join(new_ordering)
393 return self._chain(filters={**self.filters, "ordering": ordering_param})
395 def first(self) -> _Model | None:
396 """
397 Return the first object in the QuerySet, or None if empty.
399 Returns:
400 The first object or None if no objects match
402 """
403 if self._result_cache and len(self._result_cache) > 0:
404 return self._result_cache[0]
406 # If not cached, create a copy limited to 1 result
407 results = list(self._chain(filters={**self.filters, "limit": 1}))
408 return results[0] if results else None
410 def last(self) -> _Model | None:
411 """
412 Return the last object in the QuerySet, or None if empty.
414 Note: This requires fetching all results to determine the last one.
416 Returns:
417 The last object or None if no objects match
419 """
420 # If we have all results, we can just return the last one
421 if self._fetch_all:
422 if self._result_cache and len(self._result_cache) > 0:
423 return self._result_cache[-1]
424 return None
426 # We need all results to get the last one
427 self._fetch_all_results()
429 if self._result_cache and len(self._result_cache) > 0:
430 return self._result_cache[-1]
431 return None
433 def exists(self) -> bool:
434 """
435 Return True if the QuerySet contains any results.
437 Returns:
438 True if there are any objects matching the filters
440 """
441 # Check the cache before potentially making a new request
442 if self._fetch_all or self._result_cache:
443 return len(self._result_cache) > 0
445 # Check if there's at least one result
446 return self.first() is not None
448 def none(self) -> Self:
449 """
450 Return an empty QuerySet.
452 Returns:
453 An empty QuerySet
455 """
456 return self._chain(filters={"limit": 0})
458 def filter_field_by_str(self, field: str, value: str, *, exact: bool = True, case_insensitive: bool = True) -> Self:
459 """
460 Filter a queryset based on a given field.
462 This allows subclasses to easily implement custom filter methods.
464 Args:
465 field: The field name to filter by.
466 value: The value to filter against.
467 exact: Whether to filter by an exact match.
468 case_insensitive: Whether the filter should be case-insensitive.
470 Returns:
471 A new QuerySet instance with the filter applied.
473 """
474 if exact:
475 lookup = f"{field}__iexact" if case_insensitive else field
476 else:
477 lookup = f"{field}__icontains" if case_insensitive else f"{field}__contains"
479 return self.filter(**{lookup: value})
481 def _fetch_all_results(self) -> None:
482 """
483 Fetch all results from the API and populate the cache.
485 Returns:
486 None
488 """
489 if self._fetch_all:
490 return
492 # Clear existing cache if any
493 self._result_cache = []
495 # Initial fetch
496 iterator = self._request_iter(params=self.filters)
498 # Collect results from initial page
499 # TODO: Consider itertools chain for performance reasons (?)
500 self._result_cache.extend(list(iterator))
502 # Fetch additional pages if available
503 while self._last_response and self._next_url:
504 iterator = self._request_iter(url=self._next_url)
505 self._result_cache.extend(list(iterator))
507 self._fetch_all = True
509 def _request_iter(self, url: str | HttpUrl | Template | None = None, params: dict[str, Any] | None = None) -> Iterator[_Model]:
510 """
511 Get an iterator of resources.
513 Args:
514 url: The URL to request, if different from the resource's default.
515 params: Query parameters.
517 Returns:
518 An iterator over the resources.
520 Raises:
521 NotImplementedError: If the request cannot be completed.
523 Examples:
524 # Iterate over documents
525 for doc in queryset._request_iter():
526 print(doc)
528 """
529 if not (response := self.resource.request_raw(url=url, params=params)):
530 logger.debug("No response from request.")
531 return
533 self._last_response = response
535 yield from self.resource.handle_response(response)
537 def _get_next(self, response: ClientResponse | None = None) -> str | None:
538 """
539 Get the next url, and adjust our references accordingly.
540 """
541 # Allow passing a different response
542 if response is None:
543 response = self._last_response
545 if isinstance(response, list):
546 return None
548 # Last response is not set
549 if not response or not (next_url := response.get("next")):
550 self._next_url = None
551 return None
553 # For safety, check both instance attributes, even though the first check isn't strictly necessary
554 # this hopefully future proofs any changes to the implementation
555 if next_url == self._next_url or next_url in self._urls_fetched:
556 logger.debug(
557 "Next URL was previously fetched. Stopping iteration. URL: %s, Already Fetched: %s",
558 next_url,
559 self._urls_fetched,
560 )
561 self._next_url = None
562 return None
564 # Cache it
565 self._next_url = next_url
566 self._urls_fetched.append(next_url)
567 return self._next_url
569 def _chain(self, **kwargs: Any) -> Self:
570 """
571 Return a copy of the current BaseQuerySet with updated attributes.
573 Args:
574 **kwargs: Attributes to update in the new BaseQuerySet
576 Returns:
577 A new QuerySet with the updated attributes
579 """
580 # Create a new BaseQuerySet with copied attributes
581 clone = self.__class__(self.resource) # type: ignore # pyright not handling Self correctly
583 # Copy attributes from self
584 clone.filters = copy.deepcopy(self.filters)
585 # Do not copy the cache, fetch_all, etc, since filters may change it
587 # Update with provided kwargs
588 for key, value in kwargs.items():
589 if key == "filters" and value:
590 clone._update_filters(value) # pylint: disable=protected-access
591 else:
592 setattr(clone, key, value)
594 return clone
596 @override
597 def __iter__(self) -> Iterator[_Model]:
598 """
599 Iterate over the objects in the QuerySet.
601 Returns:
602 An iterator over the objects
604 """
605 # If we have a fully populated cache, use it
606 if self._fetch_all:
607 yield from self._result_cache
608 return
610 if not self._iter:
611 # Start a new iteration
612 self._iter = self._request_iter(params=self.filters)
614 # Yield objects from the current page
615 for obj in self._iter:
616 self._result_cache.append(obj)
617 yield obj
619 self._get_next()
621 # If there are more pages, keep going
622 count = 0
623 while self._next_url:
624 count += 1
625 self._iter = self._request_iter(url=self._next_url)
627 # Yield objects from the current page
628 for obj in self._iter:
629 self._result_cache.append(obj)
630 yield obj
632 self._get_next()
634 # We've fetched everything
635 self._fetch_all = True
636 self._iter = None
638 def __len__(self) -> int:
639 """
640 Return the number of objects in the QuerySet.
642 Returns:
643 The count of objects
645 """
646 return self.count()
648 def __bool__(self) -> bool:
649 """
650 Return True if the QuerySet has any results.
652 Returns:
653 True if there are any objects matching the filters
655 """
656 return self.exists()
658 def __getitem__(self, key: int | slice) -> _Model | list[_Model]:
659 """
660 Retrieve an item or slice of items from the QuerySet.
662 Args:
663 key: An integer index or slice
665 Returns:
666 A single object or list of objects
668 Raises:
669 IndexError: If the index is out of range
671 """
672 if isinstance(key, slice):
673 # Handle slicing
674 start = key.start if key.start is not None else 0
675 stop = key.stop
677 if start < 0 or (stop is not None and stop < 0):
678 # Negative indexing requires knowing the full size
679 self._fetch_all_results()
680 return self._result_cache[key]
682 # Optimize by using limit/offset if available
683 if start == 0 and stop is not None:
684 # Simple limit
685 clone = self._chain(filters={**self.filters, "limit": stop})
686 results = list(clone)
687 return results
689 if start > 0 and stop is not None:
690 # Limit with offset
691 clone = self._chain(
692 filters={
693 **self.filters,
694 "limit": stop - start,
695 "offset": start,
696 }
697 )
698 results = list(clone)
699 return results
701 if start > 0 and stop is None:
702 # Just offset
703 clone = self._chain(filters={**self.filters, "offset": start})
704 self._fetch_all_results() # We need all results after the offset
705 return self._result_cache
707 # Default to fetching all and slicing
708 self._fetch_all_results()
709 return self._result_cache[key]
711 # Handle integer indexing
712 if key < 0:
713 # Negative indexing requires the full result set
714 self._fetch_all_results()
715 return self._result_cache[key]
717 # Positive indexing - we can optimize with limit/offset
718 if len(self._result_cache) > key:
719 # Already have this item cached
720 return self._result_cache[key]
722 # Fetch specific item by position
723 clone = self._chain(filters={**self.filters, "limit": 1, "offset": key})
724 results = list(clone)
725 if not results:
726 raise IndexError(f"BaseQuerySet index {key} out of range")
727 return results[0]
729 def __contains__(self, item: Any) -> bool:
730 """
731 Return True if the QuerySet contains the given object.
733 Args:
734 item: The object to check for
736 Returns:
737 True if the object is in the QuerySet
739 """
740 if not isinstance(item, self._model):
741 return False
743 return any(obj == item for obj in self)
746class StandardQuerySet[_Model: StandardModel](BaseQuerySet[_Model]):
747 """
748 A queryset for StandardModel instances (i.e. BaseModels with standard fields, like id).
750 Returns:
751 A new instance of StandardModel.
753 Raises:
754 ValueError: If resource is not provided.
756 Examples:
757 # Create a StandardModel instance
758 model = StandardModel(id=1)
760 Args:
761 resource: The BaseResource instance.
762 filters: Initial filter parameters.
764 Returns:
765 A new instance of StandardQuerySet.
767 Raises:
768 ObjectNotFoundError: If no object or multiple objects are found.
770 Examples:
771 # Create a StandardQuerySet for documents
772 docs = StandardQuerySet(resource=client.documents)
774 """
776 resource: "StandardResource[_Model, Self]" # type: ignore # pyright is getting inheritance wrong
778 @override
779 def get(self, pk: int) -> _Model:
780 """
781 Retrieve a single object from the API.
783 Args:
784 pk: The ID of the object to retrieve
786 Returns:
787 A single object matching the query
789 Raises:
790 ObjectNotFoundError: If no object or multiple objects are found
792 Examples:
793 # Get document with ID 123
794 doc = client.documents.get(123)
796 """
797 # Attempt to find it in the result cache
798 if self._result_cache:
799 for obj in self._result_cache:
800 if obj.id == pk:
801 return obj
803 # Direct lookup by ID - use the resource's get method
804 return self.resource.get(pk)
806 def id(self, value: int | list[int]) -> Self:
807 """
808 Filter models by ID.
810 Args:
811 value: The ID or list of IDs to filter by
813 Returns:
814 Filtered QuerySet
816 """
817 if isinstance(value, list):
818 return self.filter(id__in=value)
819 return self.filter(id=value)
821 @override
822 def __contains__(self, item: Any) -> bool:
823 """
824 Return True if the QuerySet contains the given object.
826 NOTE: This method only ensures a match by ID, not by full object equality.
827 This is intentional, as the object may be outdated or not fully populated.
829 Args:
830 item: The object or ID to check for
832 Returns:
833 True if the object is in the QuerySet
835 """
836 # Handle integers directly
837 if isinstance(item, int):
838 return any(obj.id == item for obj in self)
840 # Handle model objects that have an id attribute
841 try:
842 if hasattr(item, "id"):
843 return any(obj.id == item.id for obj in self)
844 except (AttributeError, TypeError):
845 pass
847 # For any other type, it's not in the queryset
848 return False
850 def bulk_action(self, action: str, **kwargs: Any) -> ClientResponse:
851 """
852 Perform a bulk action on all objects in the queryset.
854 This method fetches all IDs in the queryset and passes them to the resource's bulk_action method.
856 Args:
857 action: The action to perform
858 **kwargs: Additional parameters for the action
860 Returns:
861 The API response
863 Raises:
864 NotImplementedError: If the resource doesn't support bulk actions
866 """
867 if not (fn := getattr(self.resource, "bulk_action", None)):
868 raise NotImplementedError(f"Resource {self.resource.name} does not support bulk actions")
870 # Fetch all IDs in the queryset
871 # We only need IDs, so optimize by requesting just the ID field if possible
872 ids = [obj.id for obj in self]
874 if not ids:
875 return {"success": True, "count": 0}
877 return fn(action, ids, **kwargs)
879 def bulk_delete(self) -> ClientResponse:
880 """
881 Delete all objects in the queryset.
883 Returns:
884 The API response
886 """
887 return self.bulk_action("delete")
889 def bulk_update(self, **kwargs: Any) -> ClientResponse:
890 """
891 Update all objects in the queryset with the given values.
893 Args:
894 **kwargs: Fields to update
896 Returns:
897 The API response
899 """
900 if not (fn := getattr(self.resource, "bulk_update", None)):
901 raise NotImplementedError(f"Resource {self.resource.name} does not support bulk updates")
903 # Fetch all IDs in the queryset
904 ids = [obj.id for obj in self]
906 if not ids:
907 return {"success": True, "count": 0}
909 return fn(ids, **kwargs)
911 def bulk_assign_tags(self, tag_ids: list[int], remove_existing: bool = False) -> ClientResponse:
912 """
913 Assign tags to all objects in the queryset.
915 Args:
916 tag_ids: List of tag IDs to assign
917 remove_existing: If True, remove existing tags before assigning new ones
919 Returns:
920 The API response
922 """
923 if not (fn := getattr(self.resource, "bulk_assign_tags", None)):
924 raise NotImplementedError(f"Resource {self.resource.name} does not support bulk tag assignment")
926 # Fetch all IDs in the queryset
927 ids = [obj.id for obj in self]
929 if not ids:
930 return {"success": True, "count": 0}
932 return fn(ids, tag_ids, remove_existing)
934 def bulk_assign_correspondent(self, correspondent_id: int) -> ClientResponse:
935 """
936 Assign a correspondent to all objects in the queryset.
938 Args:
939 correspondent_id: Correspondent ID to assign
941 Returns:
942 The API response
944 """
945 if not (fn := getattr(self.resource, "bulk_assign_correspondent", None)):
946 raise NotImplementedError(f"Resource {self.resource.name} does not support bulk correspondent assignment")
948 # Fetch all IDs in the queryset
949 ids = [obj.id for obj in self]
951 if not ids:
952 return {"success": True, "count": 0}
954 return fn(ids, correspondent_id)
956 def bulk_assign_document_type(self, document_type_id: int) -> ClientResponse:
957 """
958 Assign a document type to all objects in the queryset.
960 Args:
961 document_type_id: Document type ID to assign
963 Returns:
964 The API response
966 """
967 if not (fn := getattr(self.resource, "bulk_assign_document_type", None)):
968 raise NotImplementedError(f"Resource {self.resource.name} does not support bulk document type assignment")
970 # Fetch all IDs in the queryset
971 ids = [obj.id for obj in self]
973 if not ids:
974 return {"success": True, "count": 0}
976 return fn(ids, document_type_id)
978 def bulk_assign_storage_path(self, storage_path_id: int) -> ClientResponse:
979 """
980 Assign a storage path to all objects in the queryset.
982 Args:
983 storage_path_id: Storage path ID to assign
985 Returns:
986 The API response
988 """
989 if not (fn := getattr(self.resource, "bulk_assign_storage_path", None)):
990 raise NotImplementedError(f"Resource {self.resource.name} does not support bulk storage path assignment")
992 # Fetch all IDs in the queryset
993 ids = [obj.id for obj in self]
995 if not ids:
996 return {"success": True, "count": 0}
998 return fn(ids, storage_path_id)
1000 def bulk_assign_owner(self, owner_id: int) -> ClientResponse:
1001 """
1002 Assign an owner to all objects in the queryset.
1004 Args:
1005 owner_id: Owner ID to assign
1007 Returns:
1008 The API response
1010 """
1011 if not (fn := getattr(self.resource, "bulk_assign_owner", None)):
1012 raise NotImplementedError(f"Resource {self.resource.name} does not support bulk owner assignment")
1014 # Fetch all IDs in the queryset
1015 ids = [obj.id for obj in self]
1017 if not ids:
1018 return {"success": True, "count": 0}
1020 return fn(ids, owner_id)