mirror of
https://github.com/Donkie/Spoolman.git
synced 2026-10-23 02:22:20 -05:00
Donkie
4260a40bc7
Applies a permission level to all 49 routes and turns on the coverage check,
which from here refuses to import the API if a route is added without one.
Levels follow the design doc, with two deliberate departures:
- GET /export/{spools,filaments,vendors} require manage, not read. The doc's
rule is "read covers every GET", but these three return the entire
database in one request, and the doc's own opening section names exactly
that as a reason this project exists. Left at read they would also be
reachable anonymously once auth_anonymous_read is on.
- GET /setting/ stays at read. The client reads currency, price rounding and
base URL at boot, so raising it would break read-only browsing for the
users the level exists for. Keeping secrets out of the settings table is
handled where it belongs, at the storage rule in settings.py.
PUT /spool/{id}/use sits at edit rather than manage on purpose: it is
Moonraker's and OctoPrint's hot path, so a printer can hold a credential
that records filament usage but cannot delete anything.
Websocket routes take the ws_authenticated decorator instead of a
dependency, and their own accept() calls are removed since the decorator now
owns the handshake. One side effect worth noting: the pre-existing
close(4040) for an unknown setting key in setting.py now runs after the
accept, so that code finally reaches clients instead of being flattened into
an HTTP 403. That is an accidental fix, not the goal.
Also fixes a leak found while testing this: resolve_principal drove
get_db_session with "async for" and then returned out of the loop, leaving
the generator suspended at its yield so the session was closed later by the
garbage collector, off the event loop, where SQLAlchemy cannot await. It
produced a wall of greenlet errors on every authenticated request. Sessions
are now closed deterministically through a context manager.
Verified:
- The full existing integration suite passes against a running server with
auth off: 372 tests, no test file edited. That is the proof that this is
additive and that the disabled path is unchanged.
- Removing a single gate makes the API refuse to import and names the route.
- On a live server with auth on, /health and the SPA stay reachable, every
API route answers 401 with the {"message": ...} shape the rest of the API
uses, and secret.key is created at mode 0600.
- 48 gated routes: 29 read, 6 edit, 13 manage, plus /health public.
778 lines
27 KiB
Python
778 lines
27 KiB
Python
"""Spool related endpoints."""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Annotated, Literal
|
|
|
|
from fastapi import APIRouter, Depends, Query, Request, WebSocket, WebSocketDisconnect
|
|
from fastapi.encoders import jsonable_encoder
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, Field, field_validator
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from starlette.datastructures import QueryParams
|
|
|
|
from spoolman.api.v1.models import Filament, Message, Spool, SpoolEvent, SpoolGroup, Vendor
|
|
from spoolman.auth.dependencies import require_level, ws_authenticated
|
|
from spoolman.auth.levels import Level
|
|
from spoolman.database import spool
|
|
from spoolman.database.database import get_db_session
|
|
from spoolman.database.utils import SortOrder
|
|
from spoolman.exceptions import ItemCreateError, SpoolMeasureError
|
|
from spoolman.extra_fields import EntityType, get_extra_fields, validate_extra_field_dict
|
|
from spoolman.ws import websocket_manager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(
|
|
prefix="/spool",
|
|
tags=["spool"],
|
|
)
|
|
|
|
# ruff: noqa: D103
|
|
|
|
|
|
# Query-param prefixes for extra-field filters, longest first so the most specific one wins.
|
|
_EXTRA_FILTER_PREFIXES = (
|
|
("filament.vendor.extra.", "vendor"),
|
|
("filament.extra.", "filament"),
|
|
("extra.", "spool"),
|
|
)
|
|
|
|
|
|
def _parse_extra_field_filters(
|
|
query_params: QueryParams,
|
|
) -> tuple[dict[str, str], dict[str, str], dict[str, str]]:
|
|
"""Split extra-field filter query params into (spool, filament, vendor) dicts keyed by field key."""
|
|
buckets: dict[str, dict[str, str]] = {"spool": {}, "filament": {}, "vendor": {}}
|
|
for key, value in query_params.items():
|
|
for prefix, entity in _EXTRA_FILTER_PREFIXES:
|
|
if key.startswith(prefix):
|
|
buckets[entity][key[len(prefix) :]] = value
|
|
break
|
|
return buckets["spool"], buckets["filament"], buckets["vendor"]
|
|
|
|
|
|
class SpoolParameters(BaseModel):
|
|
first_used: datetime | None = Field(None, description="First logged occurence of spool usage.")
|
|
last_used: datetime | None = Field(None, description="Last logged occurence of spool usage.")
|
|
filament_id: int = Field(description="The ID of the filament type of this spool.")
|
|
price: float | None = Field(
|
|
None,
|
|
ge=0,
|
|
description="The price of this filament in the system configured currency.",
|
|
examples=[20.0],
|
|
)
|
|
initial_weight: float | None = Field(
|
|
None,
|
|
ge=0,
|
|
description="The initial weight of the filament on the spool, in grams. (net weight)",
|
|
examples=[200],
|
|
)
|
|
spool_weight: float | None = Field(
|
|
None,
|
|
ge=0,
|
|
description="The weight of an empty spool, in grams. (tare weight)",
|
|
examples=[200],
|
|
)
|
|
remaining_weight: float | None = Field(
|
|
None,
|
|
ge=0,
|
|
description=(
|
|
"Remaining weight of filament on the spool. Can only be used if the filament type has a weight set."
|
|
),
|
|
examples=[800],
|
|
)
|
|
used_weight: float | None = Field(
|
|
None,
|
|
ge=0,
|
|
description="Used weight of filament on the spool.",
|
|
examples=[200],
|
|
)
|
|
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(default=False, description="Whether this spool is archived and should not be used anymore.")
|
|
extra: dict[str, str] | None = Field(
|
|
None,
|
|
description="Extra fields for this spool.",
|
|
)
|
|
|
|
|
|
class SpoolUpdateParameters(SpoolParameters):
|
|
filament_id: int | None = Field(None, description="The ID of the filament type of this spool.")
|
|
|
|
@field_validator("filament_id")
|
|
@classmethod
|
|
def prevent_none(cls: type[SpoolUpdateParameters], v: int | None) -> int | None:
|
|
"""Prevent filament_id from being None."""
|
|
if v is None:
|
|
raise ValueError("Value must not be None.")
|
|
return v
|
|
|
|
|
|
class SpoolUseParameters(BaseModel):
|
|
use_length: float | None = Field(None, description="Length of filament to reduce by, in mm.", examples=[2.2])
|
|
use_weight: float | None = Field(None, description="Filament weight to reduce by, in g.", examples=[5.3])
|
|
|
|
|
|
class SpoolMeasureParameters(BaseModel):
|
|
weight: float = Field(description="Current gross weight of the spool, in g.", examples=[200])
|
|
|
|
|
|
@router.get(
|
|
"",
|
|
name="Find spool",
|
|
description=(
|
|
"Get a list of spools that matches the search query. "
|
|
"A websocket is served on the same path to listen for updates to any spool, or added or deleted spools. "
|
|
"See the HTTP Response code 299 for the content of the websocket messages."
|
|
),
|
|
response_model_exclude_none=True,
|
|
responses={
|
|
200: {"model": list[Spool]},
|
|
299: {"model": SpoolEvent, "description": "Websocket message"},
|
|
},
|
|
dependencies=[Depends(require_level(Level.READ))],
|
|
)
|
|
async def find(
|
|
*,
|
|
request: Request,
|
|
db: Annotated[AsyncSession, Depends(get_db_session)],
|
|
filament_name_old: Annotated[
|
|
str | None,
|
|
Query(alias="filament_name", title="Filament Name", description="See filament.name.", deprecated=True),
|
|
] = None,
|
|
filament_id_old: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament_id",
|
|
title="Filament ID",
|
|
description="See filament.id.",
|
|
deprecated=True,
|
|
pattern=r"^-?\d+(,-?\d+)*$",
|
|
),
|
|
] = None,
|
|
filament_material_old: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament_material",
|
|
title="Filament Material",
|
|
description="See filament.material.",
|
|
deprecated=True,
|
|
),
|
|
] = None,
|
|
vendor_name_old: Annotated[
|
|
str | None,
|
|
Query(alias="vendor_name", title="Vendor Name", description="See filament.vendor.name.", deprecated=True),
|
|
] = None,
|
|
vendor_id_old: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="vendor_id",
|
|
title="Vendor ID",
|
|
description="See filament.vendor.id.",
|
|
deprecated=True,
|
|
pattern=r"^-?\d+(,-?\d+)*$",
|
|
),
|
|
] = None,
|
|
filament_name: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament.name",
|
|
title="Filament Name",
|
|
description=(
|
|
"Partial case-insensitive search term for the filament name. Separate multiple terms with a comma. "
|
|
"Specify an empty string to match spools with no filament name. "
|
|
"Surround a term with quotes to search for the exact term."
|
|
),
|
|
),
|
|
] = None,
|
|
filament_id: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament.id",
|
|
title="Filament ID",
|
|
description="Match an exact filament ID. Separate multiple IDs with a comma.",
|
|
examples=["1", "1,2"],
|
|
pattern=r"^-?\d+(,-?\d+)*$",
|
|
),
|
|
] = None,
|
|
filament_material: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament.material",
|
|
title="Filament Material",
|
|
description=(
|
|
"Partial case-insensitive search term for the filament material. Separate multiple terms with a comma. "
|
|
"Specify an empty string to match spools with no filament material. "
|
|
"Surround a term with quotes to search for the exact term."
|
|
),
|
|
),
|
|
] = None,
|
|
filament_multi_color_direction: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament.multi_color_direction",
|
|
title="Filament Multi-Color Direction",
|
|
description=(
|
|
"Match spools by their filament's multi-color direction, e.g. coaxial or longitudinal. "
|
|
"Separate multiple terms with a comma. Specify an empty string to match single-color filaments. "
|
|
"Surround a term with quotes to search for the exact term."
|
|
),
|
|
examples=['"coaxial"', '"longitudinal"'],
|
|
),
|
|
] = None,
|
|
filament_vendor_name: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament.vendor.name",
|
|
title="Vendor Name",
|
|
description=(
|
|
"Partial case-insensitive search term for the filament vendor name. "
|
|
"Separate multiple terms with a comma. "
|
|
"Specify an empty string to match spools with no vendor name. "
|
|
"Surround a term with quotes to search for the exact term."
|
|
),
|
|
),
|
|
] = None,
|
|
filament_vendor_id: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament.vendor.id",
|
|
title="Vendor ID",
|
|
description=(
|
|
"Match an exact vendor ID. Separate multiple IDs with a comma. "
|
|
"Set it to -1 to match spools with filaments with no vendor."
|
|
),
|
|
examples=["1", "1,2"],
|
|
pattern=r"^-?\d+(,-?\d+)*$",
|
|
),
|
|
] = None,
|
|
location: Annotated[
|
|
str | None,
|
|
Query(
|
|
title="Location",
|
|
description=(
|
|
"Partial case-insensitive search term for the spool location. Separate multiple terms with a comma. "
|
|
"Specify an empty string to match spools with no location. "
|
|
"Surround a term with quotes to search for the exact term."
|
|
),
|
|
),
|
|
] = None,
|
|
lot_nr: Annotated[
|
|
str | None,
|
|
Query(
|
|
title="Lot/Batch Number",
|
|
description=(
|
|
"Partial case-insensitive search term for the spool lot number. Separate multiple terms with a comma. "
|
|
"Specify an empty string to match spools with no lot nr. "
|
|
"Surround a term with quotes to search for the exact term."
|
|
),
|
|
),
|
|
] = None,
|
|
allow_archived: Annotated[
|
|
bool,
|
|
Query(title="Allow Archived", description="Whether to include archived spools in the search results."),
|
|
] = False,
|
|
sort: Annotated[
|
|
str | None,
|
|
Query(
|
|
title="Sort",
|
|
description=(
|
|
'Sort the results by the given field. Should be a comma-separate string with "field:direction" items.'
|
|
),
|
|
examples=["filament.name:asc,filament.vendor.id:asc,location:desc"],
|
|
),
|
|
] = None,
|
|
limit: Annotated[
|
|
int | None,
|
|
Query(title="Limit", description="Maximum number of items in the response."),
|
|
] = None,
|
|
offset: Annotated[int, Query(title="Offset", description="Offset in the full result set if a limit is set.")] = 0,
|
|
) -> JSONResponse:
|
|
sort_by: dict[str, SortOrder] = {}
|
|
if sort is not None:
|
|
for sort_item in sort.split(","):
|
|
field, direction = sort_item.split(":")
|
|
sort_by[field] = SortOrder[direction.upper()]
|
|
|
|
filament_id = filament_id if filament_id is not None else filament_id_old
|
|
if filament_id is not None:
|
|
filament_ids = [int(filament_id_item) for filament_id_item in filament_id.split(",")]
|
|
else:
|
|
filament_ids = None
|
|
|
|
filament_vendor_id = filament_vendor_id if filament_vendor_id is not None else vendor_id_old
|
|
if filament_vendor_id is not None:
|
|
filament_vendor_ids = [int(vendor_id_item) for vendor_id_item in filament_vendor_id.split(",")]
|
|
else:
|
|
filament_vendor_ids = None
|
|
|
|
# Extract custom field filters from query parameters. Spool extra fields use `extra.<key>`;
|
|
# a filament's extra fields use `filament.extra.<key>` and its vendor's `filament.vendor.extra.<key>`.
|
|
spool_extra, filament_extra, vendor_extra = _parse_extra_field_filters(request.query_params)
|
|
|
|
try:
|
|
db_items, total_count = await spool.find(
|
|
db=db,
|
|
filament_name=filament_name if filament_name is not None else filament_name_old,
|
|
filament_id=filament_ids,
|
|
filament_material=filament_material if filament_material is not None else filament_material_old,
|
|
filament_multi_color_direction=filament_multi_color_direction,
|
|
vendor_name=filament_vendor_name if filament_vendor_name is not None else vendor_name_old,
|
|
vendor_id=filament_vendor_ids,
|
|
location=location,
|
|
lot_nr=lot_nr,
|
|
allow_archived=allow_archived,
|
|
extra_field_filters=spool_extra or None,
|
|
filament_extra_field_filters=filament_extra or None,
|
|
vendor_extra_field_filters=vendor_extra or None,
|
|
sort_by=sort_by,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
except ValueError as e:
|
|
return JSONResponse(status_code=400, content=Message(message=str(e)).dict())
|
|
|
|
# Set x-total-count header for pagination
|
|
return JSONResponse(
|
|
content=jsonable_encoder(
|
|
(Spool.from_db(db_item) for db_item in db_items),
|
|
exclude_none=True,
|
|
),
|
|
headers={"x-total-count": str(total_count)},
|
|
)
|
|
|
|
|
|
@router.websocket(
|
|
"",
|
|
name="Listen to spool changes",
|
|
)
|
|
@ws_authenticated(Level.READ)
|
|
async def notify_any(
|
|
websocket: WebSocket,
|
|
) -> None:
|
|
websocket_manager.connect(("spool",), websocket)
|
|
try:
|
|
while True:
|
|
await asyncio.sleep(0.5)
|
|
if await websocket.receive_text():
|
|
await websocket.send_json({"status": "healthy"})
|
|
except WebSocketDisconnect:
|
|
websocket_manager.disconnect(("spool",), websocket)
|
|
|
|
|
|
@router.get(
|
|
"/group",
|
|
name="Find spool groups",
|
|
description=(
|
|
"Group spools that match the search query by one axis (filament, vendor, material or "
|
|
"location) and return per-group aggregates: spool count, in-use count, total remaining "
|
|
"weight and most recent usage. Pagination is over groups, so a group is never split and "
|
|
"its aggregates are always complete. Uses the same filters as the spool search endpoint. "
|
|
"The total number of matching groups is returned in the x-total-count header."
|
|
),
|
|
response_model_exclude_none=True,
|
|
responses={
|
|
200: {"model": list[SpoolGroup]},
|
|
400: {"model": Message},
|
|
},
|
|
dependencies=[Depends(require_level(Level.READ))],
|
|
)
|
|
async def find_groups(
|
|
*,
|
|
request: Request,
|
|
db: Annotated[AsyncSession, Depends(get_db_session)],
|
|
group_by: Annotated[
|
|
Literal["filament", "vendor", "material", "location"],
|
|
Query(title="Group By", description="The field to group spools by."),
|
|
],
|
|
filament_name: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament.name",
|
|
title="Filament Name",
|
|
description="Partial case-insensitive search term for the filament name. See the spool search endpoint.",
|
|
),
|
|
] = None,
|
|
filament_id: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament.id",
|
|
title="Filament ID",
|
|
description="Match an exact filament ID. Separate multiple IDs with a comma.",
|
|
pattern=r"^-?\d+(,-?\d+)*$",
|
|
),
|
|
] = None,
|
|
filament_material: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament.material",
|
|
title="Filament Material",
|
|
description="Partial case-insensitive search term for the filament material.",
|
|
),
|
|
] = None,
|
|
filament_multi_color_direction: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament.multi_color_direction",
|
|
title="Filament Multi-Color Direction",
|
|
description=(
|
|
"Match by the filament's multi-color direction, e.g. coaxial or longitudinal. "
|
|
"Specify an empty string to match single-color filaments."
|
|
),
|
|
examples=['"coaxial"', '"longitudinal"'],
|
|
),
|
|
] = None,
|
|
filament_vendor_name: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament.vendor.name",
|
|
title="Vendor Name",
|
|
description="Partial case-insensitive search term for the filament vendor name.",
|
|
),
|
|
] = None,
|
|
filament_vendor_id: Annotated[
|
|
str | None,
|
|
Query(
|
|
alias="filament.vendor.id",
|
|
title="Vendor ID",
|
|
description=(
|
|
"Match an exact vendor ID. Separate multiple IDs with a comma. "
|
|
"Set it to -1 to match spools with filaments with no vendor."
|
|
),
|
|
pattern=r"^-?\d+(,-?\d+)*$",
|
|
),
|
|
] = None,
|
|
location: Annotated[
|
|
str | None,
|
|
Query(title="Location", description="Partial case-insensitive search term for the spool location."),
|
|
] = None,
|
|
lot_nr: Annotated[
|
|
str | None,
|
|
Query(title="Lot/Batch Number", description="Partial case-insensitive search term for the spool lot number."),
|
|
] = None,
|
|
allow_archived: Annotated[
|
|
bool,
|
|
Query(title="Allow Archived", description="Whether to include archived spools in the aggregates."),
|
|
] = False,
|
|
sort: Annotated[
|
|
str | None,
|
|
Query(
|
|
title="Sort",
|
|
description=(
|
|
'Sort the groups by the given field. Comma-separated "field:direction" items. '
|
|
"Available fields: group.title, group.total_remaining, group.last_used, "
|
|
"group.spool_count, group.in_use_count."
|
|
),
|
|
examples=["group.last_used:desc"],
|
|
),
|
|
] = None,
|
|
limit: Annotated[
|
|
int | None,
|
|
Query(title="Limit", description="Maximum number of groups in the response."),
|
|
] = None,
|
|
offset: Annotated[
|
|
int,
|
|
Query(title="Offset", description="Offset in the full group result set if a limit is set."),
|
|
] = 0,
|
|
) -> JSONResponse:
|
|
sort_by: dict[str, SortOrder] = {}
|
|
if sort is not None:
|
|
for sort_item in sort.split(","):
|
|
field, direction = sort_item.split(":")
|
|
sort_by[field] = SortOrder[direction.upper()]
|
|
|
|
filament_ids = [int(item) for item in filament_id.split(",")] if filament_id is not None else None
|
|
vendor_ids = [int(item) for item in filament_vendor_id.split(",")] if filament_vendor_id is not None else None
|
|
|
|
spool_extra, filament_extra, vendor_extra = _parse_extra_field_filters(request.query_params)
|
|
|
|
try:
|
|
groups, total_count = await spool.find_groups(
|
|
db=db,
|
|
group_by=group_by,
|
|
filament_name=filament_name,
|
|
filament_id=filament_ids,
|
|
filament_material=filament_material,
|
|
filament_multi_color_direction=filament_multi_color_direction,
|
|
vendor_name=filament_vendor_name,
|
|
vendor_id=vendor_ids,
|
|
location=location,
|
|
lot_nr=lot_nr,
|
|
allow_archived=allow_archived,
|
|
extra_field_filters=spool_extra or None,
|
|
filament_extra_field_filters=filament_extra or None,
|
|
vendor_extra_field_filters=vendor_extra or None,
|
|
sort_by=sort_by,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
except ValueError as e:
|
|
return JSONResponse(status_code=400, content=Message(message=str(e)).dict())
|
|
|
|
content = [
|
|
SpoolGroup(
|
|
group_by=group_by,
|
|
key=None if group.key is None else str(group.key),
|
|
spool_count=group.spool_count,
|
|
in_use_count=group.in_use_count,
|
|
total_remaining_weight=group.total_remaining_weight,
|
|
last_used=group.last_used,
|
|
filament=Filament.from_db(group.filament) if group.filament is not None else None,
|
|
vendor=Vendor.from_db(group.vendor) if group.vendor is not None else None,
|
|
)
|
|
for group in groups
|
|
]
|
|
return JSONResponse(
|
|
content=jsonable_encoder(content, exclude_none=True),
|
|
headers={"x-total-count": str(total_count)},
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/{spool_id}",
|
|
name="Get spool",
|
|
description=(
|
|
"Get a specific spool. A websocket is served on the same path to listen for changes to the spool. "
|
|
"See the HTTP Response code 299 for the content of the websocket messages."
|
|
),
|
|
response_model_exclude_none=True,
|
|
responses={404: {"model": Message}, 299: {"model": SpoolEvent, "description": "Websocket message"}},
|
|
dependencies=[Depends(require_level(Level.READ))],
|
|
)
|
|
async def get(
|
|
db: Annotated[AsyncSession, Depends(get_db_session)],
|
|
spool_id: int,
|
|
) -> Spool:
|
|
db_item = await spool.get_by_id(db, spool_id)
|
|
return Spool.from_db(db_item)
|
|
|
|
|
|
@router.websocket(
|
|
"/{spool_id}",
|
|
name="Listen to spool changes",
|
|
)
|
|
@ws_authenticated(Level.READ)
|
|
async def notify(
|
|
websocket: WebSocket,
|
|
spool_id: int,
|
|
) -> None:
|
|
websocket_manager.connect(("spool", str(spool_id)), websocket)
|
|
try:
|
|
while True:
|
|
await asyncio.sleep(0.5)
|
|
if await websocket.receive_text():
|
|
await websocket.send_json({"status": "healthy"})
|
|
except WebSocketDisconnect:
|
|
websocket_manager.disconnect(("spool", str(spool_id)), websocket)
|
|
|
|
|
|
@router.post(
|
|
"",
|
|
name="Add spool",
|
|
description=(
|
|
"Add a new spool to the database. "
|
|
"Only specify either remaining_weight or used_weight. "
|
|
"If no weight is set, the spool will be assumed to be full."
|
|
),
|
|
response_model_exclude_none=True,
|
|
response_model=Spool,
|
|
responses={
|
|
400: {"model": Message},
|
|
},
|
|
dependencies=[Depends(require_level(Level.MANAGE))],
|
|
)
|
|
async def create( # noqa: ANN201
|
|
db: Annotated[AsyncSession, Depends(get_db_session)],
|
|
body: SpoolParameters,
|
|
):
|
|
if body.remaining_weight is not None and body.used_weight is not None:
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"message": "Only specify either remaining_weight or used_weight."},
|
|
)
|
|
|
|
if body.extra:
|
|
all_fields = await get_extra_fields(db, EntityType.spool)
|
|
try:
|
|
validate_extra_field_dict(all_fields, body.extra)
|
|
except ValueError as e:
|
|
return JSONResponse(status_code=400, content=Message(message=str(e)).dict())
|
|
|
|
try:
|
|
db_item = await spool.create(
|
|
db=db,
|
|
filament_id=body.filament_id,
|
|
price=body.price,
|
|
initial_weight=body.initial_weight,
|
|
spool_weight=body.spool_weight,
|
|
remaining_weight=body.remaining_weight,
|
|
used_weight=body.used_weight,
|
|
first_used=body.first_used,
|
|
last_used=body.last_used,
|
|
location=body.location,
|
|
lot_nr=body.lot_nr,
|
|
comment=body.comment,
|
|
archived=body.archived,
|
|
extra=body.extra,
|
|
)
|
|
return Spool.from_db(db_item)
|
|
except ItemCreateError:
|
|
logger.exception("Failed to create spool.")
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"message": "Failed to create spool, see server logs for more information."},
|
|
)
|
|
|
|
|
|
@router.patch(
|
|
"/{spool_id}",
|
|
name="Update spool",
|
|
description=(
|
|
"Update any attribute of a spool. "
|
|
"Only fields specified in the request will be affected. "
|
|
"remaining_weight and used_weight can't be set at the same time. "
|
|
"If extra is set, all existing extra fields will be removed and replaced with the new ones."
|
|
),
|
|
response_model_exclude_none=True,
|
|
response_model=Spool,
|
|
responses={
|
|
400: {"model": Message},
|
|
404: {"model": Message},
|
|
},
|
|
dependencies=[Depends(require_level(Level.EDIT))],
|
|
)
|
|
async def update( # noqa: ANN201
|
|
db: Annotated[AsyncSession, Depends(get_db_session)],
|
|
spool_id: int,
|
|
body: SpoolUpdateParameters,
|
|
):
|
|
patch_data = body.model_dump(exclude_unset=True)
|
|
|
|
if body.remaining_weight is not None and body.used_weight is not None:
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"message": "Only specify either remaining_weight or used_weight."},
|
|
)
|
|
|
|
if body.extra:
|
|
all_fields = await get_extra_fields(db, EntityType.spool)
|
|
try:
|
|
validate_extra_field_dict(all_fields, body.extra)
|
|
except ValueError as e:
|
|
return JSONResponse(status_code=400, content=Message(message=str(e)).dict())
|
|
|
|
try:
|
|
db_item = await spool.update(
|
|
db=db,
|
|
spool_id=spool_id,
|
|
data=patch_data,
|
|
)
|
|
except ItemCreateError:
|
|
logger.exception("Failed to update spool.")
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"message": "Failed to update spool, see server logs for more information."},
|
|
)
|
|
|
|
return Spool.from_db(db_item)
|
|
|
|
|
|
@router.delete(
|
|
"/{spool_id}",
|
|
name="Delete spool",
|
|
description="Delete a spool.",
|
|
responses={404: {"model": Message}},
|
|
dependencies=[Depends(require_level(Level.MANAGE))],
|
|
)
|
|
async def delete(
|
|
db: Annotated[AsyncSession, Depends(get_db_session)],
|
|
spool_id: int,
|
|
) -> Message:
|
|
await spool.delete(db, spool_id)
|
|
return Message(message="Success!")
|
|
|
|
|
|
@router.put(
|
|
"/{spool_id}/use",
|
|
name="Use spool filament",
|
|
description=(
|
|
"Use some length or weight of filament from the spool. Specify either a length or a weight, not both."
|
|
),
|
|
response_model_exclude_none=True,
|
|
response_model=Spool,
|
|
responses={
|
|
400: {"model": Message},
|
|
404: {"model": Message},
|
|
},
|
|
dependencies=[Depends(require_level(Level.EDIT))],
|
|
)
|
|
async def use( # noqa: ANN201
|
|
db: Annotated[AsyncSession, Depends(get_db_session)],
|
|
spool_id: int,
|
|
body: SpoolUseParameters,
|
|
):
|
|
if body.use_weight is not None and body.use_length is not None:
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"message": "Only specify either use_weight or use_length."},
|
|
)
|
|
|
|
if body.use_weight is not None:
|
|
db_item = await spool.use_weight(db, spool_id, body.use_weight)
|
|
return Spool.from_db(db_item)
|
|
|
|
if body.use_length is not None:
|
|
db_item = await spool.use_length(db, spool_id, body.use_length)
|
|
return Spool.from_db(db_item)
|
|
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"message": "Either use_weight or use_length must be specified."},
|
|
)
|
|
|
|
|
|
@router.put(
|
|
"/{spool_id}/measure",
|
|
name="Use spool filament based on the current weight measurement",
|
|
description=("Use some weight of filament from the spool. Specify the current gross weight of the spool."),
|
|
response_model_exclude_none=True,
|
|
response_model=Spool,
|
|
responses={
|
|
400: {"model": Message},
|
|
404: {"model": Message},
|
|
},
|
|
dependencies=[Depends(require_level(Level.EDIT))],
|
|
)
|
|
async def measure( # noqa: ANN201
|
|
db: Annotated[AsyncSession, Depends(get_db_session)],
|
|
spool_id: int,
|
|
body: SpoolMeasureParameters,
|
|
):
|
|
try:
|
|
db_item = await spool.measure(db, spool_id, body.weight)
|
|
return Spool.from_db(db_item)
|
|
except SpoolMeasureError as e:
|
|
logger.exception("Failed to update spool measurement.")
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"message": e.args[0]},
|
|
)
|