diff --git a/pyproject.toml b/pyproject.toml index ef63865..aa77f3e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -61,6 +61,7 @@ ignore = [ "EM102", "DTZ003", "PLR0913", + "SIM108", ] line-length = 120 target-version = "py39" diff --git a/spoolman/database/filament.py b/spoolman/database/filament.py index 04c7cbb..657101f 100644 --- a/spoolman/database/filament.py +++ b/spoolman/database/filament.py @@ -9,7 +9,13 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import contains_eager, joinedload from spoolman.database import models, vendor -from spoolman.database.utils import SortOrder, parse_nested_field +from spoolman.database.utils import ( + SortOrder, + add_where_clause, + add_where_clause_str, + add_where_clause_str_opt, + parse_nested_field, +) from spoolman.exceptions import ItemDeleteError, ItemNotFoundError @@ -67,7 +73,7 @@ async def get_by_id(db: AsyncSession, filament_id: int) -> models.Filament: return filament -async def find( # noqa: C901 +async def find( *, db: AsyncSession, vendor_name: Optional[str] = None, @@ -91,16 +97,12 @@ async def find( # noqa: C901 .options(contains_eager(models.Filament.vendor)) .join(models.Filament.vendor, isouter=True) ) - if vendor_name is not None: - stmt = stmt.where(models.Vendor.name.ilike(f"%{vendor_name}%")) - if vendor_id is not None: - stmt = stmt.where(models.Filament.vendor_id == vendor_id) - if name is not None: - stmt = stmt.where(models.Filament.name.ilike(f"%{name}%")) - if material is not None: - stmt = stmt.where(models.Filament.material.ilike(f"%{material}%")) - if article_number is not None: - stmt = stmt.where(models.Filament.article_number.ilike(f"%{article_number}%")) + + stmt = add_where_clause(stmt, models.Filament.vendor_id, vendor_id) + stmt = add_where_clause_str(stmt, models.Vendor.name, vendor_name) + stmt = add_where_clause_str_opt(stmt, models.Filament.name, name) + stmt = add_where_clause_str_opt(stmt, models.Filament.material, material) + stmt = add_where_clause_str_opt(stmt, models.Filament.article_number, article_number) total_count = None diff --git a/spoolman/database/spool.py b/spoolman/database/spool.py index 467b8fe..d6a9a83 100644 --- a/spoolman/database/spool.py +++ b/spoolman/database/spool.py @@ -10,7 +10,13 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import contains_eager, joinedload from spoolman.database import filament, models -from spoolman.database.utils import SortOrder, parse_nested_field +from spoolman.database.utils import ( + SortOrder, + add_where_clause, + add_where_clause_str, + add_where_clause_str_opt, + parse_nested_field, +) from spoolman.exceptions import ItemCreateError, ItemNotFoundError from spoolman.math import weight_from_length @@ -76,7 +82,7 @@ async def get_by_id(db: AsyncSession, spool_id: int) -> models.Spool: return spool -async def find( # noqa: C901, PLR0912 +async def find( *, db: AsyncSession, filament_name: Optional[str] = None, @@ -104,20 +110,15 @@ async def find( # noqa: C901, PLR0912 .join(models.Filament.vendor, isouter=True) .options(contains_eager(models.Spool.filament).contains_eager(models.Filament.vendor)) ) - if filament_name is not None: - stmt = stmt.where(models.Filament.name.ilike(f"%{filament_name}%")) - if filament_id is not None: - stmt = stmt.where(models.Spool.filament_id == filament_id) - if filament_material is not None: - stmt = stmt.where(models.Filament.material.ilike(f"%{filament_material}%")) - if vendor_name is not None: - stmt = stmt.where(models.Vendor.name.ilike(f"%{vendor_name}%")) - if vendor_id is not None: - stmt = stmt.where(models.Filament.vendor_id == vendor_id) - if location is not None: - stmt = stmt.where(models.Spool.location.ilike(f"%{location}%")) - if lot_nr is not None: - stmt = stmt.where(models.Spool.lot_nr.ilike(f"%{lot_nr}%")) + + stmt = add_where_clause(stmt, models.Spool.filament_id, filament_id) + stmt = add_where_clause(stmt, models.Filament.vendor_id, vendor_id) + stmt = add_where_clause_str(stmt, models.Vendor.name, vendor_name) + stmt = add_where_clause_str_opt(stmt, models.Filament.name, filament_name) + stmt = add_where_clause_str_opt(stmt, models.Filament.material, filament_material) + stmt = add_where_clause_str_opt(stmt, models.Spool.location, location) + stmt = add_where_clause_str_opt(stmt, models.Spool.lot_nr, lot_nr) + if not allow_archived: # Since the archived field is nullable, and default is false, we need to check for both false or null stmt = stmt.where( diff --git a/spoolman/database/utils.py b/spoolman/database/utils.py index 5f95ad2..54e0c90 100644 --- a/spoolman/database/utils.py +++ b/spoolman/database/utils.py @@ -1,9 +1,11 @@ """Utility functions for the database module.""" from enum import Enum -from typing import Any +from typing import Any, Optional, TypeVar import sqlalchemy +from sqlalchemy import Select +from sqlalchemy.orm import attributes from spoolman.database import models @@ -13,7 +15,7 @@ class SortOrder(Enum): DESC = 2 -def parse_nested_field(base_obj: type[models.Base], field: str) -> sqlalchemy.Column[Any]: +def parse_nested_field(base_obj: type[models.Base], field: str) -> attributes.InstrumentedAttribute[Any]: """Parse a nested field string into a sqlalchemy field object.""" fields = field.split(".") if not hasattr(base_obj, fields[0]): @@ -33,3 +35,45 @@ def parse_nested_field(base_obj: type[models.Base], field: str) -> sqlalchemy.Co raise ValueError(f"Field '{fields[0]}' does not have any nested fields") return getattr(base_obj, fields[0]) + + +def add_where_clause_str_opt( + stmt: Select, + field: attributes.InstrumentedAttribute[Optional[str]], + value: Optional[str], +) -> Select: + """Add a where clause to a select statement for an optional string field.""" + if value is not None: + if len(value) == 0: + stmt = stmt.where(sqlalchemy.or_(field.is_(None), field == "")) + else: + stmt = stmt.where(field.ilike(f"%{value}%")) + return stmt + + +def add_where_clause_str( + stmt: Select, + field: attributes.InstrumentedAttribute[str], + value: Optional[str], +) -> Select: + """Add a where clause to a select statement for a string field.""" + if value is not None: + if len(value) == 0: + stmt = stmt.where(field == "") + else: + stmt = stmt.where(field.ilike(f"%{value}%")) + return stmt + + +T = TypeVar("T") + + +def add_where_clause( + stmt: Select, + field: attributes.InstrumentedAttribute[T], + value: Optional[T], +) -> Select: + """Add a where clause to a select statement for a field.""" + if value is not None: + stmt = stmt.where(field == value) + return stmt diff --git a/spoolman/database/vendor.py b/spoolman/database/vendor.py index a9e66a9..120c061 100644 --- a/spoolman/database/vendor.py +++ b/spoolman/database/vendor.py @@ -6,7 +6,7 @@ from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from spoolman.database import models -from spoolman.database.utils import SortOrder +from spoolman.database.utils import SortOrder, add_where_clause_str from spoolman.exceptions import ItemNotFoundError @@ -47,8 +47,8 @@ async def find( Returns a tuple containing the list of items and the total count of matching items. """ stmt = select(models.Vendor) - if name is not None: - stmt = stmt.where(models.Vendor.name.ilike(f"%{name}%")) + + stmt = add_where_clause_str(stmt, models.Vendor.name, name) total_count = None