Files
Spoolman/spoolman/api/v1/models.py
T
Donkie 21651216af Turn the locations page into a group-by-anything dashboard
The locations board could only ever answer one question: where is this
spool. The same card-and-drag interaction works for any field a spool
carries, so make the field a view mode and keep the rest.

Location is still the default view and behaves exactly as before. The
other views come from the user's own spool fields: any single-value text
or choice field now appears in a Group-by menu. A choice field renders a
card per choice, including the empty ones, which turns a "Status" field
into a kanban board.

Which field a card belongs to is decided by one small interface
(src/lib/dashboard/fields.ts), so the page never special-cases a field.
A field only qualifies if the SPOOL OWNS IT, because a drag moves one
spool and must mean exactly one change. That rules out material and
vendor, where writing one spool's card would silently rewrite every other
spool of the same filament, and lot number, which is stamped on by the
manufacturer and would be recording something untrue. Those stay library
filters. It also rules out numbers, dates, booleans and multi-choice,
whose values don't name a card you can drag into.

Backing changes:

- /spool/group grows group_by=extra.<key> for a spool's custom fields.
  The field table is aliased so it stays a distinct FROM element -- the
  extra-field filters reference the un-aliased table inside correlated
  subqueries, and without the alias those would correlate to this join
  instead of standing alone. Grouping is rejected for field types whose
  value isn't a single repeatable string. group_by is no longer a
  Literal, so an unsupported value is a 400 rather than a 422.

- The saved layout is per field: dashboard_groups and
  dashboard_spoolorders replace the location-only pair. The old
  `locations` / `locations_spoolorders` are read once as a seed for the
  location view and never written back, so an upgrade keeps the shelves
  it had and the old client keeps working.

- The view mode lives in the URL (?by=) so a board can be bookmarked and
  shared, and is mirrored to localStorage so a bare /dashboard resumes
  the last one looked at. The URL wins when it says something; a stale
  ?by= naming a deleted field falls back to location.

Verified against SQLite, PostgreSQL, MariaDB and CockroachDB -- Postgres
caught a real difference here, it sorts NULLs last where SQLite sorts
them first, so the new group assertions are keyed rather than ordered.
2026-07-30 08:41:36 +02:00

536 lines
19 KiB
Python

