87 lines
3.6 KiB
Python
87 lines
3.6 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 fetch(self, table_id: str, filter_formula: str | None = None) -> list[dict]:
|
|
url = f"{self.host}/api/database/rows/table/{table_id}/"
|
|
params = {"user_field_names": "true"}
|
|
rows = []
|
|
while url:
|
|
response = requests.get(url, headers=self.headers, params=params)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
rows.extend(data.get("results", []))
|
|
url = data.get("next")
|
|
params = {}
|
|
|
|
if filter_formula:
|
|
# Simple client-side filtering to mimic Airtable formulas
|
|
import re
|
|
match = re.search(r"^({)?(.*?)(})?\s*=\s*['\"]?(.*?)['\"]?$", filter_formula)
|
|
if match:
|
|
_, f, _, v = match.groups()
|
|
|
|
filtered_rows = []
|
|
for r in rows:
|
|
actual_val = r['id'] if f in ('RECORD_ID()', 'ID') else r.get(f)
|
|
|
|
# Direct comparison
|
|
if str(actual_val) == str(v):
|
|
filtered_rows.append(r)
|
|
continue
|
|
|
|
# Numeric comparison fallback
|
|
try:
|
|
if float(actual_val) == float(v):
|
|
filtered_rows.append(r)
|
|
except (ValueError, TypeError):
|
|
pass
|
|
rows = filtered_rows
|
|
return rows
|
|
|
|
def create_record(self, table_id: str, fields: dict[str, Any]) -> dict:
|
|
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:
|
|
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()
|