Support COT catalog number format configuration in WorkBench

specifyweb/specify/parse.py

@@ -43,11 +43,11 @@ class ParseSucess(NamedTuple):
 ParseResult = Union[ParseSucess, ParseFailure]
 
 
-def parse_field(collection, table_name: str, field_name: str, raw_value: str) -> ParseResult:
+def parse_field(collection, table_name: str, field_name: str, raw_value: str, with_formatter = None) -> ParseResult:
     table = datamodel.get_table_strict(table_name)
     field = table.get_field_strict(field_name)
 
-    formatter = get_uiformatter(collection, table_name, field_name)
+    formatter = get_uiformatter(collection, table_name, field_name) if with_formatter is None else with_formatter
 
     if field.is_relationship:
         return parse_integer(field.name, raw_value)

specifyweb/specify/uiformatters.py

@@ -324,8 +324,9 @@ def get_uiformatters(collection, obj, user) -> List[UIFormatter]:
         if f is not None
     ]
     if tablename.lower() == 'collectionobject':
-        cat_num_format = get_catalognumber_format(collection, obj, user)
-        if cat_num_format: uiformatters.append(cat_num_format)
+        cat_num_format = getattr(getattr(obj, 'collectionobjecttype', None), 'catalognumberformatname', None)
+        cat_num_formatter = get_catalognumber_format(collection, cat_num_format, user)
+        if cat_num_formatter: uiformatters.append(cat_num_formatter)
 
     logger.debug("uiformatters for %s: %s", tablename, uiformatters)
     return uiformatters
@@ -346,9 +347,10 @@ def get_uiformatter(collection, tablename: str, fieldname: str) -> Optional[UIFo
     else:
         return get_uiformatter_by_name(collection, None, field_format)
 
-def get_catalognumber_format(collection, collection_object, user) -> UIFormatter: 
-    cot_formatter = getattr(getattr(collection_object, 'collectionobjecttype', None), 'catalognumberformatname', None)
-    if cot_formatter: 
-        return get_uiformatter_by_name(collection, user, cot_formatter)
-    else: 
-        return get_uiformatter_by_name(collection, user, collection.catalognumformatname)
+def get_catalognumber_format(collection, format_name: Optional[str], user) -> UIFormatter:
+    if format_name: 
+         formatter = get_uiformatter_by_name(collection, user, format_name)
+         if formatter:
+            return formatter
+
+    return get_uiformatter_by_name(collection, user, collection.catalognumformatname)

specifyweb/workbench/upload/column_options.py

@@ -1,8 +1,20 @@
-from typing import List, Dict, Any, NamedTuple, Union, Optional, Set
+from typing import List, Dict, Any, NamedTuple, Union, Optional, Callable
 from typing_extensions import Literal
 
+from specifyweb.specify.uiformatters import UIFormatter
+
 MatchBehavior = Literal["ignoreWhenBlank", "ignoreAlways", "ignoreNever"]
 
+# A single row in the workbench. Maps column names to values in the row
+Row = Dict[str, str]
+
+""" The field formatter (uiformatter) for the column is determined by one or 
+more values for other columns in the WorkBench row. 
+
+See https://github.com/specify/specify7/issues/5473 
+"""
+DeferredUIFormatter = Callable[[Row], Optional[UIFormatter]]
+
 class ColumnOptions(NamedTuple):
     column: str
     matchBehavior: MatchBehavior
@@ -20,7 +32,7 @@ class ExtendedColumnOptions(NamedTuple):
     matchBehavior: MatchBehavior
     nullAllowed: bool
     default: Optional[str]
-    uiformatter: Any
+    uiformatter: Union[None, UIFormatter, DeferredUIFormatter]
     schemaitem: Any
     picklist: Any
     dateformat: Optional[str]

specifyweb/workbench/upload/parsing.py

@@ -55,7 +55,7 @@ def filter_and_upload(f: Filter, column: str) -> ParseResult:
 def parse_many(collection, tablename: str, mapping: Dict[str, ExtendedColumnOptions], row: Row) -> Tuple[List[ParseResult], List[WorkBenchParseFailure]]:
     results = [
         parse_value(collection, tablename, fieldname,
-                    row[colopts.column], colopts)
+                    row[colopts.column], colopts, row)
         for fieldname, colopts in mapping.items()
     ]
     return (
@@ -64,7 +64,7 @@ def parse_many(collection, tablename: str, mapping: Dict[str, ExtendedColumnOpti
     )
 
 
-def parse_value(collection, tablename: str, fieldname: str, value_in: str, colopts: ExtendedColumnOptions) -> Union[ParseResult, WorkBenchParseFailure]:
+def parse_value(collection, tablename: str, fieldname: str, value_in: str, colopts: ExtendedColumnOptions, row: Row) -> Union[ParseResult, WorkBenchParseFailure]:
     required_by_schema = colopts.schemaitem and colopts.schemaitem.isrequired
 
     result: Union[ParseResult, WorkBenchParseFailure]
@@ -80,10 +80,10 @@ def parse_value(collection, tablename: str, fieldname: str, value_in: str, colop
                                  None, colopts.column, missing_required)
         else:
             result = _parse(collection, tablename, fieldname,
-                            colopts, colopts.default)
+                            colopts, colopts.default, row)
     else:
         result = _parse(collection, tablename, fieldname,
-                        colopts, value_in.strip())
+                        colopts, value_in.strip(), row)
 
     if isinstance(result, WorkBenchParseFailure):
         return result
