Coverage for src/paperap/client.py: 95%

1"""

2----------------------------------------------------------------------------

4 METADATA:

6 File: client.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

25import re

26from pathlib import Path

27from typing import TYPE_CHECKING, Any, Literal, Unpack, overload

29import requests

30from pydantic import HttpUrl

32from paperap.auth import AuthBase, BasicAuth, TokenAuth

33from paperap.exceptions import (

34 APIError,

35 AuthenticationError,

36 BadResponseError,

37 ConfigurationError,

38 InsufficientPermissionError,

39 RelationshipNotFoundError,

40 RequestError,

41 ResourceNotFoundError,

42 ResponseParsingError,

43)

44from paperap.resources import (

45 CorrespondentResource,

46 CustomFieldResource,

47 DocumentMetadataResource,

48 DocumentNoteResource,

49 DocumentResource,

50 DocumentSuggestionsResource,

51 DocumentTypeResource,

52 DownloadedDocumentResource,

53 GroupResource,

54 ProfileResource,

55 SavedViewResource,

56 ShareLinksResource,

57 StoragePathResource,

58 TagResource,

59 TaskResource,

60 UISettingsResource,

61 UserResource,

62 WorkflowActionResource,

63 WorkflowResource,

64 WorkflowTriggerResource,

65)

66from paperap.settings import Settings, SettingsArgs

67from paperap.signals import registry

69if TYPE_CHECKING:

70 from paperap.plugins.base import Plugin

71 from paperap.plugins.manager import PluginConfig

73logger = logging.getLogger(__name__)

76class PaperlessClient:

77 """

78 Client for interacting with the Paperless-NgX API.

80 Args:

81 settings: Settings object containing client configuration.

83 Examples:

84 ```python

85 # Using token authentication

86 client = PaperlessClient(

87 Settings(

88 base_url="https://paperless.example.com",

89 token="40characterslong40characterslong40charac"

90 )

91 )

93 # Using basic authentication

94 client = PaperlessClient(

95 Settings(

96 base_url="https://paperless.example.com",

97 username="user",

98 password="pass"

99 )

100 )

101

102 # Loading all settings from environment variables (e.g. PAPERLESS_TOKEN)

103 client = PaperlessClient()

104

105 # With context manager

106 with PaperlessClient(...) as client:

107 docs = client.documents.list()

108 ```

109

