Coverage for src/paperap/models/document/queryset.py: 97%

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

14----------------------------------------------------------------------------

16 LAST MODIFIED:

18 2025-03-04 By Jess Mann

20"""

22from __future__ import annotations

24import logging

25from datetime import datetime

26from functools import singledispatchmethod

27from typing import TYPE_CHECKING, Any, NamedTuple, Self, Union, overload

29from paperap.models.abstract.queryset import BaseQuerySet, StandardQuerySet

30from paperap.models.mixins.queryset import HasOwner

32if TYPE_CHECKING:

33 from paperap.models.correspondent.model import Correspondent

34 from paperap.models.document.model import Document, DocumentNote

36logger = logging.getLogger(__name__)

38_OperationType = Union[str, "_QueryParam"]

39_QueryParam = Union["CustomFieldQuery", tuple[str, _OperationType, Any]]

42class CustomFieldQuery(NamedTuple):

43 field: str

44 operation: _OperationType

45 value: Any

48class DocumentNoteQuerySet(StandardQuerySet["DocumentNote"]):

49 pass

52class DocumentQuerySet(StandardQuerySet["Document"], HasOwner):

53 """

54 QuerySet for Paperless-ngx documents with specialized filtering methods.

56 Examples:

57 >>> # Search for documents

58 >>> docs = client.documents().search("invoice")

59 >>> for doc in docs:

60 ... print(doc.title)

62 >>> # Find documents similar to a specific document

63 >>> similar_docs = client.documents().more_like(42)

64 >>> for doc in similar_docs:

65 ... print(doc.title)

67 """

69 def tag_id(self, tag_id: int | list[int]) -> Self:

70 """

71 Filter documents that have the specified tag ID(s).

73 Args:

74 tag_id: A single tag ID or list of tag IDs

76 Returns:

77 Filtered DocumentQuerySet

79 """

80 if isinstance(tag_id, list):

81 return self.filter(tags__id__in=tag_id)

82 return self.filter(tags__id=tag_id)

84 def tag_name(self, tag_name: str, *, exact: bool = True, case_insensitive: bool = True) -> Self:

85 """

86 Filter documents that have a tag with the specified name.

88 Args:

89 tag_name: The name of the tag

90 exact: If True, match the exact tag name, otherwise use contains

92 Returns:

93 Filtered DocumentQuerySet

95 """

96 return self.filter_field_by_str("tags__name", tag_name, exact=exact, case_insensitive=case_insensitive)

98 def title(self, title: str, *, exact: bool = True, case_insensitive: bool = True) -> Self:

99 """

100 Filter documents by title.

101

102 Args:

103 title: The document title to filter by

104 exact: If True, match the exact title, otherwise use contains

105

106 Returns:

107 Filtered DocumentQuerySet

108

109 """

110 return self.filter_field_by_str("title", title, exact=exact, case_insensitive=case_insensitive)

111

112 def search(self, query: str) -> "DocumentQuerySet":

113 """

114 Search for documents using a query string.

115

116 Args:

117 query: The search query.

118

119 Returns:

120 A queryset with the search results.

121

122 Examples:

123 >>> docs = client.documents().search("invoice")

124 >>> for doc in docs:

125 ... print(doc.title)

126

127 """

128 return self.filter(query=query)

129

130 def more_like(self, document_id: int) -> "DocumentQuerySet":

131 """

132 Find documents similar to the specified document.

133

134 Args:

135 document_id: The ID of the document to find similar documents for.

136

137 Returns:

138 A queryset with similar documents.

139

140 Examples:

141 >>> similar_docs = client.documents().more_like(42)

142 >>> for doc in similar_docs:

143 ... print(doc.title)

144

145 """

146 return self.filter(more_like_id=document_id)

147

148 def correspondent(self, value: int | str | None = None, *, exact: bool = True, case_insensitive: bool = True, **kwargs: Any) -> Self:

149 """

150 Filter documents by correspondent.

151