@@ -101,7 +101,7 @@ def parse_value(collection, tablename: str, fieldname: str, value_in: str, colop
         assertNever(colopts.matchBehavior)
 
 
-def _parse(collection, tablename: str, fieldname: str, colopts: ExtendedColumnOptions, value: str) -> Union[ParseResult, WorkBenchParseFailure]:
+def _parse(collection, tablename: str, fieldname: str, colopts: ExtendedColumnOptions, value: str, row: Row) -> Union[ParseResult, WorkBenchParseFailure]:
     table = datamodel.get_table_strict(tablename)
     field = table.get_field_strict(fieldname)
 
@@ -119,8 +119,8 @@ def _parse(collection, tablename: str, fieldname: str, colopts: ExtendedColumnOp
                     colopts.column
                 )
             return result
-
-    parsed = parse_field(collection, tablename, fieldname, value)
+    formatter = colopts.uiformatter(row) if callable(colopts.uiformatter) else colopts.uiformatter
+    parsed = parse_field(collection, tablename, fieldname, value, with_formatter=formatter)
 
     if is_latlong(table, field) and isinstance(parsed, ParseSucess):
         coord_text_field = field.name.replace('itude', '') + 'text'

specifyweb/workbench/upload/scoping.py

@@ -1,16 +1,16 @@
-from typing import Dict, Any, Optional, Tuple, Callable, Union
+from typing import Dict, Any, Optional, Tuple, Callable, Union, cast
 
-from specifyweb.specify.datamodel import datamodel, Table, Relationship
+from specifyweb.specify.datamodel import datamodel, Table
 from specifyweb.specify.load_datamodel import DoesNotExistError
 from specifyweb.specify import models
-from specifyweb.specify.uiformatters import get_uiformatter
+from specifyweb.specify.uiformatters import get_uiformatter, get_catalognumber_format, UIFormatter
 from specifyweb.stored_queries.format import get_date_format
 
 from .uploadable import Uploadable, ScopedUploadable
-from .upload_table import UploadTable, DeferredScopeUploadTable, ScopedUploadTable, OneToOneTable, ScopedOneToOneTable
+from .upload_table import UploadTable, DeferredScopeUploadTable, ScopedUploadTable, ScopedOneToOneTable
 from .tomany import ToManyRecord, ScopedToManyRecord
 from .treerecord import TreeRank, TreeRankRecord, TreeRecord, ScopedTreeRecord
-from .column_options import ColumnOptions, ExtendedColumnOptions
+from .column_options import ColumnOptions, ExtendedColumnOptions, DeferredUIFormatter
 
 """ There are cases in which the scoping of records should be dependent on another record/column in a WorkBench dataset.
 
@@ -85,7 +85,8 @@ def adjust_to_ones(u: ScopedUploadable, f: str) -> ScopedUploadable:
     return adjust_to_ones
 
 
-def extend_columnoptions(colopts: ColumnOptions, collection, tablename: str, fieldname: str) -> ExtendedColumnOptions:
+def extend_columnoptions(colopts: ColumnOptions, collection, tablename: str, fieldname: str, _toOne: Optional[Dict[str, Uploadable]] = None) -> ExtendedColumnOptions:
+    toOne = {} if _toOne is None else _toOne
     schema_items = models.Splocalecontaineritem.objects.filter(
         container__discipline=collection.discipline,
         container__schematype=0,
@@ -109,11 +110,33 @@ def extend_columnoptions(colopts: ColumnOptions, collection, tablename: str, fie
         nullAllowed=colopts.nullAllowed,
         default=colopts.default,
         schemaitem=schemaitem,
-        uiformatter=get_uiformatter(collection, tablename, fieldname),
+        uiformatter=get_or_defer_formatter(collection, tablename, fieldname, toOne),
         picklist=picklist,
         dateformat=get_date_format(),
     )
 
+def get_or_defer_formatter(collection, tablename: str, fieldname: str, _toOne: Dict[str, Uploadable]) -> Union[None, UIFormatter, DeferredUIFormatter]:
+    """ The CollectionObject -> catalogNumber format can be determined by the 
+    CollectionObjectType -> catalogNumberFormatName for the CollectionObject
+
+    Similarly to PrepType, CollectionObjectType in the WorkBench is resolvable 
+    by the 'name' field
+
+    See https://github.com/specify/specify7/issues/5473 
+    """
+    toOne = {key.lower():value for key, value in _toOne.items()}
+    if tablename.lower() == 'collectionobject' and fieldname.lower() == 'catalognumber' and 'collectionobjecttype' in toOne.keys():
+        uploadTable = toOne['collectionobjecttype']
+
+        wb_col = cast(UploadTable, uploadTable).wbcols.get('name', None) if hasattr(uploadTable, 'wbcols') else None
+        optional_col_name = None if wb_col is None else wb_col.column
+        if optional_col_name is not None: 
+            col_name = cast(str, optional_col_name)
+            formats: Dict[str, Optional[UIFormatter]] = {cot.name: get_catalognumber_format(collection, cot.catalognumberformatname, None) for cot in collection.cotypes.all()}
+            return lambda row: formats.get(row[col_name], get_uiformatter(collection, tablename, fieldname))
+
+    return get_uiformatter(collection, tablename, fieldname)
+
 def apply_scoping_to_uploadtable(ut: Union[UploadTable, DeferredScopeUploadTable], collection) -> ScopedUploadTable:
     table = datamodel.get_table_strict(ut.name)
 
@@ -125,7 +148,7 @@ def apply_scoping_to_uploadtable(ut: Union[UploadTable, DeferredScopeUploadTable
 
     return ScopedUploadTable(
         name=ut.name,
-        wbcols={f: extend_columnoptions(colopts, collection, table.name, f) for f, colopts in ut.wbcols.items()},
+        wbcols={f: extend_columnoptions(colopts, collection, table.name, f, ut.toOne) for f, colopts in ut.wbcols.items()},
         static=static_adjustments(table, ut.wbcols, ut.static),
         toOne={f: adjust_to_ones(u.apply_scoping(collection), f) for f, u in ut.toOne.items()},
         toMany={f: [set_order_number(i, r.apply_scoping(collection)) for i, r in enumerate(rs)] for f, rs in ut.toMany.items()},
@@ -177,7 +200,7 @@ def apply_scoping_to_tomanyrecord(tmr: ToManyRecord, collection) -> ScopedToMany
 
     return ScopedToManyRecord(
         name=tmr.name,
-        wbcols={f: extend_columnoptions(colopts, collection, table.name, f) for f, colopts in tmr.wbcols.items()},
+        wbcols={f: extend_columnoptions(colopts, collection, table.name, f, tmr.toOne) for f, colopts in tmr.wbcols.items()},
         static=static_adjustments(table, tmr.wbcols, tmr.static),
         toOne={f: adjust_to_ones(u.apply_scoping(collection), f) for f, u in tmr.toOne.items()},
         scopingAttrs=scoping_relationships(collection, table),