110 """

111

112 settings: Settings

113 auth: AuthBase

114 session: requests.Session

115 plugins: dict[str, "Plugin"]

116

117 # Resources

118 correspondents: CorrespondentResource

119 custom_fields: CustomFieldResource

120 document_types: DocumentTypeResource

121 document_metadata: DocumentMetadataResource

122 document_suggestions: DocumentSuggestionsResource

123 downloaded_documents: DownloadedDocumentResource

124 documents: DocumentResource

125 document_notes: DocumentNoteResource

126 groups: GroupResource

127 profile: ProfileResource

128 saved_views: SavedViewResource

129 share_links: ShareLinksResource

130 storage_paths: StoragePathResource

131 tags: TagResource

132 tasks: TaskResource

133 ui_settings: UISettingsResource

134 users: UserResource

135 workflow_actions: WorkflowActionResource

136 workflow_triggers: WorkflowTriggerResource

137 workflows: WorkflowResource

138

139 def __init__(self, settings: Settings | None = None, **kwargs: Unpack[SettingsArgs]) -> None:

140 if not settings:

141 # Any params not provided in kwargs will be loaded from env vars

142 settings = Settings(**kwargs)

143

144 self.settings = settings

145 # Prioritize username/password over token if both are provided

146 if self.settings.username and self.settings.password:

147 self.auth = BasicAuth(username=self.settings.username, password=self.settings.password)

148 elif self.settings.token:

149 self.auth = TokenAuth(token=self.settings.token)

150 else:

151 raise ValueError("Provide a token, or a username and password")

152

153 self.session = requests.Session()

154

155 # Set default headers

156 self.session.headers.update(

157 {

158 "Accept": "application/json; version=2",

159 # Don't set Content-Type here as it will be set appropriately per request

160 # "Content-Type": "application/json",

161 }

162 )

163

164 # Initialize resources

165 self._init_resources()

166 self._initialize_plugins()

167 super().__init__()

168

169 @property

170 def base_url(self) -> HttpUrl:

171 """Get the base URL."""

172 return self.settings.base_url

173

174 def __enter__(self) -> PaperlessClient:

175 return self

176

177 def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:

178 self.close()

179

180 def _init_resources(self) -> None:

181 """Initialize all API resources."""

182 # Initialize resources

183 self.correspondents = CorrespondentResource(self)

184 self.custom_fields = CustomFieldResource(self)

185 self.document_types = DocumentTypeResource(self)

186 self.document_metadata = DocumentMetadataResource(self)

187 self.document_suggestions = DocumentSuggestionsResource(self)

188 self.downloaded_documents = DownloadedDocumentResource(self)

189 self.documents = DocumentResource(self)

190 self.document_notes = DocumentNoteResource(self)

191 self.groups = GroupResource(self)

192 self.profile = ProfileResource(self)

193 self.saved_views = SavedViewResource(self)

194 self.share_links = ShareLinksResource(self)

195 self.storage_paths = StoragePathResource(self)

196 self.tags = TagResource(self)

197 self.tasks = TaskResource(self)

198 self.ui_settings = UISettingsResource(self)

199 self.users = UserResource(self)

200 self.workflow_actions = WorkflowActionResource(self)

201 self.workflow_triggers = WorkflowTriggerResource(self)

202 self.workflows = WorkflowResource(self)

203

204 def _initialize_plugins(self, plugin_config: "PluginConfig | None" = None) -> None:

205 """

206 Initialize plugins based on configuration.

207

208 Args:

209 plugin_config: Optional configuration dictionary for plugins.

210

211 """

212 from paperap.plugins.manager import PluginManager # pylint: disable=import-outside-toplevel

213

214 PluginManager.model_rebuild()

215

216 # Create and configure the plugin manager

217 self.manager = PluginManager(client=self)

218

219 # Discover available plugins

220 self.manager.discover_plugins()

221

222 # Configure plugins

223 plugin_config = plugin_config or {

224 "enabled_plugins": ["SampleDataCollector"],

225 "settings": {

226 "SampleDataCollector": {

227 "test_dir": str(Path(__file__).parents[3] / "tests/sample_data"),

228 },

229 },

230 }

231 self.manager.configure(plugin_config)

232

233 # Initialize all enabled plugins

234 self.plugins = self.manager.initialize_all_plugins()

235

236 def _get_auth_params(self) -> dict[str, Any]:

237 """Get authentication parameters for requests."""

238 return self.auth.get_auth_params()

239

240 def get_headers(self) -> dict[str, str]:

241 """Get headers for requests."""

242 headers = {}

243

244 headers.update(self.auth.get_auth_headers())

245

246 return headers

247

248 def close(self) -> None:

249 """Close the client and release resources."""

250 if hasattr(self, "session"):

251 self.session.close()

252

253 def request_raw(

254 self,

255 method: str,

256 endpoint: str | HttpUrl,

257 *,

258 params: dict[str, Any] | None = None,

259 data: dict[str, Any] | None = None,

260 files: dict[str, Any] | None = None,

261 ) -> requests.Response | None:

262 """

263 Make a request to the Paperless-NgX API.

264

265 Args:

266 method: HTTP method (GET, POST, PUT, DELETE).

267 endpoint: API endpoint relative to base URL.

268 params: Query parameters for the request.

269 data: Request body data.

270 files: Files to upload.

271 json_response: Whether to parse the response as JSON.

272

273 Returns:

