tonym/spoolman2
1
0
Fork 0
Code Issues 19 Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Merge branch 'master' into master

Browse Source
This commit is contained in:
ecarjat
2025-06-06 09:22:44 +02:00
committed by GitHub
parent 1ecf70773f 488b37c5a0
commit de7fccdc1a
65 changed files with 3854 additions and 1975 deletions
Download Patch File Download Diff File Expand all files Collapse all files

216
spoolman/api/v1/filament.py
View File

@@ -121,14 +121,13 @@ class FilamentParameters(BaseModel):
@field_validator("color_hex")
@classmethod
def color_hex_validator(cls, v: Optional[str]) -> Optional[str]: # noqa: ANN102
def color_hex_validator(cls, v: Optional[str]) -> Optional[str]:
"""Validate the color_hex field."""
if not v:
return None
clr = v.upper()
if clr.startswith("#"):
clr = clr[1:]
clr = clr.removeprefix("#")
for c in clr:
if c not in "0123456789ABCDEF":
@@ -141,14 +140,13 @@ class FilamentParameters(BaseModel):
@field_validator("multi_color_hexes")
@classmethod
def multi_color_hexes_validator(cls, v: Optional[str]) -> Optional[str]: # noqa: ANN102
def multi_color_hexes_validator(cls, v: Optional[str]) -> Optional[str]:
"""Validate the multi_color_hexes field."""
if not v:
return None
for clr_raw in v.split(","):
clr = clr_raw.upper()
if clr.startswith("#"):
clr = clr[1:]
clr = clr.removeprefix("#")
for c in clr:
if c not in "0123456789ABCDEF":
@@ -204,111 +202,123 @@ class FilamentUpdateParameters(FilamentParameters):
async def find(
*,
db: Annotated[AsyncSession, Depends(get_db_session)],
vendor_name_old: Optional[str] = Query(
alias="vendor_name",
default=None,
title="Vendor Name",
description="See vendor.name.",
deprecated=True,
),
vendor_id_old: Optional[str] = Query(
alias="vendor_id",
default=None,
title="Vendor ID",
description="See vendor.id.",
deprecated=True,
pattern=r"^-?\d+(,-?\d+)*$",
),
vendor_name: Optional[str] = Query(
alias="vendor.name",
default=None,
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 filaments with no vendor name. "
"Surround a term with quotes to search for the exact term."
vendor_name_old: Annotated[
Optional[str],
Query(alias="vendor_name", title="Vendor Name", description="See vendor.name.", deprecated=True),
] = None,
vendor_id_old: Annotated[
Optional[str],
Query(
alias="vendor_id",
title="Vendor ID",
description="See vendor.id.",
deprecated=True,
pattern=r"^-?\d+(,-?\d+)*$",
),
),
vendor_id: Optional[str] = Query(
alias="vendor.id",
default=None,
title="Vendor ID",
description=(
"Match an exact vendor ID. Separate multiple IDs with a comma. "
"Specify -1 to match filaments with no vendor."
] = None,
vendor_name: Annotated[
Optional[str],
Query(
alias="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 filaments with no vendor name. "
"Surround a term with quotes to search for the exact term."
),
),
pattern=r"^-?\d+(,-?\d+)*$",
examples=["1", "1,2"],
),
name: Optional[str] = Query(
default=None,
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 filaments with no name. "
"Surround a term with quotes to search for the exact term."
] = None,
vendor_id: Annotated[
Optional[str],
Query(
alias="vendor.id",
title="Vendor ID",
description=(
"Match an exact vendor ID. Separate multiple IDs with a comma. "
"Specify -1 to match filaments with no vendor."
),
pattern=r"^-?\d+(,-?\d+)*$",
examples=["1", "1,2"],
),
),
material: Optional[str] = Query(
default=None,
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 filaments with no material. "
"Surround a term with quotes to search for the exact term."
] = None,
name: Annotated[
Optional[str],
Query(
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 filaments with no name. "
"Surround a term with quotes to search for the exact term."
),
),
),
article_number: Optional[str] = Query(
default=None,
title="Filament Article Number",
description=(
"Partial case-insensitive search term for the filament article number. "
"Separate multiple terms with a comma. "
"Specify an empty string to match filaments with no article number. "
"Surround a term with quotes to search for the exact term."
] = None,
material: Annotated[
Optional[str],
Query(
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 filaments with no material. "
"Surround a term with quotes to search for the exact term."
),
),
),
color_hex: Optional[str] = Query(
default=None,
title="Filament Color",
description="Match filament by similar color. Slow operation!",
),
color_similarity_threshold: float = Query(
default=20.0,
description=(
"The similarity threshold for color matching. "
"A value between 0.0-100.0, where 0 means match only exactly the same color."
] = None,
article_number: Annotated[
Optional[str],
Query(
title="Filament Article Number",
description=(
"Partial case-insensitive search term for the filament article number. "
"Separate multiple terms with a comma. "
"Specify an empty string to match filaments with no article number. "
"Surround a term with quotes to search for the exact term."
),
),
example=20.0,
),
external_id: Optional[str] = Query(
default=None,
description=(
"Find filaments imported by the given external ID. "
"Separate multiple IDs with a comma. "
"Specify empty string to match filaments with no external ID. "
"Surround a term with quotes to search for the exact term."
] = None,
color_hex: Annotated[
Optional[str],
Query(
title="Filament Color",
description="Match filament by similar color. Slow operation!",
),
example="polymaker_pla_polysonicblack_1000_175",
),
sort: Optional[str] = Query(
default=None,
title="Sort",
description=(
'Sort the results by the given field. Should be a comma-separate string with "field:direction" items.'
] = None,
color_similarity_threshold: Annotated[
float,
Query(
description=(
"The similarity threshold for color matching. "
"A value between 0.0-100.0, where 0 means match only exactly the same color."
),
example=20.0,
),
example="vendor.name:asc,spool_weight:desc",
),
limit: Optional[int] = Query(
default=None,
title="Limit",
description="Maximum number of items in the response.",
),
offset: int = Query(
default=0,
title="Offset",
description="Offset in the full result set if a limit is set.",
),
] = 20.0,
external_id: Annotated[
Optional[str],
Query(
description=(
"Find filaments imported by the given external ID. "
"Separate multiple IDs with a comma. "
"Specify empty string to match filaments with no external ID. "
"Surround a term with quotes to search for the exact term."
),
example="polymaker_pla_polysonicblack_1000_175",
),
] = None,
sort: Annotated[
Optional[str],
Query(
title="Sort",
description=(
'Sort the results by the given field. Should be a comma-separate string with "field:direction" items.'
),
example="vendor.name:asc,spool_weight:desc",
),
] = None,
limit: Annotated[
Optional[int],
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:

251
spoolman/api/v1/spool.py
View File

@@ -128,133 +128,144 @@ class SpoolMeasureParameters(BaseModel):
async def find(
*,
db: Annotated[AsyncSession, Depends(get_db_session)],
filament_name_old: Optional[str] = Query(
alias="filament_name",
default=None,
title="Filament Name",
description="See filament.name.",
deprecated=True,
),
filament_id_old: Optional[str] = Query(
alias="filament_id",
default=None,
title="Filament ID",
description="See filament.id.",
deprecated=True,
pattern=r"^-?\d+(,-?\d+)*$",
),
filament_material_old: Optional[str] = Query(
alias="filament_material",
default=None,
title="Filament Material",
description="See filament.material.",
deprecated=True,
),
vendor_name_old: Optional[str] = Query(
alias="vendor_name",
default=None,
title="Vendor Name",
description="See filament.vendor.name.",
deprecated=True,
),
vendor_id_old: Optional[str] = Query(
alias="vendor_id",
default=None,
title="Vendor ID",
description="See filament.vendor.id.",
deprecated=True,
pattern=r"^-?\d+(,-?\d+)*$",
),
filament_name: Optional[str] = Query(
alias="filament.name",
default=None,
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."
filament_name_old: Annotated[
Optional[str],
Query(alias="filament_name", title="Filament Name", description="See filament.name.", deprecated=True),
] = None,
filament_id_old: Annotated[
Optional[str],
Query(
alias="filament_id",
title="Filament ID",
description="See filament.id.",
deprecated=True,
pattern=r"^-?\d+(,-?\d+)*$",
),
),
filament_id: Optional[str] = Query(
alias="filament.id",
default=None,
title="Filament ID",
description="Match an exact filament ID. Separate multiple IDs with a comma.",
examples=["1", "1,2"],
pattern=r"^-?\d+(,-?\d+)*$",
),
filament_material: Optional[str] = Query(
alias="filament.material",
default=None,
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_material_old: Annotated[
Optional[str],
Query(
alias="filament_material",
title="Filament Material",
description="See filament.material.",
deprecated=True,
),
),
filament_vendor_name: Optional[str] = Query(
alias="filament.vendor.name",
default=None,
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,
vendor_name_old: Annotated[
Optional[str],
Query(alias="vendor_name", title="Vendor Name", description="See filament.vendor.name.", deprecated=True),
] = None,
vendor_id_old: Annotated[
Optional[str],
Query(
alias="vendor_id",
title="Vendor ID",
description="See filament.vendor.id.",
deprecated=True,
pattern=r"^-?\d+(,-?\d+)*$",
),
),
filament_vendor_id: Optional[str] = Query(
alias="filament.vendor.id",
default=None,
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."
] = None,
filament_name: Annotated[
Optional[str],
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."
),
),
examples=["1", "1,2"],
pattern=r"^-?\d+(,-?\d+)*$",
),
location: Optional[str] = Query(
default=None,
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,
filament_id: Annotated[
Optional[str],
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+)*$",
),
),
lot_nr: Optional[str] = Query(
default=None,
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,
filament_material: Annotated[
Optional[str],
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."
),
),
),
allow_archived: bool = Query(
default=False,
title="Allow Archived",
description="Whether to include archived spools in the search results.",
),
sort: Optional[str] = Query(
default=None,
title="Sort",
description=(
'Sort the results by the given field. Should be a comma-separate string with "field:direction" items.'
] = None,
filament_vendor_name: Annotated[
Optional[str],
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."
),
),
example="filament.name:asc,filament.vendor.id:asc,location:desc",
),
limit: Optional[int] = Query(
default=None,
title="Limit",
description="Maximum number of items in the response.",
),
offset: int = Query(
default=0,
title="Offset",
description="Offset in the full result set if a limit is set.",
),
] = None,
filament_vendor_id: Annotated[
Optional[str],
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[
Optional[str],
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[
Optional[str],
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[
Optional[str],
Query(
title="Sort",
description=(
'Sort the results by the given field. Should be a comma-separate string with "field:direction" items.'
),
example="filament.name:asc,filament.vendor.id:asc,location:desc",
),
] = None,
limit: Annotated[
Optional[int],
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:

67
spoolman/api/v1/vendor.py
View File

@@ -80,42 +80,43 @@ class VendorUpdateParameters(VendorParameters):
)
async def find(
db: Annotated[AsyncSession, Depends(get_db_session)],
name: Optional[str] = Query(
default=None,
title="Vendor Name",
description=(
"Partial case-insensitive search term for the vendor name. Separate multiple terms with a comma. "
"Surround a term with quotes to search for the exact term."
name: Annotated[
Optional[str],
Query(
title="Vendor Name",
description=(
"Partial case-insensitive search term for the vendor name. Separate multiple terms with a comma. "
"Surround a term with quotes to search for the exact term."
),
),
),
external_id: Optional[str] = Query(
default=None,
title="Vendor External ID",
description=(
"Exact match for the vendor external ID. "
"Separate multiple IDs with a comma. "
"Specify empty string to match filaments with no external ID. "
"Surround a term with quotes to search for the exact term."
] = None,
external_id: Annotated[
Optional[str],
Query(
title="Vendor External ID",
description=(
"Exact match for the vendor external ID. "
"Separate multiple IDs with a comma. "
"Specify empty string to match filaments with no external ID. "
"Surround a term with quotes to search for the exact term."
),
),
),
sort: Optional[str] = Query(
default=None,
title="Sort",
description=(
'Sort the results by the given field. Should be a comma-separate string with "field:direction" items.'
] = None,
sort: Annotated[
Optional[str],
Query(
title="Sort",
description=(
'Sort the results by the given field. Should be a comma-separate string with "field:direction" items.'
),
example="name:asc,id:desc",
),
example="name:asc,id:desc",
),
limit: Optional[int] = Query(
default=None,
title="Limit",
description="Maximum number of items in the response.",
),
offset: int = Query(
default=0,
title="Offset",
description="Offset in the full result set if a limit is set.",
),
] = None,
limit: Annotated[
Optional[int],
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:

15
spoolman/database/database.py
View File

@@ -59,11 +59,11 @@ class Database:
engine: Optional[AsyncEngine]
session_maker: Optional[async_sessionmaker[AsyncSession]]
def __init__(self: "Database", connection_url: URL) -> None:
def __init__(self, connection_url: URL) -> None:
"""Construct the Database wrapper and set config parameters."""
self.connection_url = connection_url
def is_file_based_sqlite(self: "Database") -> bool:
def is_file_based_sqlite(self) -> bool:
"""Return True if the database is file based."""
return (
self.connection_url.drivername[:6] == "sqlite"
@@ -71,7 +71,7 @@ class Database:
and self.connection_url.database != ":memory:"
)
def connect(self: "Database") -> None:
def connect(self) -> None:
"""Connect to the database."""
if env.get_logging_level() == logging.DEBUG:
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)
@@ -79,15 +79,18 @@ class Database:
connect_args = {}
if self.connection_url.drivername == "sqlite+aiosqlite":
connect_args["timeout"] = 60
connection_options = {}
if self.connection_url.drivername == "mysql+aiomysql":
connection_options["pool_recycle"] = 3600
self.engine = create_async_engine(
self.connection_url,
connect_args=connect_args,
pool_pre_ping=True,
**connection_options,
)
self.session_maker = async_sessionmaker(self.engine, autocommit=False, autoflush=True, expire_on_commit=False)
def backup(self: "Database", target_path: Union[str, PathLike[str]]) -> None:
def backup(self, target_path: Union[str, PathLike[str]]) -> None:
"""Backup the database."""
if not self.is_file_based_sqlite() or self.connection_url.database is None:
return
@@ -108,7 +111,7 @@ class Database:
logger.info("Backup complete.")
def backup_and_rotate(
self: "Database",
self,
backup_folder: Union[str, PathLike[str]],
num_backups: int = 5,
) -> Optional[Path]:

3
spoolman/database/spool.py
View File

@@ -231,7 +231,8 @@ async def update(
elif isinstance(v, datetime):
setattr(spool, k, utc_timezone_naive(v))
elif k == "extra":
spool.extra = [models.SpoolField(key=k, value=v) for k, v in v.items()]
spool.extra = [f for f in spool.extra if f.key not in v]
spool.extra.extend([models.SpoolField(key=k, value=v) for k, v in v.items()])
else:
setattr(spool, k, v)
await db.commit()

2
spoolman/database/utils.py
View File

@@ -126,6 +126,6 @@ def add_where_clause_int_in(
value: Optional[Sequence[T]],
) -> Select:
"""Add a where clause to a select statement for a field."""
if value is not None and len(value) > 0:
if value is not None:
stmt = stmt.where(field.in_(value))
return stmt

2
spoolman/env.py
View File

@@ -23,7 +23,7 @@ class DatabaseType(Enum):
SQLITE = "sqlite"
COCKROACHDB = "cockroachdb"
def to_drivername(self: "DatabaseType") -> str:
def to_drivername(self) -> str:
"""Get the drivername for the database type.
Returns:

7
spoolman/externaldb.py
View File

@@ -7,6 +7,7 @@ from collections.abc import Iterator
from enum import Enum
from pathlib import Path
from typing import Optional
from urllib.parse import urljoin
import hishel
from pydantic import BaseModel, Field, RootModel
@@ -121,7 +122,7 @@ class ExternalMaterialsFile(RootModel):
def get_external_db_url() -> str:
"""Get the external database URL from environment variables. Defaults to DEFAULT_EXTERNAL_DB_URL."""
return os.getenv("EXTERNAL_DB_URL", DEFAULT_EXTERNAL_DB_URL).strip("/") + "/"
return os.getenv("EXTERNAL_DB_URL", DEFAULT_EXTERNAL_DB_URL)
def get_external_db_sync_interval() -> int:
@@ -170,8 +171,8 @@ async def _sync() -> None:
url = get_external_db_url()
filaments = _parse_filaments_from_bytes(await _download_file(url + "filaments.json"))
materials = _parse_materials_from_bytes(await _download_file(url + "materials.json"))
filaments = _parse_filaments_from_bytes(await _download_file(urljoin(url, "filaments.json")))
materials = _parse_materials_from_bytes(await _download_file(urljoin(url, "materials.json")))
_write_to_local_cache("filaments.json", filaments.json().encode())
_write_to_local_cache("materials.json", materials.json().encode())

13
spoolman/main.py
View File

@@ -31,15 +31,18 @@ root_logger.addHandler(console_handler)
# Fix uvicorn logging
logging.getLogger("uvicorn").setLevel(log_level)
logging.getLogger("uvicorn").removeHandler(logging.getLogger("uvicorn").handlers[0])
if logging.getLogger("uvicorn").handlers:
logging.getLogger("uvicorn").removeHandler(logging.getLogger("uvicorn").handlers[0])
logging.getLogger("uvicorn").addHandler(console_handler)
logging.getLogger("uvicorn.error").setLevel(log_level)
logging.getLogger("uvicorn.error").addHandler(console_handler)
logging.getLogger("uvicorn.access").setLevel(log_level)
logging.getLogger("uvicorn.access").removeHandler(logging.getLogger("uvicorn.access").handlers[0])
logging.getLogger("uvicorn.access").addHandler(console_handler)
access_handlers = logging.getLogger("uvicorn.access").handlers
if access_handlers:
logging.getLogger("uvicorn.access").setLevel(log_level)
logging.getLogger("uvicorn.access").removeHandler(access_handlers[0])
logging.getLogger("uvicorn.access").addHandler(console_handler)
# Get logger instance for this module
logger = logging.getLogger(__name__)
@@ -155,7 +158,7 @@ async def startup() -> None:
# There is some issue with the uvicorn worker that causes the process to hang when running alembic directly.
# See: https://github.com/sqlalchemy/alembic/discussions/1155
project_root = Path(__file__).parent.parent
subprocess.run(["alembic", "upgrade", "head"], check=True, cwd=project_root) # noqa: S603, S607, ASYNC101
subprocess.run(["alembic", "upgrade", "head"], check=True, cwd=project_root) # noqa: S603, S607, ASYNC221
# Setup scheduler
schedule = Scheduler()

1
spoolman/settings.py
View File

@@ -62,6 +62,7 @@ def parse_setting(key: str) -> SettingDefinition:
register_setting("currency", SettingType.STRING, json.dumps("EUR"))
register_setting("round_prices", SettingType.BOOLEAN, json.dumps(obj=False))
register_setting("print_presets", SettingType.ARRAY, json.dumps([]))
register_setting("extra_fields_vendor", SettingType.ARRAY, json.dumps([]))