152 Any number of filter arguments can be provided, but at least one must be specified.

153

154 Args:

155 value: The correspondent ID or name to filter by

156 exact: If True, match the exact value, otherwise use contains

157 **kwargs: Additional filters (slug, id, name)

158

159 Returns:

160 Filtered DocumentQuerySet

161

162 Raises:

163 ValueError: If no valid filters are provided

164

165 Examples:

166 # Filter by ID

167 client.documents().all().correspondent(1)

168 client.documents().all().correspondent(id=1)

169

170 # Filter by name

171 client.documents().all().correspondent("John Doe")

172 client.documents().all().correspondent(name="John Doe")

173

174 # Filter by name (exact match)

175 client.documents().all().correspondent("John Doe", exact=True)

176 client.documents().all().correspondent(name="John Doe", exact=True)

177

178 # Filter by slug

179 client.documents().all().correspondent(slug="john-doe")

180

181 # Filter by ID and name

182 client.documents().all().correspondent(1, name="John Doe")

183 client.documents().all().correspondent(id=1, name="John Doe")

184 client.documents().all().correspondent("John Doe", id=1)

185

186 """

187 # Track if any filters were applied

188 filters_applied = False

189 result = self

190

191 if value is not None:

192 if isinstance(value, int):

193 result = self.correspondent_id(value)

194 filters_applied = True

195 elif isinstance(value, str):

196 result = self.correspondent_name(value, exact=exact, case_insensitive=case_insensitive)

197 filters_applied = True

198 else:

199 raise TypeError("Invalid value type for correspondent filter")

200

201 if (slug := kwargs.get("slug")) is not None:

202 result = result.correspondent_slug(slug, exact=exact, case_insensitive=case_insensitive)

203 filters_applied = True

204 if (pk := kwargs.get("id")) is not None:

205 result = result.correspondent_id(pk)

206 filters_applied = True

207 if (name := kwargs.get("name")) is not None:

208 result = result.correspondent_name(name, exact=exact, case_insensitive=case_insensitive)

209 filters_applied = True

210

211 # If no filters have been applied, raise an error

212 if not filters_applied:

213 raise ValueError("No valid filters provided for correspondent")

214

215 return result

216

217 def correspondent_id(self, correspondent_id: int) -> Self:

218 """

219 Filter documents by correspondent ID.

220

221 Args:

222 correspondent_id: The correspondent ID to filter by

223

224 Returns:

225 Filtered DocumentQuerySet

226

227 """

228 return self.filter(correspondent__id=correspondent_id)

229

230 def correspondent_name(self, name: str, *, exact: bool = True, case_insensitive: bool = True) -> Self:

231 """

232 Filter documents by correspondent name.

233

234 Args:

235 name: The correspondent name to filter by

236 exact: If True, match the exact name, otherwise use contains

237

238 Returns:

239 Filtered DocumentQuerySet

240

241 """

242 return self.filter_field_by_str("correspondent__name", name, exact=exact, case_insensitive=case_insensitive)

243

244 def correspondent_slug(self, slug: str, *, exact: bool = True, case_insensitive: bool = True) -> Self:

245 """

246 Filter documents by correspondent slug.

247

248 Args:

249 slug: The correspondent slug to filter by

250 exact: If True, match the exact slug, otherwise use contains

251

252 Returns:

253 Filtered DocumentQuerySet

254

255 """

256 return self.filter_field_by_str("correspondent__slug", slug, exact=exact, case_insensitive=case_insensitive)

257

258 def document_type(self, value: int | str | None = None, *, exact: bool = True, case_insensitive: bool = True, **kwargs: Any) -> Self:

259 """

260 Filter documents by document type.

261

262 Any number of filter arguments can be provided, but at least one must be specified.

263

264 Args:

265 value: The document type ID or name to filter by

266 exact: If True, match the exact value, otherwise use contains

267 **kwargs: Additional filters (id, name)

268

269 Returns:

270 Filtered DocumentQuerySet

