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

Added support for multiple int where's

Browse Source
This commit is contained in:
Donkie
2023-09-09 10:18:26 +02:00
parent 6d68a40313
commit 6ee3b8df70
8 changed files with 168 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -119,42 +119,55 @@ async def find(
alias="vendor_name",
default=None,
title="Vendor Name",
description="Partial case-insensitive search term for the filament vendor name.",
description=(
"Partial case-insensitive search term for the filament vendor name. "
"Separate multiple terms with a comma."
),
deprecated=True,
),
vendor_id_old: Optional[int] = Query(
vendor_id_old: Optional[str] = Query(
alias="vendor_id",
default=None,
title="Vendor ID",
description="Match an exact vendor ID.",
description="Match an exact vendor ID. Separate multiple IDs with a comma.",
deprecated=True,
examples=["1", "1,2"],
),
vendor_name: Optional[str] = Query(
alias="vendor.name",
default=None,
title="Vendor Name",
description="Partial case-insensitive search term for the filament vendor name.",
description=(
"Partial case-insensitive search term for the filament vendor name. "
"Separate multiple terms with a comma."
),
),
vendor_id: Optional[int] = Query(
vendor_id: Optional[str] = Query(
alias="vendor.id",
default=None,
title="Vendor ID",
description="Match an exact vendor ID.",
description="Match an exact vendor ID. Separate multiple IDs with a comma.",
examples=["1", "1,2"],
),
name: Optional[str] = Query(
default=None,
title="Filament Name",
description="Partial case-insensitive search term for the filament name.",
description="Partial case-insensitive search term for the filament name. Separate multiple terms with a comma.",
),
material: Optional[str] = Query(
default=None,
title="Filament Material",
description="Partial case-insensitive search term for the filament material.",
description=(
"Partial case-insensitive search term for the filament material. Separate multiple terms with a comma."
),
),
article_number: Optional[str] = Query(
default=None,
title="Filament Article Number",
description="Partial case-insensitive search term for the filament article number.",
description=(
"Partial case-insensitive search term for the filament article number. "
"Separate multiple terms with a comma."
),
),
sort: Optional[str] = Query(
default=None,
@@ -181,10 +194,19 @@ async def find(
field, direction = sort_item.split(":")
sort_by[field] = SortOrder[direction.upper()]
vendor_id = vendor_id if vendor_id is not None else vendor_id_old
if vendor_id is not None:
try:
vendor_ids = [int(vendor_id_item) for vendor_id_item in vendor_id.split(",")]
except ValueError as e:
raise RequestValidationError([ErrorWrapper(ValueError("Invalid vendor_id"), ("query", "vendor_id"))]) from e
else:
vendor_ids = None
db_items, total_count = await filament.find(
db=db,
vendor_name=vendor_name if vendor_name is not None else vendor_name_old,
vendor_id=vendor_id if vendor_id is not None else vendor_id_old,
vendor_id=vendor_ids,
name=name,
material=material,
article_number=article_number,

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

@@ -77,76 +77,92 @@ async def find(
alias="filament_name",
default=None,
title="Filament Name",
description="Partial case-insensitive search term for the filament name.",
description="Partial case-insensitive search term for the filament name. Separate multiple terms with a comma.",
deprecated=True,
),
filament_id_old: Optional[int] = Query(
filament_id_old: Optional[str] = Query(
alias="filament_id",
default=None,
title="Filament ID",
description="Match an exact filament ID.",
description="Match an exact filament ID. Separate multiple IDs with a comma.",
deprecated=True,
examples=["1", "1,2"],
),
filament_material_old: Optional[str] = Query(
alias="filament_material",
default=None,
title="Filament Material",
description="Partial case-insensitive search term for the filament material.",
description=(
"Partial case-insensitive search term for the filament material. Separate multiple terms with a comma."
),
deprecated=True,
),
vendor_name_old: Optional[str] = Query(
alias="vendor_name",
default=None,
title="Vendor Name",
description="Partial case-insensitive search term for the filament vendor name.",
description=(
"Partial case-insensitive search term for the filament vendor name. Separate multiple terms with a comma."
),
deprecated=True,
),
vendor_id_old: Optional[int] = Query(
vendor_id_old: Optional[str] = Query(
alias="vendor_id",
default=None,
title="Vendor ID",
description="Match an exact vendor ID.",
description="Match an exact vendor ID. Separate multiple IDs with a comma.",
deprecated=True,
examples=["1", "1,2"],
),
filament_name: Optional[str] = Query(
alias="filament.name",
default=None,
title="Filament Name",
description="Partial case-insensitive search term for the filament name.",
description="Partial case-insensitive search term for the filament name. Separate multiple terms with a comma.",
),
filament_id: Optional[int] = Query(
filament_id: Optional[str] = Query(
alias="filament.id",
default=None,
title="Filament ID",
description="Match an exact filament ID.",
description="Match an exact filament ID. Separate multiple IDs with a comma.",
examples=["1", "1,2"],
),
filament_material: Optional[str] = Query(
alias="filament.material",
default=None,
title="Filament Material",
description="Partial case-insensitive search term for the filament material.",
description=(
"Partial case-insensitive search term for the filament material. Separate multiple terms with a comma."
),
),
vendor_name: Optional[str] = Query(
alias="vendor.name",
default=None,
title="Vendor Name",
description="Partial case-insensitive search term for the filament vendor name.",
description=(
"Partial case-insensitive search term for the filament vendor name. Separate multiple terms with a comma."
),
),
vendor_id: Optional[int] = Query(
vendor_id: Optional[str] = Query(
alias="vendor.id",
default=None,
title="Vendor ID",
description="Match an exact vendor ID.",
description="Match an exact vendor ID. Separate multiple IDs with a comma.",
examples=["1", "1,2"],
),
location: Optional[str] = Query(
default=None,
title="Location",
description="Partial case-insensitive search term for the spool location.",
description=(
"Partial case-insensitive search term for the spool location. Separate multiple terms with a comma."
),
),
lot_nr: Optional[str] = Query(
default=None,
title="Lot/Batch Number",
description="Partial case-insensitive search term for the spool lot number.",
description=(
"Partial case-insensitive search term for the spool lot number. Separate multiple terms with a comma."
),
),
allow_archived: bool = Query(
default=False,
@@ -178,13 +194,33 @@ async def find(
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:
try:
filament_ids = [int(filament_id_item) for filament_id_item in filament_id.split(",")]
except ValueError as e:
raise RequestValidationError(
[ErrorWrapper(ValueError("Invalid filament_id"), ("query", "filament_id"))],
) from e
else:
filament_ids = None
vendor_id = vendor_id if vendor_id is not None else vendor_id_old
if vendor_id is not None:
try:
vendor_ids = [int(vendor_id_item) for vendor_id_item in vendor_id.split(",")]
except ValueError as e:
raise RequestValidationError([ErrorWrapper(ValueError("Invalid vendor_id"), ("query", "vendor_id"))]) from e
else:
vendor_ids = None
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_id if filament_id is not None else filament_id_old,
filament_id=filament_ids,
filament_material=filament_material if filament_material is not None else filament_material_old,
vendor_name=vendor_name if vendor_name is not None else vendor_name_old,
vendor_id=vendor_id if vendor_id is not None else vendor_id_old,
vendor_id=vendor_ids,
location=location,
lot_nr=lot_nr,
allow_archived=allow_archived,

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

@@ -53,7 +53,7 @@ async def find(
name: Optional[str] = Query(
default=None,
title="Vendor Name",
description="Partial case-insensitive search term for the vendor name.",
description="Partial case-insensitive search term for the vendor name. Separate multiple terms with a comma.",
),
sort: Optional[str] = Query(
default=None,

5
spoolman/database/filament.py
View File

@@ -1,7 +1,8 @@
"""Helper functions for interacting with filament database objects."""
import logging
from typing import Optional
from collections.abc import Sequence
from typing import Optional, Union
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
@@ -77,7 +78,7 @@ async def find(
*,
db: AsyncSession,
vendor_name: Optional[str] = None,
vendor_id: Optional[int] = None,
vendor_id: Optional[Union[int, Sequence[int]]] = None,
name: Optional[str] = None,
material: Optional[str] = None,
article_number: Optional[str] = None,

7
spoolman/database/spool.py
View File

@@ -1,7 +1,8 @@
"""Helper functions for interacting with spool database objects."""
from collections.abc import Sequence
from datetime import datetime, timezone
from typing import Optional
from typing import Optional, Union
import sqlalchemy
from sqlalchemy import case, func
@@ -87,10 +88,10 @@ async def find(
*,
db: AsyncSession,
filament_name: Optional[str] = None,
filament_id: Optional[int] = None,
filament_id: Optional[Union[int, Sequence[int]]] = None,
filament_material: Optional[str] = None,
vendor_name: Optional[str] = None,
vendor_id: Optional[int] = None,
vendor_id: Optional[Union[int, Sequence[int]]] = None,
location: Optional[str] = None,
lot_nr: Optional[str] = None,
allow_archived: bool = False,

24
spoolman/database/utils.py
View File

@@ -1,7 +1,8 @@
"""Utility functions for the database module."""
from collections.abc import Sequence
from enum import Enum
from typing import Any, Optional
from typing import Any, Optional, Union
import sqlalchemy
from sqlalchemy import Select
@@ -68,23 +69,30 @@ def add_where_clause_str(
def add_where_clause_int(
stmt: Select,
field: attributes.InstrumentedAttribute[int],
value: Optional[int],
value: Optional[Union[int, Sequence[int]]],
) -> Select:
"""Add a where clause to a select statement for a field."""
if value is not None:
stmt = stmt.where(field == value)
if isinstance(value, int):
value = [value]
stmt = stmt.where(field.in_(value))
return stmt
def add_where_clause_int_opt(
stmt: Select,
field: attributes.InstrumentedAttribute[Optional[int]],
value: Optional[int],
value: Optional[Union[int, Sequence[int]]],
) -> Select:
"""Add a where clause to a select statement for a field."""
if value is not None:
if value == -1:
stmt = stmt.where(field.is_(None))
else:
stmt = stmt.where(field == value)
if isinstance(value, int):
value = [value]
statements = []
for value_part in value:
if value_part == -1:
statements.append(field.is_(None))
else:
statements.append(field == value_part)
stmt = stmt.where(sqlalchemy.or_(*statements))
return stmt

22
tests_integration/tests/filament/test_find.py
View File

@@ -332,6 +332,28 @@ def test_find_filaments_by_vendor_id(filaments: Fixture, field_name: str):
assert filament_lists_equal(filaments_result, filaments.filaments[:2])
def test_find_filaments_by_multiple_vendor_ids(filaments: Fixture):
# Execute
vendor_1 = filaments.filaments[0]["vendor"]["id"]
vendor_2 = filaments.filaments[4]["vendor"]["id"]
result = httpx.get(
f"{URL}/api/v1/filament",
params={"vendor.id": f"{vendor_1},{vendor_2}"},
)
result.raise_for_status()
# Verify
filaments_result = result.json()
assert filament_lists_equal(
filaments_result,
[
filaments.filaments[0],
filaments.filaments[1],
filaments.filaments[4],
],
)
def test_find_filaments_by_empty_vendor_id(filaments: Fixture):
# Execute
result = httpx.get(

36
tests_integration/tests/spool/test_find.py
View File

@@ -302,6 +302,22 @@ def test_find_spools_by_filament_id(spools: Fixture, field_name: str):
assert spool_lists_equal(spools_result, (spools.spools[0], spools.spools[1]))
def test_find_spools_by_multiple_filament_ids(spools: Fixture):
# Execute
filament_1 = spools.spools[0]["filament"]["id"]
filament_2 = spools.spools[3]["filament"]["id"]
result = httpx.get(
f"{URL}/api/v1/spool",
params={"filament.id": f"{filament_1},{filament_2}"},
)
result.raise_for_status()
# Verify
spools_result = result.json()
assert spool_lists_equal(spools_result, (spools.spools[0], spools.spools[1], spools.spools[3]))
@pytest.mark.parametrize("field_name", ["filament_material", "filament.material"])
def test_find_spools_by_filament_material(spools: Fixture, field_name: str):
# Execute
@@ -370,6 +386,26 @@ def test_find_spools_by_filament_vendor_id(spools: Fixture, field_name: str):
assert spool_lists_equal(spools_result, (spools.spools[0], spools.spools[1]))
def test_find_spools_by_multiple_vendor_ids(spools: Fixture):
# Execute
vendor_1 = spools.spools[0]["filament"]["vendor"]["id"]
vendor_2 = spools.spools[4]["filament"]["vendor"]["id"]
result = httpx.get(
f"{URL}/api/v1/spool",
params={
"vendor.id": f"{vendor_1},{vendor_2}",
"allow_archived": True,
},
)
result.raise_for_status()
# Verify
spools_result = result.json()
assert len(spools_result) == 4
assert spool_lists_equal(spools_result, (spools.spools[0], spools.spools[1], spools.spools[2], spools.spools[4]))
def test_find_spools_by_empty_filament_vendor_id(spools: Fixture):
# Execute
result = httpx.get(