Add coverage for namesets

liminal/base/base_operation.py

@@ -19,9 +19,10 @@
     12. UnarchiveField
     13. UpdateField
     14. ArchiveField
-    15. ReorderFields
-    16. ArchiveSchema
-    17. ArchiveDropdown
+    15. UpdateEntitySchemaNameTemplate
+    16. ReorderFields
+    17. ArchiveSchema
+    18. ArchiveDropdown
 """
 
 

liminal/base/name_template_parts.py

@@ -0,0 +1,96 @@
+from typing import Any, ClassVar
+
+from pydantic import BaseModel, ConfigDict, field_validator
+
+from liminal.enums.name_template_part_type import NameTemplatePartType
+
+
+class NameTemplatePart(BaseModel):
+    """Base class for all name template parts. These are put together in a list (where order matters) to form a name template.
+
+    Parameters
+    ----------
+    component_type : NameTemplatePartType
+        The type of the component. One of the values in the NameTemplatePartType enum.
+
+    """
+
+    component_type: ClassVar[NameTemplatePartType]
+
+    _type_map: ClassVar[dict[NameTemplatePartType, type["NameTemplatePart"]]] = {}
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    def __init_subclass__(cls, **kwargs: Any) -> None:
+        super().__init_subclass__(**kwargs)
+        cls._type_map[cls.component_type] = cls
+
+    @classmethod
+    def resolve_type(cls, type: NameTemplatePartType) -> type["NameTemplatePart"]:
+        if type not in cls._type_map:
+            raise ValueError(f"Invalid name template part type: {type}")
+        return cls._type_map[type]
+
+
+class SeparatorPart(NameTemplatePart):
+    component_type: ClassVar[NameTemplatePartType] = NameTemplatePartType.SEPARATOR
+    value: str
+
+    @field_validator("value")
+    def validate_value(cls, v: str) -> str:
+        if not v:
+            raise ValueError("value cannot be empty")
+        return v
+
+
+class TextPart(NameTemplatePart):
+    component_type: ClassVar[NameTemplatePartType] = NameTemplatePartType.TEXT
+    value: str
+
+    @field_validator("value")
+    def validate_value(cls, v: str) -> str:
+        if not v:
+            raise ValueError("value cannot be empty")
+        return v
+
+
+class CreationYearPart(NameTemplatePart):
+    component_type: ClassVar[NameTemplatePartType] = NameTemplatePartType.CREATION_YEAR
+
+
+class CreationDatePart(NameTemplatePart):
+    component_type: ClassVar[NameTemplatePartType] = NameTemplatePartType.CREATION_DATE
+
+
+class FieldPart(NameTemplatePart):
+    component_type: ClassVar[NameTemplatePartType] = NameTemplatePartType.FIELD
+    wh_field_name: str
+
+
+class ParentLotNumberPart(NameTemplatePart):
+    component_type: ClassVar[NameTemplatePartType] = (
+        NameTemplatePartType.CHILD_ENTITY_LOT_NUMBER
+    )
+    wh_field_name: str
+
+
+class RegistryIdentifierNumberPart(NameTemplatePart):
+    component_type: ClassVar[NameTemplatePartType] = (
+        NameTemplatePartType.REGISTRY_IDENTIFIER_NUMBER
+    )
+
+
+class ProjectPart(NameTemplatePart):
+    component_type: ClassVar[NameTemplatePartType] = NameTemplatePartType.PROJECT
+
+
+NameTemplateParts = (
+    SeparatorPart
+    | TextPart
+    | CreationYearPart
+    | CreationDatePart
+    | FieldPart
+    | RegistryIdentifierNumberPart
+    | ProjectPart
+    | ParentLotNumberPart
+)

liminal/base/properties/base_name_template.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict
+
+from liminal.base.name_template_parts import NameTemplateParts
+
+
+class BaseNameTemplate(BaseModel):
+    """
+    This class is the generic class for defining the name template.
+    It is used to create a diff between the old and new name template.
+
+    Parameters
+    ----------
+    parts : list[NameTemplatePart] | None
+        The list of name template parts that make up the name template (order matters).
+    order_name_parts_by_sequence : bool | None
+        Whether to order the name parts by sequence. This can only be set to True for sequence enity types. If one or many part link fields are included in the name template,
+        list parts in the order they appear on the sequence map, sorted by start position and then end position.
+    """
+
+    parts: list[NameTemplateParts] | None = None
+    order_name_parts_by_sequence: bool | None = None
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    def merge(self, new_props: BaseNameTemplate) -> dict[str, Any]:
+        """Creates a diff between the current name template and the new name template.
+        Sets value to None if the values are equal, otherwise sets the value to the new value.
+
+        Parameters
+        ----------
+        new_props : BaseNameTemplate
+            The new name template.
+
+        Returns
+        -------
+        dict[str, Any]
+            A dictionary of the differences between the old and new name template.
+        """
+        diff = {}
+        for field_name in self.model_fields:
+            new_val = getattr(new_props, field_name)
+            if getattr(self, field_name) != new_val:
+                diff[field_name] = new_val
+        return diff
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, BaseNameTemplate):
+            return False
+        return self.model_dump() == other.model_dump()
+
+    def __str__(self) -> str:
+        parts_str = (
+            f"parts=[{', '.join(repr(part) for part in self.parts)}]"
+            if self.parts is not None
+            else None
+        )
+        order_name_parts_by_sequence_str = (
+            f"order_name_parts_by_sequence={self.order_name_parts_by_sequence}"
+            if self.order_name_parts_by_sequence is not None
+            else None
+        )
+        return ", ".join(filter(None, [parts_str, order_name_parts_by_sequence_str]))
+
+    def __repr__(self) -> str:
+        """Generates a string representation of the class so that it can be executed."""
+        model_dump = self.model_dump(exclude_defaults=True, exclude_unset=True)
+        props = []
+        if "parts" in model_dump:
+            parts_repr = (
+                f"[{', '.join(repr(part) for part in self.parts)}]"
+                if self.parts
+                else "[]"
+            )
+            props.append(f"parts={parts_repr}")
+        if "order_name_parts_by_sequence" in model_dump:
+            props.append(
+                f"order_name_parts_by_sequence={self.order_name_parts_by_sequence}"
+            )
+        return f"{self.__class__.__name__}({', '.join(props)})"

liminal/dropdowns/operations.py

@@ -76,7 +76,7 @@ def describe(self) -> str:
 
 
 class ArchiveDropdown(BaseOperation):
-    order: ClassVar[int] = 170
+    order: ClassVar[int] = 180
 
     def __init__(self, dropdown_name: str) -> None:
         self.dropdown_name = dropdown_name

liminal/entity_schemas/api.py

@@ -75,3 +75,21 @@ def update_tag_schema(
         return await_queued_response(
             queued_response.json()["status_url"], benchling_service
         )
+
+
+def set_tag_schema_name_template(
+    benchling_service: BenchlingService, entity_schema_id: str, payload: dict[str, Any]
+) -> dict[str, Any]:
+    """
+    Update the tag schema name template. Must be in a separate endpoint compared to update_tag_schema.
+    """
+    with requests.Session() as session:
+        response = session.post(
+            f"https://{benchling_service.benchling_tenant}.benchling.com/1/api/tag-schemas/{entity_schema_id}/actions/set-name-template",
+            data=json.dumps(payload),
+            headers=benchling_service.custom_post_headers,
+            cookies=benchling_service.custom_post_cookies,
+        )
+        if not response.ok:
+            raise Exception("Failed to set tag schema name template:", response.content)
+        return response.json()

liminal/entity_schemas/compare.py

@@ -1,5 +1,6 @@
 from liminal.base.compare_operation import CompareOperation
 from liminal.base.properties.base_field_properties import BaseFieldProperties
+from liminal.base.properties.base_name_template import BaseNameTemplate
 from liminal.base.properties.base_schema_properties import BaseSchemaProperties
 from liminal.connection import BenchlingService
 from liminal.entity_schemas.operations import (
@@ -12,6 +13,7 @@
     UnarchiveEntitySchemaField,
     UpdateEntitySchema,
     UpdateEntitySchemaField,
+    UpdateEntitySchemaNameTemplate,
 )
 from liminal.entity_schemas.utils import get_converted_tag_schemas
 from liminal.orm.base_model import BaseModel
@@ -47,12 +49,12 @@ def compare_entity_schemas(
         if not m.__schema_properties__._archived
     ]
     archived_benchling_schema_wh_names = [
-        s.warehouse_name for s, _ in benchling_schemas if s._archived is True
+        s.warehouse_name for s, _, _ in benchling_schemas if s._archived is True
     ]
     # Running list of schema names from benchling. As each model is checked, remove the schema name from this list.
     # This is used at the end to check if there are any schemas left (schemas that exist in benchling but not in code) and archive them if they are.
     running_benchling_schema_names = list(
-        [s.warehouse_name for s, _ in benchling_schemas]
+        [s.warehouse_name for s, _, _ in benchling_schemas]
     )
     # Iterate through each benchling model defined in code.
     for model in models:
@@ -64,12 +66,14 @@ def compare_entity_schemas(
         model.validate_model()
         # if the model table_name is found in the benchling schemas, check for changes...
         if (model_wh_name := model.__schema_properties__.warehouse_name) in [
-            s.warehouse_name for s, _ in benchling_schemas
+            s.warehouse_name for s, _, _ in benchling_schemas
         ]:
-            benchling_schema_props, benchling_schema_fields = next(
-                (s, lof)
-                for s, lof in benchling_schemas
-                if s.warehouse_name == model_wh_name
+            benchling_schema_props, benchling_name_template, benchling_schema_fields = (
+                next(
+                    (s, nt, lof)
+                    for s, nt, lof in benchling_schemas
+                    if s.warehouse_name == model_wh_name
+                )
             )
             archived_benchling_schema_fields = {
                 k: v for k, v in benchling_schema_fields.items() if v._archived is True
@@ -237,6 +241,23 @@ def compare_entity_schemas(
                         ),
                     ),
                 )
+            if benchling_name_template != model.__name_template__:
+                ops.append(
+                    CompareOperation(
+                        op=UpdateEntitySchemaNameTemplate(
+                            model.__schema_properties__.warehouse_name,
+                            BaseNameTemplate(
+                                **benchling_name_template.merge(model.__name_template__)
+                            ),
+                        ),
+                        reverse_op=UpdateEntitySchemaNameTemplate(
+                            model.__schema_properties__.warehouse_name,
+                            BaseNameTemplate(
+                                **model.__name_template__.merge(benchling_name_template)
+                            ),
+                        ),
+                    )
+                )
         # If the model is not found as the benchling schema, Create.
         # Benchling api does not allow for setting a custom warehouse_name,
         # so we need to run another UpdateEntitySchema to set the warehouse_name if it is different from the snakecase version of the model name.
@@ -276,6 +297,30 @@ def compare_entity_schemas(
                         ),
                     )
                 )
+            benchling_given_name_template = BaseNameTemplate(
+                parts=[], order_name_parts_by_sequence=False
+            )
+            if benchling_name_template != model.__name_template__:
+                ops.append(
+                    CompareOperation(
+                        op=UpdateEntitySchemaNameTemplate(
+                            model.__schema_properties__.warehouse_name,
+                            BaseNameTemplate(
+                                **benchling_given_name_template.merge(
+                                    model.__name_template__
+                                )
+                            ),
+                        ),
+                        reverse_op=UpdateEntitySchemaNameTemplate(
+                            model.__schema_properties__.warehouse_name,
+                            BaseNameTemplate(
+                                **benchling_given_name_template.merge(
+                                    model.__name_template__
+                                )
+                            ),
+                        ),
+                    )
+                )
 
         model_operations[model.__schema_properties__.warehouse_name] = ops
         running_benchling_schema_names = [

liminal/entity_schemas/generate_files.py

@@ -7,6 +7,7 @@
 from liminal.entity_schemas.utils import get_converted_tag_schemas
 from liminal.enums import BenchlingEntityType, BenchlingFieldType
 from liminal.mappers import convert_benchling_type_to_python_type
+from liminal.orm.name_template import NameTemplate
 from liminal.utils import pascalize, to_snake_case
 
 
@@ -62,13 +63,13 @@ def generate_all_entity_schema_files(
         for dropdown_name in benchling_dropdowns.keys()
     }
     wh_name_to_classname: dict[str, str] = {
-        sp.warehouse_name: pascalize(sp.name) for sp, _ in models
+        sp.warehouse_name: pascalize(sp.name) for sp, _, _ in models
     }
 
-    for schema_properties, columns in models:
+    for schema_properties, name_template, columns in models:
         classname = pascalize(schema_properties.name)
 
-    for schema_properties, columns in models:
+    for schema_properties, name_template, columns in models:
         classname = pascalize(schema_properties.name)
         filename = to_snake_case(schema_properties.name) + ".py"
         columns = {key: columns[key] for key in columns}
@@ -132,6 +133,12 @@ def generate_all_entity_schema_files(
             init_strings.append(f"{tab}self.{col_name} = {col_name}")
         if len(dropdowns) > 0:
             import_strings.append(f"from ...dropdowns import {', '.join(dropdowns)}")
+        if name_template != NameTemplate():
+            import_strings.append("from liminal.orm.name_template import NameTemplate")
+            parts_imports = [
+                f"from liminal.base.name_template_parts import {', '.join(set([part.__class__.__name__ for part in name_template.parts]))}"
+            ]
+            import_strings.extend(parts_imports)
         for col_name, col in columns.items():
             if col.dropdown_link:
                 init_strings.append(
@@ -153,6 +160,7 @@ def get_validators(self) -> list[BenchlingValidator]:
 
 class {classname}(BaseModel, {get_entity_mixin(schema_properties.entity_type)}):
     __schema_properties__ = {schema_properties.__repr__()}
+    {f"__name_template__ = {name_template.__repr__()}" if name_template != NameTemplate() else ""}
 
 {columns_string}