63 lines
2.9 KiB
Python
63 lines
2.9 KiB
Python
from pathlib import Path
|
|
import requests # pyright: ignore[reportMissingModuleSource]
|
|
from typing import Any
|
|
|
|
class BaserowBackend:
|
|
def __init__(self, host: str, token: str):
|
|
self.host = host.rstrip('/')
|
|
self.token = token
|
|
self.headers = {
|
|
"Authorization": f"Token {self.token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
def get_record(self, table_id: str, record_id: int) -> dict[str, Any]:
|
|
url = f"{self.host}/api/database/rows/table/{table_id}/{record_id}/"
|
|
response = requests.get(url, headers=self.headers, params={"user_field_names": "true"})
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def upsert_record(self, table_id: str, record_id: int, fields: dict[str, Any]) -> dict[str, Any]:
|
|
try:
|
|
self.get_record(table_id, record_id)
|
|
return self.update_record(table_id, record_id, fields)
|
|
except (requests.exceptions.HTTPError, ValueError):
|
|
return self.create_record(table_id, fields)
|
|
|
|
def create_record(self, table_id: str, fields: dict[str, Any]) -> dict[str, Any]:
|
|
url = f"{self.host}/api/database/rows/table/{table_id}/"
|
|
response = requests.post(url, headers=self.headers, json=fields, params={"user_field_names": "true"})
|
|
try:
|
|
response.raise_for_status()
|
|
except requests.exceptions.HTTPError as e:
|
|
print(f"Baserow Create Error: {response.status_code} - {response.text}")
|
|
raise e
|
|
return response.json()
|
|
|
|
def update_record(self, table_id: str, record_id: int, fields: dict[str, Any]) -> dict[str, Any]:
|
|
url = f"{self.host}/api/database/rows/table/{table_id}/{record_id}/"
|
|
response = requests.patch(url, headers=self.headers, json=fields, params={"user_field_names": "true"})
|
|
try:
|
|
response.raise_for_status()
|
|
except requests.exceptions.HTTPError as e:
|
|
print(f"Baserow Update Error: {response.status_code} - {response.text}")
|
|
raise e
|
|
return response.json()
|
|
|
|
def upload_attachment(self, table_id: str, record_id: int, field_id: str, filepath: str | Path) -> Any:
|
|
# 1. Upload the file to Baserow
|
|
with open(filepath, "rb") as f:
|
|
response = requests.post(
|
|
f"{self.host}/api/user-files/upload-file/",
|
|
headers={"Authorization": f"Token {self.token}"},
|
|
files={'file': f}
|
|
)
|
|
response.raise_for_status()
|
|
file_name = response.json()['name']
|
|
|
|
# 2. Link the uploaded file to the record
|
|
url = f"{self.host}/api/database/rows/table/{table_id}/{record_id}/"
|
|
payload = {field_id: [{"name": file_name}]}
|
|
response = requests.patch(url, headers=self.headers, json=payload, params={"user_field_names": "true"})
|
|
response.raise_for_status()
|
|
return response.json()
|