"""Pydantic data models for typing the FastAPI request/responses."""
from datetime import datetime, timezone
from enum import Enum
from typing import Annotated, Literal
from pydantic import BaseModel, Field, PlainSerializer
from spoolman.database import models
from spoolman.math import length_from_weight
from spoolman.settings import SettingDefinition, SettingType
def datetime_to_str(dt: datetime) -> str:
"""Convert a datetime object to a string."""
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.isoformat().replace("+00:00", "Z")
SpoolmanDateTime = Annotated[datetime, PlainSerializer(datetime_to_str)]
def _extra_fields_description(entity: str) -> str:
r"""Build the description for an entity's ``extra`` field.
Every value in the ``extra`` map is a JSON-encoded string, regardless of the field's
configured type. For example, an ``integer`` field returns ``"42"`` (not ``42``) and a
``text`` field returns ``"\"hello\""``. Consumers must JSON-decode each value to get the
typed value. This keeps the map uniformly typed as ``dict[str, str]`` on the wire.
"""
return (
f"Extra fields for this {entity}. Every value is a JSON-encoded string, regardless of the field's "
'configured type: e.g. an integer field returns "42" (not 42) and a text field returns "\\"hello\\"". '
"Consumers must JSON-decode each value. Query the /fields endpoint for the type of each field."
)
class Message(BaseModel):
message: str = Field()
class SettingResponse(BaseModel):
value: str = Field(description="Setting value.")
is_set: bool = Field(description="Whether the setting has been set. If false, 'value' contains the default value.")
type: SettingType = Field(description="Setting type. This corresponds with JSON types.")
class SettingKV(BaseModel):
key: str = Field(description="Setting key.")
setting: SettingResponse = Field(description="Setting value.")
@staticmethod
def from_db(definition: SettingDefinition, set_value: str | None) -> "SettingKV":
"""Create a new Pydantic vendor object from a database vendor object."""
return SettingKV(
key=definition.key,
setting=SettingResponse(
value=set_value if set_value is not None else definition.default,
is_set=set_value is not None,
type=definition.type,
),
)
class Vendor(BaseModel):
id: int = Field(description="Unique internal ID of this vendor.")
registered: SpoolmanDateTime = Field(description="When the vendor was registered in the database. UTC Timezone.")
name: str = Field(max_length=64, description="Vendor name.", examples=["Polymaker"])
comment: str | None = Field(
None,
max_length=1024,
description="Free text comment about this vendor.",
examples=[""],
)
empty_spool_weight: float | None = Field(
None,
ge=0,
description="The empty spool weight, in grams.",
examples=[140],
)
external_id: str | None = Field(
None,
max_length=256,
description=(
"Set if this vendor comes from an external database. This contains the ID in the external database."
),
examples=["eSun"],
)
extra: dict[str, str] = Field(
description=_extra_fields_description("vendor"),
)
@staticmethod
def from_db(item: models.Vendor) -> "Vendor":
"""Create a new Pydantic vendor object from a database vendor object."""
return Vendor(
id=item.id,
registered=item.registered,
name=item.name,
comment=item.comment,
empty_spool_weight=item.empty_spool_weight,
external_id=item.external_id,
extra={field.key: field.value for field in item.extra},
)
class MultiColorDirection(Enum):
"""Enum for multi-color direction."""
COAXIAL = "coaxial"
LONGITUDINAL = "longitudinal"
class Filament(BaseModel):
id: int = Field(description="Unique internal ID of this filament type.")
registered: SpoolmanDateTime = Field(description="When the filament was registered in the database. UTC Timezone.")
name: str | None = Field(
None,
max_length=64,
description=(
"Filament name, to distinguish this filament type among others from the same vendor."
"Should contain its color for example."
),
examples=["PolyTerra™ Charcoal Black"],
)
vendor: Vendor | None = Field(None, description="The vendor of this filament type.")
material: str | None = Field(
None,
max_length=64,
description="The material of this filament, e.g. PLA.",
examples=["PLA"],
)
price: float | None = Field(
None,
ge=0,
description="The price of this filament in the system configured currency.",
examples=[20.0],
)
density: float = Field(gt=0, description="The density of this filament in g/cm3.", examples=[1.24])
diameter: float = Field(gt=0, description="The diameter of this filament in mm.", examples=[1.75])
weight: float | None = Field(
None,
gt=0,
description="The weight of the filament in a full spool, in grams.",
examples=[1000],
)
spool_weight: float | None = Field(None, ge=0, description="The empty spool weight, in grams.", examples=[140])
article_number: str | None = Field(
None,
max_length=64,
description="Vendor article number, e.g. EAN, QR code, etc.",
examples=["PM70820"],
)
comment: str | None = Field(
None,
max_length=1024,
description="Free text comment about this filament type.",
examples=[""],
)
settings_extruder_temp: int | None = Field(
None,
ge=0,
description="Overridden extruder temperature, in °C.",
examples=[210],
)
settings_bed_temp: int | None = Field(
None,
ge=0,
description="Overridden bed temperature, in °C.",
examples=[60],
)
color_hex: str | None = Field(
None,
min_length=6,
max_length=8,
description=(
"Hexadecimal color code of the filament, e.g. FF0000 for red. Supports alpha channel at the end. "
"If it's a multi-color filament, the multi_color_hexes field is used instead."
),
examples=["FF0000"],
)
multi_color_hexes: str | None = Field(
None,
min_length=6,
description=(
"Hexadecimal color code of the filament, e.g. FF0000 for red. Supports alpha channel at the end. "
"Specifying multiple colors separated by commas. "
"Also set the multi_color_direction field if you specify multiple colors."
),
examples=["FF0000,00FF00,0000FF"],
)
multi_color_direction: MultiColorDirection | None = Field(
None,
description=("Type of multi-color filament. Only set if the multi_color_hexes field is set."),
examples=["coaxial", "longitudinal"],
)
external_id: str | None = Field(
None,
max_length=256,
description=(
"Set if this filament comes from an external database. This contains the ID in the external database."
),
examples=["polymaker_pla_polysonicblack_1000_175"],
)
extra: dict[str, str] = Field(
description=_extra_fields_description("filament"),
)
@staticmethod
def from_db(item: models.Filament) -> "Filament":
"""Create a new Pydantic filament object from a database filament object."""
return Filament(
id=item.id,
registered=item.registered,
name=item.name,
vendor=Vendor.from_db(item.vendor) if item.vendor is not None else None,
material=item.material,
price=item.price,
density=item.density,
diameter=item.diameter,
weight=item.weight,
spool_weight=item.spool_weight,
article_number=item.article_number,
comment=item.comment,
settings_extruder_temp=item.settings_extruder_temp,
settings_bed_temp=item.settings_bed_temp,
color_hex=item.color_hex,
multi_color_hexes=item.multi_color_hexes,
multi_color_direction=(
MultiColorDirection(item.multi_color_direction) if item.multi_color_direction is not None else None
),
external_id=item.external_id,
extra={field.key: field.value for field in item.extra},
)
class Spool(BaseModel):
id: int = Field(description="Unique internal ID of this spool of filament.")
registered: SpoolmanDateTime = Field(description="When the spool was registered in the database. UTC Timezone.")
first_used: SpoolmanDateTime | None = Field(
None,
description="First logged occurence of spool usage. UTC Timezone.",
)
last_used: SpoolmanDateTime | None = Field(
None,
description="Last logged occurence of spool usage. UTC Timezone.",
)
filament: Filament = Field(description="The filament type of this spool.")
price: float | None = Field(
None,
ge=0,
description="The price of this spool in the system configured currency.",
examples=[20.0],
)
remaining_weight: float | None = Field(
default=None,
ge=0,
description=(
"Estimated remaining weight of filament on the spool in grams. "
"Only set if the filament type has a weight set."
),
examples=[500.6],
)
initial_weight: float | None = Field(
default=None,
ge=0,
description=("The initial weight, in grams, of the filament on the spool (net weight)."),
examples=[1246],
)
spool_weight: float | None = Field(
default=None,
ge=0,
description=("Weight of an empty spool (tare weight)."),
examples=[246],
)
used_weight: float = Field(
ge=0,
description="Consumed weight of filament from the spool in grams.",
examples=[500.3],
)
remaining_length: float | None = Field(
default=None,
ge=0,
description=(
"Estimated remaining length of filament on the spool in millimeters."
" Only set if the filament type has a weight set."
),
examples=[5612.4],
)
used_length: float = Field(
ge=0,
description="Consumed length of filament from the spool in millimeters.",
examples=[50.7],
)
location: str | None = Field(
None,
max_length=64,
description="Where this spool can be found.",
examples=["Shelf A"],
)
lot_nr: str | None = Field(
None,
max_length=64,
description="Vendor manufacturing lot/batch number of the spool.",
examples=["52342"],
)
comment: str | None = Field(
None,
max_length=1024,
description="Free text comment about this specific spool.",
examples=[""],
)
archived: bool = Field(description="Whether this spool is archived and should not be used anymore.")
extra: dict[str, str] = Field(
description=_extra_fields_description("spool"),
)
@staticmethod
def from_db(item: models.Spool) -> "Spool":
"""Create a new Pydantic spool object from a database spool object."""
filament = Filament.from_db(item.filament)
remaining_weight: float | None = None
remaining_length: float | None = None
if item.initial_weight is not None:
remaining_weight = max(item.initial_weight - item.used_weight, 0)
remaining_length = length_from_weight(
weight=remaining_weight,
density=filament.density,
diameter=filament.diameter,
)
elif filament.weight is not None:
remaining_weight = max(filament.weight - item.used_weight, 0)
remaining_length = length_from_weight(
weight=remaining_weight,
density=filament.density,
diameter=filament.diameter,
)
used_length = length_from_weight(
weight=item.used_weight,
density=filament.density,
diameter=filament.diameter,
)
return Spool(
id=item.id,
registered=item.registered,
first_used=item.first_used,
last_used=item.last_used,
filament=filament,
price=item.price,
initial_weight=item.initial_weight,
spool_weight=item.spool_weight,
used_weight=item.used_weight,
used_length=used_length,
remaining_weight=remaining_weight,
remaining_length=remaining_length,
location=item.location,
lot_nr=item.lot_nr,
comment=item.comment,
archived=item.archived if item.archived is not None else False,
extra={field.key: field.value for field in item.extra},
)
class SpoolGroup(BaseModel):
"""A group of spools with server-computed aggregates.
Returned by the ``/spool/group`` endpoint. Spools are grouped by one axis
(``group_by``); the aggregates are computed over the matching spools of each
group so the client can paginate whole groups without fetching every spool.
"""
group_by: str = Field(
description="The field the spools are grouped by.",
examples=["filament"],
)
key: str | None = Field(
None,
description=(
"The group key. For group_by=filament/vendor this is the entity ID as a string; for "
"material/location and extra fields it is the value. Null when the grouped "
"field is unset (e.g. spools with no location or a filament with no vendor)."
),
examples=["12"],
)
spool_count: int = Field(description="Number of matching spools in this group.", examples=[6])
in_use_count: int = Field(
description="Number of matching spools that have been used (used_weight > 0).",
examples=[2],
)
total_remaining_weight: float | None = Field(
None,
description="Sum of remaining filament weight across the group's matching spools, in grams.",
examples=[3120.0],
)
last_used: SpoolmanDateTime | None = Field(
None,
description="Most recent last_used across the group's matching spools. UTC Timezone.",
)
filament: Filament | None = Field(
None,
description="The filament, embedded for group_by=filament so the header needs no extra request.",
)
vendor: Vendor | None = Field(
None,
description="The vendor, embedded for group_by=vendor.",
)
class SearchResultSpool(BaseModel):
"""A spool that matched a search, with which field matched."""
spool: Spool = Field(description="The matching spool.")
match_field: str = Field(
description=(
"Which field matched the query: a native field name (e.g. 'comment', 'location', "
"'lot_nr'), 'id' for an exact spool-id match, or 'extra.<key>' for an extra field."
),
examples=["comment"],
)
class SearchResultFilament(BaseModel):
"""A filament that matched a search, with which field matched."""
filament: Filament = Field(description="The matching filament.")
match_field: str = Field(
description=(
"Which field matched the query: a native field name (e.g. 'name', 'material', "
"'article_number', 'comment'), 'color' for a color-similarity match, or 'extra.<key>'."
),
examples=["color"],
)
class SearchResultVendor(BaseModel):
"""A vendor that matched a search, with which field matched."""
vendor: Vendor = Field(description="The matching vendor.")
match_field: str = Field(
description="Which field matched the query: 'name', 'comment', or 'extra.<key>'.",
examples=["name"],
)
class SearchResults(BaseModel):
"""Categorized results of a cross-entity search."""
spools: list[SearchResultSpool] = Field(description="Matching spools, best matches first.")
filaments: list[SearchResultFilament] = Field(description="Matching filaments, best matches first.")
vendors: list[SearchResultVendor] = Field(description="Matching vendors, best matches first.")
is_color_query: bool = Field(
description=(
"Whether the query was recognized as a color (hex code or CSS color name), in which case "
"the filament results include color-similarity matches and a threshold slider is relevant."
),
examples=[False],
)
class Info(BaseModel):
version: str = Field(examples=["0.7.0"])
debug_mode: bool = Field(examples=[False])
automatic_backups: bool = Field(examples=[True])
data_dir: str = Field(examples=["/home/app/.local/share/spoolman"])
logs_dir: str = Field(examples=["/home/app/.local/share/spoolman"])
backups_dir: str = Field(examples=["/home/app/.local/share/spoolman/backups"])
db_type: str = Field(examples=["sqlite"])
external_db_name: str = Field(
description="Display name for the external filament library, configurable via EXTERNAL_DB_NAME.",
examples=["SpoolmanDB"],
)
git_commit: str | None = Field(None, examples=["a1b2c3d"])
build_date: SpoolmanDateTime | None = Field(None, examples=["2021-01-01T00:00:00Z"])
class HealthCheck(BaseModel):
status: str = Field(examples=["healthy"])
class BackupResponse(BaseModel):
path: str = Field(
default=None,
description="Path to the created backup file.",
examples=["/home/app/.local/share/spoolman/backups/spoolman.db"],
)
class EventType(str, Enum):
"""Event types."""
ADDED = "added"
UPDATED = "updated"
DELETED = "deleted"
class Event(BaseModel):
"""Event."""
type: EventType = Field(description="Event type.")
resource: str = Field(description="Resource type.")
date: SpoolmanDateTime = Field(description="When the event occured. UTC Timezone.")
payload: BaseModel
class SpoolEvent(Event):
"""Event."""
payload: Spool = Field(description="Updated spool.")
resource: Literal["spool"] = Field(description="Resource type.")
class FilamentEvent(Event):
"""Event."""
payload: Filament = Field(description="Updated filament.")
resource: Literal["filament"] = Field(description="Resource type.")
class VendorEvent(Event):
"""Event."""
payload: Vendor = Field(description="Updated vendor.")
resource: Literal["vendor"] = Field(description="Resource type.")
class SettingEvent(Event):
"""Event."""
payload: SettingKV = Field(description="Updated setting.")
resource: Literal["setting"] = Field(description="Resource type.")