271

272 Raises:

273 ValueError: If no valid filters are provided

274

275 Examples:

276 # Filter by ID

277 client.documents().all().document_type(1)

278 client.documents().all().document_type(id=1)

279

280 # Filter by name

281 client.documents().all().document_type("Invoice")

282 client.documents().all().document_type(name="Invoice")

283

284 # Filter by name (exact match)

285 client.documents().all().document_type("Invoice", exact=True)

286 client.documents().all().document_type(name="Invoice", exact=True)

287

288 # Filter by ID and name

289 client.documents().all().document_type(1, name="Invoice")

290 client.documents().all().document_type(id=1, name="Invoice")

291 client.documents().all().document_type("Invoice", id=1)

292

293 """

294 # Track if any filters were applied

295 filters_applied = False

296 result = self

297

298 if value is not None:

299 if isinstance(value, int):

300 result = self.document_type_id(value)

301 filters_applied = True

302 elif isinstance(value, str):

303 result = self.document_type_name(value, exact=exact, case_insensitive=case_insensitive)

304 filters_applied = True

305 else:

306 raise TypeError("Invalid value type for document type filter")

307

308 if (pk := kwargs.get("id")) is not None:

309 result = result.document_type_id(pk)

310 filters_applied = True

311 if (name := kwargs.get("name")) is not None:

312 result = result.document_type_name(name, exact=exact, case_insensitive=case_insensitive)

313 filters_applied = True

314

315 # If no filters have been applied, raise an error

316 if not filters_applied:

317 raise ValueError("No valid filters provided for document type")

318

319 return result

320

321 def document_type_id(self, document_type_id: int) -> Self:

322 """

323 Filter documents by document type ID.

324

325 Args:

326 document_type_id: The document type ID to filter by

327

328 Returns:

329 Filtered DocumentQuerySet

330

331 """

332 return self.filter(document_type__id=document_type_id)

333

334 def document_type_name(self, name: str, *, exact: bool = True, case_insensitive: bool = True) -> Self:

335 """

336 Filter documents by document type name.

337

338 Args:

339 name: The document type name to filter by

340 exact: If True, match the exact name, otherwise use contains

341

342 Returns:

343 Filtered DocumentQuerySet

344

345 """

346 return self.filter_field_by_str("document_type__name", name, exact=exact, case_insensitive=case_insensitive)

347

348 def storage_path(self, value: int | str | None = None, *, exact: bool = True, case_insensitive: bool = True, **kwargs: Any) -> Self:

349 """

350 Filter documents by storage path.

351

352 Any number of filter arguments can be provided, but at least one must be specified.

353

354 Args:

355 value: The storage path ID or name to filter by

356 exact: If True, match the exact value, otherwise use contains

357 **kwargs: Additional filters (id, name)

358

359 Returns:

360 Filtered DocumentQuerySet

361

362 Raises:

363 ValueError: If no valid filters are provided

364

365 Examples:

366 # Filter by ID

367 client.documents().all().storage_path(1)

368 client.documents().all().storage_path(id=1)

369

370 # Filter by name

371 client.documents().all().storage_path("Invoices")

372 client.documents().all().storage_path(name="Invoices")

373

374 # Filter by name (exact match)

375 client.documents().all().storage_path("Invoices", exact=True)

376 client.documents().all().storage_path(name="Invoices", exact=True)

377

378 # Filter by ID and name

379 client.documents().all().storage_path(1, name="Invoices")

380 client.documents().all().storage_path(id=1, name="Invoices")

381 client.documents().all().storage_path("Invoices", id=1)

382

