nicolas/nodeslicer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

upsert

Browse source
This commit is contained in:
Nicolas Predella 2026-05-18 15:16:22 +03:00
parent 0c33dae183
commit 02f40c55dc
2 changed files with 23 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
core/nodeslicer/nodes.py
View file

@ -10,7 +10,7 @@ from typing import Any, TypeAlias
import bpy
import numpy as np
from mathutils import Euler
from .bpy_ns_types import NfStringProperty, NfEnumProperty, NfBoolProperty
from .bpy_ns_types import NfIntProperty, NfStringProperty, NfEnumProperty, NfBoolProperty
from ...registry import register_class
from .. import slice
@ -1160,7 +1160,7 @@ class SLICENODE_BaserowSubmit(ExecSliceNode):
host: NfStringProperty(name="Host", default="https://baserow.io") # pyright: ignore[reportInvalidTypeForm]
token: NfStringProperty(name="Token", default="") # pyright: ignore[reportInvalidTypeForm]
table_id: NfStringProperty(name="Table ID", default="") # pyright: ignore[reportInvalidTypeForm]
gcode_field: NfStringProperty(name="Gcode Field", default="GCODE") # pyright: ignore[reportInvalidTypeForm]
record_id: NfStringProperty(name="Record ID", default="") # pyright: ignore[reportInvalidTypeForm]
def init(self, context):
self._init_exec()
@ -1172,7 +1172,7 @@ class SLICENODE_BaserowSubmit(ExecSliceNode):
layout.prop(self, "host")
layout.prop(self, "token")
layout.prop(self, "table_id")
layout.prop(self, "gcode_field")
layout.prop(self, "record_id")
def evaluate(self, operator=None):
from ...infra.networking.baserow_backend import BaserowBackend
@ -1184,7 +1184,7 @@ class SLICENODE_BaserowSubmit(ExecSliceNode):
if operator: operator.report({'ERROR'}, "No Gcode to send")
return 'no gcode'
if not self.token or not self.table_id:
if not self.token or not self.table_id or not self.host:
if operator: operator.report({'ERROR'}, "Baserow configuration incomplete")
return 'incomplete config'
@ -1195,10 +1195,14 @@ class SLICENODE_BaserowSubmit(ExecSliceNode):
temp_path = Path(tf.name)
try:
res = db.create_record(self.table_id, {})
record_id = int(res['id'])
db.upload_attachment(self.table_id, record_id, self.gcode_field, temp_path)
if not self.record_id:
res = db.create_record(self.table_id, {})
record_id = int(res['id'])
else: record_id = self.record_id
db.upload_attachment(self.table_id, record_id, 'GCODE', temp_path)
return 'ok'
except Exception as e:
if operator: operator.report({'ERROR'}, f"Baserow Error: {str(e)}")
return 'error'

46
infra/networking/baserow_backend.py
View file

@ -11,42 +11,18 @@ class BaserowBackend:
"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 = {}
def get_record(self, table_id: str, record_id: int) -> dict:
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()
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 upsert_record(self, table_id: str, record_id: int, fields: dict[str, Any]) -> dict:
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:
url = f"{self.host}/api/database/rows/table/{table_id}/"