32 lines
1.2 KiB
Python
32 lines
1.2 KiB
Python
import json
|
|
from urllib import request
|
|
from urllib.parse import urljoin
|
|
|
|
from app.config import settings
|
|
|
|
|
|
class PaperlessClient:
|
|
def __init__(self) -> None:
|
|
self.base_url = settings.paperless_base_url.rstrip('/') + '/'
|
|
self.token = settings.paperless_api_token
|
|
|
|
def _req(self, method: str, path: str, payload: dict | None = None) -> dict:
|
|
url = urljoin(self.base_url, path.lstrip('/'))
|
|
body = None if payload is None else json.dumps(payload).encode('utf-8')
|
|
headers = {
|
|
'Authorization': f'Token {self.token}',
|
|
'Accept': 'application/json',
|
|
}
|
|
if body is not None:
|
|
headers['Content-Type'] = 'application/json'
|
|
req = request.Request(url, data=body, headers=headers, method=method)
|
|
with request.urlopen(req, timeout=120) as resp:
|
|
raw = resp.read().decode('utf-8', errors='replace')
|
|
return json.loads(raw) if raw else {}
|
|
|
|
def get_document(self, document_id: int | str) -> dict:
|
|
return self._req('GET', f'/api/documents/{document_id}/')
|
|
|
|
def add_note(self, document_id: int | str, note: str) -> dict:
|
|
return self._req('POST', f'/api/documents/{document_id}/notes/', {'note': note})
|