383 """

384 # Track if any filters were applied

385 filters_applied = False

386 result = self

387

388 if value is not None:

389 if isinstance(value, int):

390 result = self.storage_path_id(value)

391 filters_applied = True

392 elif isinstance(value, str):

393 result = self.storage_path_name(value, exact=exact, case_insensitive=case_insensitive)

394 filters_applied = True

395 else:

396 raise TypeError("Invalid value type for storage path filter")

397

398 if (pk := kwargs.get("id")) is not None:

399 result = result.storage_path_id(pk)

400 filters_applied = True

401 if (name := kwargs.get("name")) is not None:

402 result = result.storage_path_name(name, exact=exact, case_insensitive=case_insensitive)

403 filters_applied = True

404

405 # If no filters have been applied, raise an error

406 if not filters_applied:

407 raise ValueError("No valid filters provided for storage path")

408

409 return result

410

411 def storage_path_id(self, storage_path_id: int) -> Self:

412 """

413 Filter documents by storage path ID.

414

415 Args:

416 storage_path_id: The storage path ID to filter by

417

418 Returns:

419 Filtered DocumentQuerySet

420

421 """

422 return self.filter(storage_path__id=storage_path_id)

423

424 def storage_path_name(self, name: str, *, exact: bool = True, case_insensitive: bool = True) -> Self:

425 """

426 Filter documents by storage path name.

427

428 Args:

429 name: The storage path name to filter by

430 exact: If True, match the exact name, otherwise use contains

431

432 Returns:

433 Filtered DocumentQuerySet

434

435 """

436 return self.filter_field_by_str("storage_path__name", name, exact=exact, case_insensitive=case_insensitive)

437

438 def content(self, text: str) -> Self:

439 """

440 Filter documents whose content contains the specified text.

441

442 Args:

443 text: The text to search for in document content

444

445 Returns:

446 Filtered DocumentQuerySet

447

448 """

449 return self.filter(content__contains=text)

450

451 def added_after(self, date_str: str) -> Self:

452 """

453 Filter documents added after the specified date.

454

455 Args:

456 date_str: ISO format date string (YYYY-MM-DD)

457

458 Returns:

459 Filtered DocumentQuerySet

460

461 """

462 return self.filter(added__gt=date_str)

463

464 def added_before(self, date_str: str) -> Self:

465 """

466 Filter documents added before the specified date.

467

468 Args:

469 date_str: ISO format date string (YYYY-MM-DD)

470

471 Returns:

472 Filtered DocumentQuerySet

473

474 """

475 return self.filter(added__lt=date_str)

476

477 def asn(self, value: str, *, exact: bool = True, case_insensitive: bool = True) -> Self:

478 """

479 Filter documents by archive serial number.

480

481 Args:

482 value: The archive serial number to filter by

483 exact: If True, match the exact value, otherwise use contains

484

485 Returns:

486 Filtered DocumentQuerySet

487

488 """

489 return self.filter_field_by_str("asn", value, exact=exact, case_insensitive=case_insensitive)

490

491 def original_filename(self, name: str, *, exact: bool = True, case_insensitive: bool = True) -> Self:

492 """

493 Filter documents by original file name.

494

495 Args:

496 name: The original file name to filter by

497 exact: If True, match the exact name, otherwise use contains

498

499 Returns:

500 Filtered DocumentQuerySet

501

502 """

503 return self.filter_field_by_str("original_filename", name, exact=exact, case_insensitive=case_insensitive)

504

505 def user_can_change(self, value: bool) -> Self:

506 """

507 Filter documents by user change permission.

508

509 Args:

510 value: True to filter documents the user can change

511

512 Returns:

513 Filtered DocumentQuerySet

514

515 """

516 return self.filter(user_can_change=value)

517

518 def custom_field_fullsearch(self, value: str, *, case_insensitive: bool = True) -> Self:

519 """

520 Filter documents by searching through both custom field name and value.

521

522 Args:

523 value: The search string

524

525 Returns:

526 Filtered DocumentQuerySet

527

528 """

529 if case_insensitive:

530 return self.filter(custom_fields__icontains=value)

531 raise NotImplementedError("Case-sensitive custom field search is not supported by Paperless NGX")

532

533 def custom_field(self, field: str, value: Any, *, exact: bool = False, case_insensitive: bool = True) -> Self:

534 """