274 Response object or None if no content.

275

276 Raises:

277 AuthenticationError: If authentication fails.

278 ResourceNotFoundError: If the requested resource doesn't exist.

279 APIError: If the API returns an error.

280 PaperapError: For other errors.

281

282 """

283 if isinstance(endpoint, HttpUrl):

284 # Use URL object directly

285 url = str(endpoint)

286 elif isinstance(endpoint, str):

287 if endpoint.startswith("http"):

288 url = endpoint

289 else:

290 url = f"{self.base_url}{endpoint.lstrip('/')}"

291 else:

292 url = f"{self.base_url}{str(endpoint).lstrip('/')}"

293

294 logger.debug("Requesting %s %s", method, url)

295

296 # Add headers from authentication and session defaults

297 headers = {**self.session.headers, **self.get_headers()}

298

299 # Set the appropriate Content-Type header based on the request type

300 if files:

301 # For file uploads, let requests set the multipart/form-data Content-Type with boundary

302 headers.pop("Content-Type", None)

303 elif "Content-Type" not in headers:

304 # For JSON requests, explicitly set the Content-Type

305 headers["Content-Type"] = "application/json"

306

307 try:

308 # TODO: Temporary hack

309 params = params.get("params", params) if params else params

310

311 logger.debug(

312 "Request (%s) url %s, params %s, data %s, files %s, headers %s",

313 method,

314 url,

315 params,

316 data,

317 files,

318 headers,

319 )

320 # When uploading files, we need to pass data as form data, not JSON

321 # The key difference is that with files, we MUST use data parameter, not json

322 if files:

323 # For file uploads, use data parameter (not json) to ensure proper multipart/form-data encoding

324 response = self.session.request(

325 method=method,

326 url=url,

327 headers=headers,

328 params=params,

329 data=data, # Use data for form fields with files

330 files=files,

331 timeout=self.settings.timeout,

332 **self._get_auth_params(),

333 )

334 else:

335 # For regular JSON requests

336 response = self.session.request(

337 method=method,

338 url=url,

339 headers=headers,

340 params=params,

341 json=data, # Use json for regular requests

342 timeout=self.settings.timeout,

343 **self._get_auth_params(),

344 )

345

346 # Handle HTTP errors

347 if response.status_code >= 400:

348 return self._handle_request_errors(response, url, params=params, data=data, files=files)

349

350 # No content

351 if response.status_code == 204:

352 return None

353

354 except requests.exceptions.ConnectionError as ce:

355 logger.error(

356 "Unable to connect to Paperless server: %s url %s, params %s, data %s, files %s",

357 method,

358 url,

359 params,

360 data,

361 files,

362 )

363 raise RequestError(f"Connection error: {str(ce)}") from ce

364 except requests.exceptions.RequestException as re:

365 raise RequestError(f"Request failed: {str(re)}") from re

366

367 return response

368

369 def _handle_request_errors(

370 self,

371 response: requests.Response,

372 url: str,

373 *,

374 params: dict[str, Any] | None = None,

375 data: dict[str, Any] | None = None,

376 files: dict[str, Any] | None = None,

377 ) -> None:

378 error_message = self._extract_error_message(response)

379

380 if response.status_code == 400:

381 if "This field is required" in error_message:

382 raise ValueError(f"Required field missing: {error_message}")

383 if matches := re.match(r"([a-zA-Z_-]+): Invalid pk", error_message):

384 raise RelationshipNotFoundError(f"Invalid relationship {matches.group(1)}: {error_message}")

385 if response.status_code == 401:

386 raise AuthenticationError(f"Authentication failed: {error_message}")

387 if response.status_code == 403:

388 if "this site requires a CSRF" in error_message:

389 raise ConfigurationError(f"Response claims CSRF token required. Is the url correct? {url}")

390 raise InsufficientPermissionError(f"Permission denied: {error_message}")

391 if response.status_code == 404:

392 raise ResourceNotFoundError(f"Paperless returned 404 for {url}")

393

394 # All else...

395 raise BadResponseError(error_message, response.status_code)

396

397 @overload

398 def _handle_response(self, response: requests.Response, *, json_response: Literal[True] = True) -> dict[str, Any]: ...

399

400 @overload

401 def _handle_response(self, response: None, *, json_response: bool = True) -> None: ...

402

403 @overload

404 def _handle_response(self, response: requests.Response | None, *, json_response: Literal[False]) -> bytes | None: ...

405

406 @overload

407 def _handle_response(self, response: requests.Response | None, *, json_response: bool = True) -> dict[str, Any] | bytes | None: ...

408

409 def _handle_response(self, response: requests.Response | None, *, json_response: bool = True) -> dict[str, Any] | bytes | None:

410 """Handle the response based on the content type."""

411 if response is None:

412 return None

413

414 # Try to parse as JSON if requested

415 if json_response:

416 try:

417 return response.json() # type: ignore # mypy can't infer the return type correctly

418 except ValueError as e:

419 url = getattr(response, "url", "unknown URL")

420 logger.error("Failed to parse JSON response: %s -> url %s -> content: %s", e, url, response.content)

421 raise ResponseParsingError(f"Failed to parse JSON response: {str(e)} -> url {url}") from e

422

423 return response.content

424

425 @overload

426 def request(

427 self,

428 method: str,

429 endpoint: str | HttpUrl,

430 *,

431 params: dict[str, Any] | None = None,

432 data: dict[str, Any] | None = None,

433 files: dict[str, Any] | None = None,

434 ) -> dict[str, Any] | None: ...

435

436 @overload

437 def request(

438 self,

439 method: str,

440 endpoint: str | HttpUrl,

441 *,

442 params: dict[str, Any] | None = None,

443 data: dict[str, Any] | None = None,

444 files: dict[str, Any] | None = None,

445 json_response: Literal[False],

446 ) -> bytes | None: ...

447

448 @overload

449 def request(

450 self,

451 method: str,

452 endpoint: str | HttpUrl,

453 *,

454 params: dict[str, Any] | None = None,

455 data: dict[str, Any] | None = None,

456 files: dict[str, Any] | None = None,

457 json_response: bool = True,

458 ) -> dict[str, Any] | bytes | None: ...

459

460 def request(

461 self,

462 method: str,

463 endpoint: str | HttpUrl,

464 *,

465 params: dict[str, Any] | None = None,

466 data: dict[str, Any] | None = None,

467 files: dict[str, Any] | None = None,

468 json_response: bool = True,

469 ) -> dict[str, Any] | bytes | None:

470 """

