diff --git a/pyproject.toml b/pyproject.toml index 525b020..135814a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", diff --git a/spoolman/api/v1/filament.py b/spoolman/api/v1/filament.py index e164f01..07e3492 100644 --- a/spoolman/api/v1/filament.py +++ b/spoolman/api/v1/filament.py @@ -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: diff --git a/spoolman/api/v1/spool.py b/spoolman/api/v1/spool.py index aea65e3..0467ca8 100644 --- a/spoolman/api/v1/spool.py +++ b/spoolman/api/v1/spool.py @@ -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: diff --git a/spoolman/api/v1/vendor.py b/spoolman/api/v1/vendor.py index 14bc150..b175a0c 100644 --- a/spoolman/api/v1/vendor.py +++ b/spoolman/api/v1/vendor.py @@ -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: diff --git a/spoolman/database/database.py b/spoolman/database/database.py index 01003a0..e13fdf9 100644 --- a/spoolman/database/database.py +++ b/spoolman/database/database.py @@ -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]: diff --git a/spoolman/env.py b/spoolman/env.py index d4596c9..cbb4a40 100644 --- a/spoolman/env.py +++ b/spoolman/env.py @@ -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: diff --git a/spoolman/main.py b/spoolman/main.py index 3e48386..bd9a3d6 100644 --- a/spoolman/main.py +++ b/spoolman/main.py @@ -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() diff --git a/tests_integration/tests/conftest.py b/tests_integration/tests/conftest.py index 4e133da..36087a3 100644 --- a/tests_integration/tests/conftest.py +++ b/tests_integration/tests/conftest.py @@ -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: diff --git a/tests_integration/tests/spool/test_use.py b/tests_integration/tests/spool/test_use.py index 6c07116..9b59dea 100644 --- a/tests_integration/tests/spool/test_use.py +++ b/tests_integration/tests/spool/test_use.py @@ -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()