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

Update python dependencies and ruff/black versions

Browse Source
This commit is contained in:
Donkie
2025-03-16 21:33:08 +01:00
parent 92065fd8df
commit 70ea7ad414
9 changed files with 295 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pyproject.toml
View File

@@ -60,7 +60,6 @@ select = [
"ALL",
]
ignore = [
"ANN101",
"A003",
"D101",
"D104",
@@ -85,6 +84,7 @@ ignore = [
[tool.ruff.lint.per-file-ignores]
"tests*/*" = [
"ANN201",
"ASYNC210",
"S101",
"PLR2004",
"D103",

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:

10
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)
@@ -87,7 +87,7 @@ class Database:
)
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 +108,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]:

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:

2
spoolman/main.py
View File

@@ -153,7 +153,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()

10
tests_integration/tests/conftest.py
View File

@@ -162,35 +162,35 @@ def random_empty_filament_empty_vendor_impl():
httpx.delete(f"{URL}/api/v1/filament/{filament['id']}").raise_for_status()
@pytest.fixture()
@pytest.fixture
def random_vendor():
"""Return a random vendor."""
with random_vendor_impl() as random_vendor:
yield random_vendor
@pytest.fixture()
@pytest.fixture
def random_empty_vendor():
"""Return a random vendor with only required fields specified."""
with random_empty_vendor_impl() as random_empty_vendor:
yield random_empty_vendor
@pytest.fixture()
@pytest.fixture
def random_filament():
"""Return a random filament."""
with random_filament_impl() as random_filament:
yield random_filament
@pytest.fixture()
@pytest.fixture
def random_empty_filament():
"""Return a random filament with only required fields specified."""
with random_empty_filament_impl() as random_empty_filament:
yield random_empty_filament
@pytest.fixture()
@pytest.fixture
def random_empty_filament_empty_vendor():
"""Return a random filament with only required fields specified and a vendor with only required fields specified."""
with random_empty_filament_empty_vendor_impl() as random_empty_filament_empty_vendor:

8
tests_integration/tests/spool/test_use.py
View File

@@ -138,12 +138,12 @@ def test_use_spool_not_found():
assert "123456789" in message
@pytest.mark.asyncio()
@pytest.mark.asyncio
async def test_use_spool_concurrent(random_filament: dict[str, Any]):
"""Test using a spool with many concurrent requests."""
# Setup
start_weight = 1000
result = httpx.post( # noqa: ASYNC100
result = httpx.post(
f"{URL}/api/v1/spool",
json={
"filament_id": random_filament["id"],
@@ -171,10 +171,10 @@ async def test_use_spool_concurrent(random_filament: dict[str, Any]):
)
# Verify
result = httpx.get(f"{URL}/api/v1/spool/{spool['id']}") # noqa: ASYNC100
result = httpx.get(f"{URL}/api/v1/spool/{spool['id']}")
result.raise_for_status()
spool = result.json()
assert spool["remaining_weight"] == pytest.approx(start_weight - (used_weight * requests))
# Clean up
httpx.delete(f"{URL}/api/v1/spool/{spool['id']}").raise_for_status() # noqa: ASYNC100
httpx.delete(f"{URL}/api/v1/spool/{spool['id']}").raise_for_status()