471 Make a request to the Paperless-NgX API.

472

473 Generally, this should be done using resources, not by calling this method directly.

474

475 Args:

476 method: HTTP method (GET, POST, PUT, DELETE).

477 endpoint: API endpoint relative to base URL.

478 params: Query parameters for the request.

479 data: Request body data.

480 files: Files to upload.

481 json_response: Whether to parse the response as JSON.

482

483 Returns:

484 Parsed response data.

485

486 """

487 kwargs = {

488 "client": self,

489 "method": method,

490 "endpoint": endpoint,

491 "params": params,

492 "data": data,

493 "files": files,

494 "json_response": json_response,

495 }

496

497 registry.emit("client.request:before", "Before a request is sent to the Paperless server", args=[self], kwargs=kwargs)

498

499 if not (response := self.request_raw(method, endpoint, params=params, data=data, files=files)):

500 return None

501

502 registry.emit(

503 "client.request__response",

504 "After a response is received, before it is parsed",

505 args=[response],

506 kwargs=kwargs,

507 )

508

509 parsed_response = self._handle_response(response, json_response=json_response)

510 parsed_response = registry.emit(

511 "client.request:after",

512 "After a request is parsed.",

513 args=parsed_response,

514 kwargs=kwargs,

515 )

516

517 return parsed_response

518

519 def _extract_error_message(self, response: requests.Response) -> str:

520 """Extract error message from response."""

521 try:

522 error_data = response.json()

523 if isinstance(error_data, dict):

524 # Try different possible error formats

525 if "detail" in error_data:

526 return str(error_data["detail"])

527 if "error" in error_data:

528 return str(error_data["error"])

529 if "non_field_errors" in error_data:

530 return ", ".join(error_data["non_field_errors"])

531

532 # Handle nested error messages

533 messages = []

534 for key, value in error_data.items():

535 if isinstance(value, list):

536 values = [str(i) for i in value]

537 messages.append(f"{key}: {', '.join(values)}")

538 else:

539 messages.append(f"{key}: {value}")

540 return "; ".join(messages)

541 return str(error_data)

542 except ValueError:

543 return response.text or f"HTTP {response.status_code}"

544

545 def generate_token(

546 self,

547 base_url: str,

548 username: str,

549 password: str,

550 timeout: int | None = None,

551 ) -> str:

552 """