535 Filter documents by custom field.

536

537 Args:

538 field: The name of the custom field

539 value: The value to filter by

540 exact: If True, match the exact value, otherwise use contains

541

542 Returns:

543 Filtered DocumentQuerySet

544

545 """

546 if exact:

547 if case_insensitive:

548 return self.custom_field_query(field, "iexact", value)

549 return self.custom_field_query(field, "exact", value)

550 if case_insensitive:

551 return self.custom_field_query(field, "icontains", value)

552 return self.custom_field_query(field, "contains", value)

553

554 def has_custom_field_id(self, pk: int | list[int], *, exact: bool = False) -> Self:

555 """

556 Filter documents that have a custom field with the specified ID(s).

557

558 Args:

559 pk: A single custom field ID or list of custom field IDs

560 exact: If True, return results that have exactly these ids and no others

561

562 Returns:

563 Filtered DocumentQuerySet

564

565 """

566 if exact:

567 return self.filter(custom_fields__id__all=pk)

568 return self.filter(custom_fields__id__in=pk)

569

570 def _normalize_custom_field_query_item(self, value: Any) -> str:

571 if isinstance(value, tuple):

572 # Check if it's a CustomFieldQuery

573 try:

574 converted_value = CustomFieldQuery(*value)

575 return self._normalize_custom_field_query(converted_value)

576 except TypeError:

577 # It's a tuple, not a CustomFieldQuery

578 pass

579

580 if isinstance(value, str):

581 return f'"{value}"'

582 if isinstance(value, (list, tuple)):

583 values = [str(self._normalize_custom_field_query_item(v)) for v in value]

584 return f"[{', '.join(values)}]"

585 if isinstance(value, bool):

586 return str(value).lower()

587

588 return str(value)

589

590 def _normalize_custom_field_query(self, query: _QueryParam) -> str:

591 try:

592 if not isinstance(query, CustomFieldQuery):

593 query = CustomFieldQuery(*query)

594 except TypeError as te:

595 raise TypeError("Invalid custom field query format") from te

596

597 field, operation, value = query

598 operation = self._normalize_custom_field_query_item(operation)

599 value = self._normalize_custom_field_query_item(value)

600 return f'["{field}", {operation}, {value}]'

601

602 @overload

603 def custom_field_query(self, query: _QueryParam) -> Self:

604 """

605 Filter documents by custom field query.

606

607 Args:

608 query: A list representing a custom field query

609

610 Returns:

611 Filtered DocumentQuerySet

612

613 """

614 ...

615

616 @overload

617 def custom_field_query(self, field: str, operation: _OperationType, value: Any) -> Self:

618 """

619 Filter documents by custom field query.

620

621 Args:

622 field: The name of the custom field

623 operation: The operation to perform

624 value: The value to filter by

625

626 Returns:

627 Filtered DocumentQuerySet

628

629 """

630 ...

631

632 @singledispatchmethod # type: ignore # mypy does not handle singledispatchmethod with overloads correctly

633 def custom_field_query(self, *args: Any, **kwargs: Any) -> Self:

634 """

635 Filter documents by custom field query.

636 """

637 raise TypeError("Invalid custom field query format")

638

639 @custom_field_query.register # type: ignore # mypy does not handle singledispatchmethod with overloads correctly

640 def _(self, query: CustomFieldQuery) -> Self:

641 query_str = self._normalize_custom_field_query(query)

642 return self.filter(custom_field_query=query_str)

643

644 @custom_field_query.register # type: ignore # mypy does not handle singledispatchmethod with overloads correctly

645 def _(self, field: str, operation: str | CustomFieldQuery | tuple[str, Any, Any], value: Any) -> Self:

646 query = CustomFieldQuery(field, operation, value)

647 query_str = self._normalize_custom_field_query(query)

648 return self.filter(custom_field_query=query_str)

649

650 def custom_field_range(self, field: str, start: str, end: str) -> Self:

651 """

