Coverage for src/paperap/resources/base.py: 91%
215 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-03-22 16:02 -0400
1"""
6----------------------------------------------------------------------------
8METADATA:
10File: base.py
11 Project: paperap
12Created: 2025-03-21
13 Version: 0.0.10
14Author: Jess Mann
15Email: jess@jmann.me
16 Copyright (c) 2025 Jess Mann
18----------------------------------------------------------------------------
20LAST MODIFIED:
222025-03-21 By Jess Mann
24"""
26from __future__ import annotations
28import copy
29import logging
30from abc import ABC, ABCMeta
31from string import Template
32from typing import TYPE_CHECKING, Any, ClassVar, Final, Generic, Iterator, overload, override
34from pydantic import HttpUrl, field_validator
35from typing_extensions import TypeVar
37from paperap.const import URLS, Endpoints
38from paperap.exceptions import (
39 ConfigurationError,
40 ModelValidationError,
41 ObjectNotFoundError,
42 ResourceNotFoundError,
43 ResponseParsingError,
44)
45from paperap.signals import registry
47if TYPE_CHECKING:
48 from paperap.client import PaperlessClient
49 from paperap.models.abstract.model import BaseModel, StandardModel
50 from paperap.models.abstract.queryset import BaseQuerySet, StandardQuerySet
52_BaseModel = TypeVar("_BaseModel", bound="BaseModel", default="BaseModel")
53_BaseQuerySet = TypeVar("_BaseQuerySet", bound="BaseQuerySet[Any]", default="BaseQuerySet")
54_StandardModel = TypeVar("_StandardModel", bound="StandardModel", default="StandardModel")
55_StandardQuerySet = TypeVar("_StandardQuerySet", bound="StandardQuerySet[Any]", default="StandardQuerySet")
57logger = logging.getLogger(__name__)
60class BaseResource(ABC, Generic[_BaseModel, _BaseQuerySet]):
61 """
62 Base class for API resources.
64 Args:
65 client: The PaperlessClient instance.
66 endpoint: The API endpoint for this resource.
67 model_class: The model class for this resource.
69 """
71 # The model class for this resource.
72 model_class: type[_BaseModel]
73 queryset_class: type[_BaseQuerySet]
75 # The PaperlessClient instance.
76 client: "PaperlessClient"
77 # The name of the model. This must line up with the API endpoint
78 # It will default to the model's name
79 name: str
80 # The API endpoint for this model.
81 # It will default to a standard schema used by the API
82 # Setting it will allow you to contact a different schema or even a completely different API.
83 # this will usually not need to be overridden
84 endpoints: ClassVar[Endpoints]
86 def __init__(self, client: "PaperlessClient") -> None:
87 self.client = client
88 if not hasattr(self, "name"):
89 self.name = f"{self._meta.name.lower()}s"
91 # Allow templating
92 for key, value in self.endpoints.items():
93 # endpoints is always dict[str, Template]
94 self.endpoints[key] = Template(value.safe_substitute(resource=self.name))
96 # Ensure the model has a link back to this resource
97 self.model_class._resource = self # type: ignore # allow private access
99 super().__init__()
101 @override
102 @classmethod
103 def __init_subclass__(cls, **kwargs: Any) -> None:
104 """
105 Initialize the subclass.
107 Args:
108 **kwargs: Arbitrary keyword arguments
110 """
111 super().__init_subclass__(**kwargs)
113 # Skip processing for the base class itself. TODO: This is a hack
114 if cls.__name__ in ["BaseResource", "StandardResource"]:
115 return
117 # model_class is required
118 if not (_model_class := getattr(cls, "model_class", None)):
119 raise ConfigurationError(f"model_class must be defined in {cls.__name__}")
121 # API Endpoint must be defined
122 if not (endpoints := getattr(cls, "endpoints", {})):
123 endpoints = {
124 "list": URLS.list,
125 "detail": URLS.detail,
126 "create": URLS.create,
127 "update": URLS.update,
128 "delete": URLS.delete,
129 }
131 cls.endpoints = cls._validate_endpoints(endpoints) # type: ignore # Allow assigning in subclass
133 @property
134 def _meta(self) -> "BaseModel.Meta[_BaseModel]":
135 return self.model_class._meta # pyright: ignore[reportPrivateUsage] # pylint: disable=protected-access
137 @classmethod
138 def _validate_endpoints(cls, value: Any) -> Endpoints:
139 if not isinstance(value, dict):
140 raise ModelValidationError("endpoints must be a dictionary")
142 converted: Endpoints = {}
143 for k, v in value.items():
144 if isinstance(v, Template):
145 converted[k] = v
146 continue
148 if not isinstance(v, str):
149 raise ModelValidationError(f"endpoints[{k}] must be a string or template")
151 try:
152 converted[k] = Template(v)
153 except ValueError as e:
154 raise ModelValidationError(f"endpoints[{k}] is not a valid template: {e}") from e
156 # We validated that converted matches endpoints above
157 return converted
159 def get_endpoint(self, name: str, **kwargs: Any) -> str | HttpUrl:
160 if not (template := self.endpoints.get(name, None)):
161 raise ConfigurationError(f"Endpoint {name} not defined for resource {self.name}")
163 if "resource" not in kwargs:
164 kwargs["resource"] = self.name
166 url = template.safe_substitute(**kwargs)
168 if not url.startswith("http"):
169 url = f"{self.client.base_url}{url.lstrip('/')}"
171 return HttpUrl(url)
173 def all(self) -> _BaseQuerySet:
174 """
175 Return a QuerySet representing all objects of this resource type.
177 Returns:
178 A QuerySet for this resource
180 """
181 return self.queryset_class(self) # type: ignore # _meta.queryset is always the right queryset type
183 def filter(self, **kwargs: Any) -> _BaseQuerySet:
184 """
185 Return a QuerySet filtered by the given parameters.
187 Args:
188 **kwargs: Filter parameters
190 Returns:
191 A filtered QuerySet
193 """
194 return self.all().filter(**kwargs)
196 def get(self, *args: Any, **kwargs: Any) -> _BaseModel:
197 """
198 Get a model by ID.
200 Raises NotImplementedError. Subclasses may implement this.
202 Raises:
203 NotImplementedError: Unless implemented by a subclass.
205 Returns:
206 The model retrieved.
208 """
209 raise NotImplementedError("get method not available for resources without an id")
211 def create(self, **kwargs: Any) -> _BaseModel:
212 """
213 Create a new resource.
215 Args:
216 data: Resource data.
218 Returns:
219 The created resource.
221 """
222 # Signal before creating resource
223 signal_params = {"resource": self.name, "data": kwargs}
224 registry.emit("resource.create:before", "Emitted before creating a resource", kwargs=signal_params)
226 if not (url := self.get_endpoint("create", resource=self.name)):
227 raise ConfigurationError(f"Create endpoint not defined for resource {self.name}")
229 if not (response := self.client.request("POST", url, data=kwargs)):
230 raise ResourceNotFoundError("Resource {resource} not found after create.", resource_name=self.name)
232 model = self.parse_to_model(response)
234 # Signal after creating resource
235 registry.emit(
236 "resource.create:after",
237 "Emitted after creating a resource",
238 args=[self],
239 kwargs={"model": model, **signal_params},
240 )
242 return model
244 def update(self, model: _BaseModel) -> _BaseModel:
245 """
246 Update a resource.
248 Args:
249 resource: The resource to update.
251 Returns:
252 The updated resource.
254 """
255 raise NotImplementedError("update method not available for resources without an id")
257 def update_dict(self, *args: Any, **kwargs: Any) -> _BaseModel:
258 """
259 Update a resource.
261 Subclasses may implement this.
262 """
263 raise NotImplementedError("update_dict method not available for resources without an id")
265 def delete(self, *args: Any, **kwargs: Any) -> None:
266 """
267 Delete a resource.
269 Args:
270 model_id: ID of the resource.
272 """
273 raise NotImplementedError("delete method not available for resources without an id")
275 def parse_to_model(self, item: dict[str, Any]) -> _BaseModel:
276 """
277 Parse an item dictionary into a model instance, handling date parsing.
279 Args:
280 item: The item dictionary.
282 Returns:
283 The parsed model instance.
285 """
286 try:
287 data = self.transform_data_input(**item)
288 return self.model_class.model_validate(data)
289 except Exception as e:
290 logger.error('Error parsing model "%s" with data: %s -> %s', self.name, item, e)
291 raise
293 def transform_data_input(self, **data: Any) -> dict[str, Any]:
294 """
295 Transform data after receiving it from the API.
297 Args:
298 data: The data to transform.
300 Returns:
301 The transformed data.
303 """
304 for key, value in self._meta.field_map.items():
305 if key in data:
306 data[value] = data.pop(key)
307 return data
309 @overload
310 def transform_data_output(self, model: _BaseModel, exclude_unset: bool = True) -> dict[str, Any]: ...
312 @overload
313 def transform_data_output(self, **data: Any) -> dict[str, Any]: ...
315 def transform_data_output(self, model: _BaseModel | None = None, exclude_unset: bool = True, **data: Any) -> dict[str, Any]:
316 """
317 Transform data before sending it to the API.
319 Args:
320 model: The model to transform.
321 exclude_unset: If model is provided, exclude unset fields when calling to_dict()
322 data: The data to transform.
324 Returns:
325 The transformed data.
327 """
328 if model:
329 if data:
330 # Combining model.to_dict() and data is ambiguous, so not allowed.
331 raise ValueError("Only one of model or data should be provided")
332 data = model.to_dict(exclude_unset=exclude_unset)
334 for key, value in self._meta.field_map.items():
335 if value in data:
336 data[key] = data.pop(value)
337 return data
339 def create_model(self, **kwargs: Any) -> _BaseModel:
340 """
341 Create a new model instance.
343 Args:
344 **kwargs: Model field values
346 Returns:
347 A new model instance.
349 """
350 # Mypy output:
351 # base.py:326:52: error: Argument "resource" to "BaseModel" has incompatible type
352 # "BaseResource[_BaseModel, _BaseQuerySet]"; expected "BaseResource[BaseModel, BaseQuerySet[BaseModel]] | None
353 return self.model_class(**kwargs, resource=self) # type: ignore
355 def request_raw(
356 self,
357 url: str | Template | HttpUrl | None = None,
358 method: str = "GET",
359 params: dict[str, Any] | None = None,
360 data: dict[str, Any] | None = None,
361 ) -> dict[str, Any] | list[dict[str, Any]] | None:
362 """
363 Make an HTTP request to the API, and return the raw json response.
365 Args:
366 method: The HTTP method to use
367 url: The full URL to request
368 params: Query parameters
369 data: Request body data
371 Returns:
372 The JSON-decoded response from the API
374 """
375 if not url and not (url := self.get_endpoint("list", resource=self.name)):
376 raise ConfigurationError(f"List endpoint not defined for resource {self.name}")
378 if isinstance(url, Template):
379 url = url.safe_substitute(resource=self.name)
381 response = self.client.request(method, url, params=params, data=data)
382 return response
384 def handle_response(self, response: Any) -> Iterator[_BaseModel]:
385 registry.emit(
386 "resource._handle_response:before",
387 "Emitted before listing resources",
388 return_type=dict[str, Any],
389 args=[self],
390 kwargs={"response": response, "resource": self.name},
391 )
393 if isinstance(response, list):
394 yield from self.handle_results(response)
395 elif isinstance(response, dict):
396 yield from self.handle_dict_response(**response)
397 else:
398 raise ResponseParsingError(f"Expected response to be list/dict, got {type(response)} -> {response}")
400 registry.emit(
401 "resource._handle_response:after",
402 "Emitted after listing resources",
403 return_type=dict[str, Any],
404 args=[self],
405 kwargs={"response": response, "resource": self.name},
406 )
408 def handle_dict_response(self, **response: dict[str, Any]) -> Iterator[_BaseModel]:
409 """
410 Handle a response from the API and yield results.
412 Override in subclasses to implement custom response logic.
413 """
414 if not (results := response.get("results", response)):
415 return
417 # Signal after receiving response
418 registry.emit(
419 "resource._handle_response:after",
420 "Emitted after list response, before processing",
421 args=[self],
422 kwargs={"response": {**response}, "resource": self.name, "results": results},
423 )
425 # If this is a single-item response (not a list), handle it differently
426 if isinstance(results, dict):
427 # For resources that return a single object directly
428 registry.emit(
429 "resource._handle_results:before",
430 "Emitted for direct object response",
431 args=[self],
432 kwargs={"resource": self.name, "item": {**results}},
433 )
434 yield self.parse_to_model(results)
435 return
437 if isinstance(results, list):
438 yield from self.handle_results(results)
439 return
441 raise ResponseParsingError(f"Expected {self.name} results to be list/dict, got {type(results)} -> {results}")
443 def handle_results(self, results: list[dict[str, Any]]) -> Iterator[_BaseModel]:
444 """
445 Yield parsed models from a list of results.
447 Override in subclasses to implement custom result handling.
448 """
449 if not isinstance(results, list):
450 raise ResponseParsingError(f"Expected {self.name} results to be a list, got {type(results)} -> {results}")
452 for item in results:
453 if not isinstance(item, dict):
454 raise ResponseParsingError(f"Expected type of elements in results is dict, got {type(item)}")
456 registry.emit(
457 "resource._handle_results:before",
458 "Emitted for each item in a list response",
459 args=[self],
460 kwargs={"resource": self.name, "item": {**item}},
461 )
462 yield self.parse_to_model(item)
464 def __call__(self, *args: Any, **keywords: Any) -> _BaseQuerySet:
465 """
466 Make the resource callable to get a BaseQuerySet.
468 This allows usage like: client.documents(title__contains='invoice')
470 Args:
471 *args: Unused
472 **keywords: Filter parameters
474 Returns:
475 A filtered QuerySet
477 """
478 return self.filter(**keywords)
481class StandardResource(BaseResource[_StandardModel, _StandardQuerySet]):
482 """
483 Base class for API resources.
485 Args:
486 client: The PaperlessClient instance.
487 endpoint: The API endpoint for this resource.
488 model_class: The model class for this resource.
490 """
492 @override
493 def get(self, model_id: int, *args: Any, **kwargs: Any) -> _StandardModel:
494 """
495 Get a model within this resource by ID.
497 Args:
498 model_id: ID of the model to retrieve.
500 Returns:
501 The model retrieved
503 """
504 # Signal before getting resource
505 signal_params = {"resource": self.name, "model_id": model_id}
506 registry.emit("resource.get:before", "Emitted before getting a resource", args=[self], kwargs=signal_params)
508 if not (url := self.get_endpoint("detail", resource=self.name, pk=model_id)):
509 raise ConfigurationError(f"Get detail endpoint not defined for resource {self.name}")
511 if not (response := self.client.request("GET", url)):
512 raise ObjectNotFoundError(resource_name=self.name, model_id=model_id)
514 # If the response doesn't have an ID, it's likely a 404
515 if not response.get("id"):
516 message = response.get("detail") or f"No ID found in {self.name} response"
517 raise ObjectNotFoundError(message, resource_name=self.name, model_id=model_id)
519 model = self.parse_to_model(response)
521 # Signal after getting resource
522 registry.emit(
523 "resource.get:after",
524 "Emitted after getting a single resource by id",
525 args=[self],
526 kwargs={**signal_params, "model": model},
527 )
529 return model
531 @override
532 def update(self, model: _StandardModel) -> _StandardModel:
533 """
534 Update a model.
536 Args:
537 model: The model to update.
539 Returns:
540 The updated model.
542 """
543 data = model.to_dict()
544 data = self.transform_data_output(**data)
546 # Save the model ID
547 model_id = model.id
549 # Remove ID from the data dict to avoid duplicating it in the call
550 data.pop("id", None)
552 return self.update_dict(model_id, **data)
554 @override
555 def delete(self, model_id: int | _StandardModel) -> None:
556 """
557 Delete a resource.
559 Args:
560 model_id: ID of the resource.
562 """
563 if not model_id:
564 raise ValueError("model_id is required to delete a resource")
565 if not isinstance(model_id, int):
566 model_id = model_id.id
568 # Signal before deleting resource
569 signal_params = {"resource": self.name, "model_id": model_id}
570 registry.emit("resource.delete:before", "Emitted before deleting a resource", args=[self], kwargs=signal_params)
572 if not (url := self.get_endpoint("delete", resource=self.name, pk=model_id)):
573 raise ConfigurationError(f"Delete endpoint not defined for resource {self.name}")
575 self.client.request("DELETE", url)
577 # Signal after deleting resource
578 registry.emit("resource.delete:after", "Emitted after deleting a resource", args=[self], kwargs=signal_params)
580 @override
581 def update_dict(self, model_id: int, **data: dict[str, Any]) -> _StandardModel:
582 """
583 Update a resource.
585 Args:
586 model_id: ID of the resource.
587 data: Resource data.
589 Raises:
590 ResourceNotFoundError: If the resource with the given id is not found
592 Returns:
593 The updated resource.
595 """
596 # Signal before updating resource
597 signal_params = {"resource": self.name, "model_id": model_id, "data": data}
598 registry.emit("resource.update:before", "Emitted before updating a resource", kwargs=signal_params)
600 if not (url := self.get_endpoint("update", resource=self.name, pk=model_id)):
601 raise ConfigurationError(f"Update endpoint not defined for resource {self.name}")
603 if not (response := self.client.request("PUT", url, data=data)):
604 raise ResourceNotFoundError("Resource ${resource} not found after update.", resource_name=self.name)
606 model = self.parse_to_model(response)
608 # Signal after updating resource
609 registry.emit(
610 "resource.update:after",
611 "Emitted after updating a resource",
612 args=[self],
613 kwargs={**signal_params, "model": model},
614 )
616 return model
619class BulkEditing:
620 def bulk_edit_objects( # type: ignore
621 self: BaseResource, # type: ignore
622 object_type: str,
623 ids: list[int],
624 operation: str,
625 permissions: dict[str, Any] | None = None,
626 owner_id: int | None = None,
627 merge: bool = False,
628 ) -> dict[str, Any]:
629 """
630 Bulk edit non-document objects (tags, correspondents, document types, storage paths).
632 Args:
633 object_type: Type of objects to edit ('tags', 'correspondents', 'document_types', 'storage_paths')
634 ids: List of object IDs to edit
635 operation: Operation to perform ('set_permissions' or 'delete')
636 permissions: Permissions object for 'set_permissions' operation
637 owner_id: Owner ID to assign
638 merge: Whether to merge permissions with existing ones (True) or replace them (False)
640 Returns:
641 The API response
643 Raises:
644 ValueError: If operation is not valid
645 ConfigurationError: If the bulk edit endpoint is not defined
647 """
648 if operation not in ("set_permissions", "delete"):
649 raise ValueError(f"Invalid operation '{operation}'. Must be 'set_permissions' or 'delete'")
651 # Signal before bulk action
652 signal_params = {
653 "object_type": object_type,
654 "operation": operation,
655 "ids": ids,
656 "permissions": permissions,
657 "owner_id": owner_id,
658 "merge": merge,
659 }
660 registry.emit(
661 "resource.bulk_edit_objects:before",
662 "Emitted before bulk edit objects",
663 args=[self],
664 kwargs=signal_params,
665 )
667 data: dict[str, Any] = {"objects": ids, "object_type": object_type, "operation": operation, "merge": merge}
669 if permissions:
670 data["permissions"] = permissions
671 if owner_id is not None:
672 data["owner"] = owner_id
674 # Use the special endpoint for bulk editing objects
675 url = HttpUrl(f"{self.client.base_url}/api/bulk_edit_objects/")
677 response = self.client.request("POST", url, data=data)
679 # Signal after bulk action
680 registry.emit(
681 "resource.bulk_edit_objects:after",
682 "Emitted after bulk edit objects",
683 args=[self],
684 kwargs={**signal_params, "response": response},
685 )
687 return response or {}