553 Generate an API token using username and password.

554

555 Args:

556 base_url: The base URL of the Paperless-NgX instance.

557 username: Username for authentication.

558 password: Password for authentication.

559 timeout: Request timeout in seconds.

560

561 Returns:

562 Generated API token.

563

564 Raises:

565 AuthenticationError: If authentication fails.

566 PaperapError: For other errors.

567

568 """

569 if timeout is None:

570 timeout = self.settings.timeout

571

572 if not base_url.startswith(("http://", "https://")):

573 base_url = f"https://{base_url}"

574

575 url = f"{base_url.rstrip('/')}/api/token/"

576

577 registry.emit(

578 "client.generate_token__before",

579 "Before a new token is generated",

580 kwargs={"url": url, "username": username},

581 )

582

583 try:

584 response = requests.post(

585 url,

586 json={"username": username, "password": password},

587 headers={"Accept": "application/json"},

588 timeout=timeout,

589 )

590

591 response.raise_for_status()

592 data = response.json()

593

594 registry.emit(

595 "client.generate_token__after",

596 "After a new token is generated",

597 kwargs={"url": url, "username": username, "response": data},

598 )

599

600 if "token" not in data:

601 raise ResponseParsingError("Token not found in response")

602

603 return str(data["token"])

604 except requests.exceptions.HTTPError as he:

605 if he.response.status_code == 401:

606 raise AuthenticationError("Invalid username or password") from he

607 try:

608 error_data = he.response.json()

609 error_message = error_data.get("detail", str(he))

610 except (ValueError, KeyError):

611 error_message = str(he)

612

613 raise RequestError(f"Failed to generate token: {error_message}") from he

614 except requests.exceptions.RequestException as re:

615 raise RequestError(f"Error while requesting a new token: {str(re)}") from re

616 except (ValueError, KeyError) as ve:

617 raise ResponseParsingError(f"Failed to parse response when generating token: {str(ve)}") from ve

618

619 def get_statistics(self) -> dict[str, Any]:

620 """

621 Get system statistics.

622

623 Returns:

624 Dictionary containing system statistics.

625

626 """

627 if result := self.request("GET", "api/statistics/"):

628 return result

629 raise APIError("Failed to get statistics")

630

631 def get_system_status(self) -> dict[str, Any]:

632 """

633 Get system status.

634

635 Returns:

636 Dictionary containing system status information.

637

638 """

639 if result := self.request("GET", "api/status/"):

640 return result

641 raise APIError("Failed to get system status")

642

643 def get_config(self) -> dict[str, Any]:

644 """

645 Get system configuration.

646

647 Returns:

648 Dictionary containing system configuration.

649

650 """

651 if result := self.request("GET", "api/config/"):

652 return result

653 raise APIError("Failed to get system configuration")