652 Filter documents with a custom field value within a specified range.

653

654 Args:

655 field: The name of the custom field

656 start: The start value of the range

657 end: The end value of the range

658

659 Returns:

660 Filtered DocumentQuerySet

661

662 """

663 return self.custom_field_query(field, "range", [start, end])

664

665 def custom_field_exact(self, field: str, value: Any) -> Self:

666 """

667 Filter documents with a custom field value that matches exactly.

668

669 Args:

670 field: The name of the custom field

671 value: The exact value to match

672

673 Returns:

674 Filtered DocumentQuerySet

675

676 """

677 return self.custom_field_query(field, "exact", value)

678

679 def custom_field_in(self, field: str, values: list[Any]) -> Self:

680 """

681 Filter documents with a custom field value in a list of values.

682

683 Args:

684 field: The name of the custom field

685 values: The list of values to match

686

687 Returns:

688 Filtered DocumentQuerySet

689

690 """

691 return self.custom_field_query(field, "in", values)

692

693 def custom_field_isnull(self, field: str) -> Self:

694 """

695 Filter documents with a custom field that is null or empty.

696

697 Args:

698 field: The name of the custom field

699

700 Returns:

701 Filtered DocumentQuerySet

702

703 """

704 return self.custom_field_query("OR", (field, "isnull", True), [field, "exact", ""])

705

706 def custom_field_exists(self, field: str, exists: bool = True) -> Self:

707 """

708 Filter documents based on the existence of a custom field.

709

710 Args:

711 field: The name of the custom field

712 exists: True to filter documents where the field exists, False otherwise

713

714 Returns:

715 Filtered DocumentQuerySet

716

717 """

718 return self.custom_field_query(field, "exists", exists)

719

720 def custom_field_contains(self, field: str, values: list[Any]) -> Self:

721 """

722 Filter documents with a custom field that contains all specified values.

723

724 Args:

725 field: The name of the custom field

726 values: The list of values that the field should contain

727

728 Returns:

729 Filtered DocumentQuerySet

730

731 """

732 return self.custom_field_query(field, "contains", values)

733

734 def has_custom_fields(self) -> Self:

735 """

736 Filter documents that have custom fields.

737 """

738 return self.filter(has_custom_fields=True)

739

740 def no_custom_fields(self) -> Self:

741 """

742 Filter documents that do not have custom fields.

743 """

744 return self.filter(has_custom_fields=False)

745

746 def notes(self, text: str) -> Self:

747 """

748 Filter documents whose notes contain the specified text.

749

750 Args:

751 text: The text to search for in document notes

752

753 Returns:

754 Filtered DocumentQuerySet

755

756 """

757 return self.filter(notes__contains=text)

758

759 def created_before(self, date: datetime | str) -> Self:

760 """

761 Filter models created before a given date.

762

763 Args:

764 date: The date to filter by

765

766 Returns:

767 Filtered QuerySet

768

769 """

770 if isinstance(date, datetime):

771 return self.filter(created__lt=date.strftime("%Y-%m-%d"))

772 return self.filter(created__lt=date)

773

774 def created_after(self, date: datetime | str) -> Self:

775 """

776 Filter models created after a given date.

777

778 Args:

779 date: The date to filter by

780

781 Returns:

782 Filtered QuerySet

783

784 """

785 if isinstance(date, datetime):

786 return self.filter(created__gt=date.strftime("%Y-%m-%d"))

787 return self.filter(created__gt=date)

788

789 def created_between(self, start: datetime | str, end: datetime | str) -> Self:

790 """

791 Filter models created between two dates.

792

793 Args:

794 start: The start date to filter by

795 end: The end date to filter by

796

797 Returns:

798 Filtered QuerySet

799

800 """

801 if isinstance(start, datetime):

802 start = start.strftime("%Y-%m-%d")

803 if isinstance(end, datetime):

804 end = end.strftime("%Y-%m-%d")

805

806 return self.filter(created__range=(start, end))