mirror of
https://github.com/Donkie/Spoolman.git
synced 2026-10-22 10:02:17 -05:00
Donkie
6520ce6d1b
A tag can now be linked to a filament type, not just to one spool of it. The
tag table already carried filament_id and target_type for exactly this, so
there is no migration.
API, all additive:
POST /filament/{id}/tag link; 409 if anything else holds the UID
DELETE /filament/{id}/tag/{uid} unlink
GET /filament?tag=<uid> exact lookup on the normalized UID
Filament.tags also inside every spool's embedded filament
TagScan.matched_filament_id always present, like matched_spool_id
TagConflictMessage.filament_id names a filament holder on a 409
One UID still identifies one thing across both kinds: a spool's tag cannot be
linked to a filament or the reverse, and neither ?tag= lookup answers with the
other kind. The scan debounce key now includes the kind, since spool 7 and
filament 7 are different answers.
SpoolTag and SpoolTagParameters are renamed Tag and TagLinkParameters while
they are unreleased, so the schema does not describe filament tags as spool
tags forever.
client_v2: the filament inspector gets the Tags section, the link dialog names
and moves a tag held by either kind, and scanning a filament's tag opens that
filament when auto-navigate is on.
Tag having two delete-orphan parents was checked on SQLAlchemy 2.0.51: new tags
flush, unlinking deletes the row, and deleting either parent cascades.
Proven on sqlite, postgres, mariadb and cockroachdb: 558 integration tests
each. Unit tests, svelte-check and the Playwright tag specs pass.
99 lines
4.1 KiB
Python
99 lines
4.1 KiB
Python
"""Functionality for exporting data in various format."""
|
|
|
|
import asyncio
|
|
import csv
|
|
import json
|
|
from collections.abc import Iterable
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from spoolman.database import models
|
|
|
|
if TYPE_CHECKING:
|
|
from _typeshed import SupportsWrite
|
|
|
|
banned_attrs = {"awaitable_attrs", "metadata", "registry", "spools", "filaments"}
|
|
|
|
# A spreadsheet treats a cell starting with any of these as a formula, so a vendor named
|
|
# `=cmd|' /C calc'!A0` executes when the export is opened (CWE-1236). Prefixing with a single
|
|
# quote makes the spreadsheet read it as literal text; the quote is not part of the value.
|
|
FORMULA_PREFIXES = ("=", "+", "-", "@", "\t", "\r")
|
|
|
|
|
|
def escape_csv_value(value: Any) -> Any: # noqa: ANN401
|
|
"""Neutralize a value that a spreadsheet would otherwise interpret as a formula.
|
|
|
|
Only strings are escaped. Numbers reach the writer as numeric types, so a negative number
|
|
cannot be mistaken for a formula and must not grow a stray quote.
|
|
|
|
Args:
|
|
value: The cell value.
|
|
|
|
Returns:
|
|
Any: The value, prefixed with a single quote if it would be read as a formula.
|
|
|
|
"""
|
|
if isinstance(value, str) and value.startswith(FORMULA_PREFIXES):
|
|
return "'" + value
|
|
return value
|
|
|
|
|
|
async def flatten_sqlalchemy_object(obj: models.Base, parent_key: str = "", sep: str = ".") -> dict[str, Any]:
|
|
"""Recursively flattens a SQLAlchemy object into a dictionary with dot-separated keys."""
|
|
fields = {}
|
|
for attr in dir(obj):
|
|
# Check if the attribute is a column or a relationship
|
|
if not attr.startswith("_") and attr not in banned_attrs:
|
|
value = await getattr(obj.awaitable_attrs, attr)
|
|
|
|
if attr == "extra":
|
|
# Handle extra fields
|
|
for v in value:
|
|
fields[f"{parent_key}extra.{v.key}"] = v.value
|
|
continue
|
|
|
|
if attr == "tags":
|
|
# A collection, like `extra`, so it needs collapsing into a cell rather
|
|
# than being left as a list of ORM objects: CSV would write their repr and
|
|
# JSON would stringify them. Tags, a spool's or a filament's, have no keys to
|
|
# spread across columns, so they become one comma-separated list of UIDs.
|
|
fields[f"{parent_key}tags"] = ",".join(v.uid for v in value)
|
|
continue
|
|
|
|
# Handle nested SQLAlchemy objects
|
|
if isinstance(value, models.Base):
|
|
nested_fields = await flatten_sqlalchemy_object(value, f"{parent_key}{attr}{sep}", sep=sep)
|
|
fields.update(nested_fields)
|
|
else:
|
|
# Use only columns and simple data types
|
|
fields[f"{parent_key}{attr}"] = value
|
|
return fields
|
|
|
|
|
|
async def dump_as_csv(sqlalchemy_objects: Iterable[models.Base], writer: "SupportsWrite[str]") -> None:
|
|
"""Export a list of objects as CSV to a writer. Nested objects are flattened with dot-separated keys."""
|
|
# Flatten each object and get all column names
|
|
all_flattened = await asyncio.gather(*[flatten_sqlalchemy_object(obj) for obj in sqlalchemy_objects])
|
|
|
|
# Collect all unique headers across flattened objects
|
|
headers = set()
|
|
for flattened_obj in all_flattened:
|
|
headers.update(flattened_obj.keys())
|
|
|
|
headers = sorted(headers) # Sort headers for consistent column ordering
|
|
|
|
# Write to CSV. Header names need no escaping: they are either fixed attribute names or
|
|
# extra-field keys, which are constrained to ^[a-z0-9_]+$.
|
|
csv_writer = csv.DictWriter(writer, fieldnames=headers)
|
|
csv_writer.writeheader()
|
|
for flattened_obj in all_flattened:
|
|
csv_writer.writerow({key: escape_csv_value(value) for key, value in flattened_obj.items()})
|
|
|
|
|
|
async def dump_as_json(sqlalchemy_objects: Iterable[models.Base], writer: "SupportsWrite[str]") -> None:
|
|
"""Export a list of objects as JSON to a writer. Nested objects are flattened with dot-separated keys."""
|
|
# Flatten each object and get all column names
|
|
all_flattened = await asyncio.gather(*[flatten_sqlalchemy_object(obj) for obj in sqlalchemy_objects])
|
|
|
|
# Write to JSON
|
|
json.dump(all_flattened, writer, default=str)
|