Installing and Using a Sequence-based Naming Format for Entities, Experiments, and other models

Server side extensions use the lab7 modules which is a Python interpreter separate from the Pipeline interpreter and client modules esp and espclient.  A Python function decorated with the from lab7.extensions import invokable decorator creates a custom endpoint at the path /api/invoke/my_function_name.  This can be a possible entrypoint for setting up custom naming logic, including automated naming and bulk creation of other models - particularly using sequence based naming.

The main function of interest for using a sequence is from lab7.sample.api import next_sequence_value.  The next_sequence_value function will consume and return the next integer in the sequence.  This action is immediately persisted to the database.

This could be used to create an endpoint to create a container with the next name in a sequence.  This could be as simple as creating one's own calculation of a name string using the next_sequence_value or a more Configuration driven example like below:

from lab7.extensions import invokable
from sqlalchemy.orm import Session
from typing import Union


@invokable(session="session")
def create_container_autoname(
    agent: str,
    session: Session,
    def_name: str,
    lab7_id_seq: Union[str, None] = None,
    return_dict: bool = True
    **values
):
    """
    Uses the next sequential name to create a new Container.  Requires an update to the "esp" Configuration
    to add a new block to the id_sequences section. "experiment", "entity", and "item" are already part of
    the standard configuration.  Add a new "container" block with the naming formats as you would for
    other model types (see example).  As this is a custom backend feature, there is no way in L7|Master
    to constrain the name format to a subset of Container Types (which is normally required for standard
    autonaming) in this implementation.  Standard UI creation of containers in the Locations app will also
    
    Arguments
    agent: The current User object.
    session: The currently bound SQLAlchemy ESPSession.
    def_name: the name of the container type targeted for creation.  The most recent definition will be used.
    lab7_id_seq: name of naming format to use, otherwise uses default for container type
    **values: keyword argument capture used in general model creation. The kwargs are filtered
    based on keyword match in the function.  Currently accepted keywords include name, desc,
    tags, items, barcode, barcode_type, resource_vals, view_template, augment, status. See REST API for contianer
    for more details
    return_dict: if set to True, will return the dictionary format of the container representation.  Otherwise
    it will return the class instance of the Container ORM model which can be more efficient in more complex
    functionality (i,e. sub-function of a larger invokable)
    
    Returns
    Dictionary or model representation of lab7.container.models.Container - a SQL Alchemy ORM model for a L7|ESP Container, 
    a subclass of Resource.
    
    Configuration
    Container name format configuration block for "esp" Configuration:
    "container": [
      {
        "name": "96-Well Sequence",
        "format": "96W{sequence_number:06}",
        "sequence": "96w_sequence"
      }
    Note that the 96w_sequence integer should be initialized (usually to 1) in the sequences configuration block 
    for the above example.  Left padding shown here to 6 digits, but could be increased depending on the lab's
    throughput.
    """
    from lab7.container.api import query_container_types, create_container
    from lab7.sample.api import next_sequence_value, ListFormatter
    from lab7.utils import configs

    name_context = {}

	# Searches by the Container Type fixed ID and display name for convenience
    # Ultimately we need the Container Type Definition
    # Query existing by fixed_id
    container_types = None
    container_types = query_container_types(
        filters={"barcode": def_name},
        return_dict=False,
        ignore_deleted=True, # default, assume archived models are not intended for use
        session=session,
        agent=agent,
    )
    # If no containers were found by 'fixed_id', try then by name
    if not container_types:
        container_types = query_container_types(
            filters={"name": def_name},
            return_dict=False,
            ignore_deleted=True,
            session=session,
            agent=agent,
        )
	
	# Use the most recent Container Type Definition for the matched Container Type
    container_type = container_types[0]
    current_container_def = container_type.head

	# Use Configuration, similar to Entities, Experiments, and Items to drive naming
    if lab7_id_seq is None:
        lab7_id_seq = current_container_def.meta["lab7_id_sequence"][0]
    tenant_config = configs.get()["esp"]["config"]
    container_id_sequences = tenant_config["id_sequences"].get("container", [])

    seq_name = None
    id_fmt = None
    if lab7_id_seq is not None:
        for container_id_seq in container_id_sequences:
            if container_id_seq["name"] == lab7_id_seq:
                seq_name = container_id_seq["sequence"]
                id_fmt = container_id_seq["format"]
                break
    else:
        raise ValueError("Cannot create container without an ID sequence")

    if not (seq_name and id_fmt):
        raise ValueError(
            "There are no configured ID sequence formats available for this Container. Please assign a Naming Format or provide lab7_id_seq argument with the format name."
        )

    # This function immediately obtains the next integer in a naming sequence
    sequence_value = next_sequence_value(seq_name, session)

    # This block converts the formatting string into a name, by plugging in the text placeholders
    name_context.update(
        dict(
            date=session.info["utc_timestamp"].strftime("%Y-%m-%d"),
            time=session.info["utc_timestamp"].strftime("%H%M%S"),
            datetime=session.info["utc_timestamp"],
            item_type=current_container_def.name,
            sequence_number=sequence_value,
        )
    )

    formatter = ListFormatter()
    try:
        name = formatter.format(id_fmt, **name_context)
    except KeyError:
        raise ValueError(f"Sample could not be created due to an invalid name format, possibly due to an invalid \
        naming variable.  Please update the naming format in the esp Configuration and try again. '{lab7_id_seq}' \
        Format:  {id_fmt}, Name Context Variables: {name_context}")
    

    other_opts = {
        arg: values.get(arg)
        for arg in [
            "name",
            "desc",
            "tags",
            "items",
            "barcode",
            "barcode_type",
            "resource_vals",
            "view_template",
            "augment",
            "status",
        ]
        if values.get(arg, None) is not None
    }

    return create_container(
        current_container_def, session=session, agent=agent, name=name, **other_opts
    )

A custom function might also be used to consume the next N names from a sequence.  This should be slightly more efficient as it represents less roundtrips between the application and the database.

def next_sequence_value_bulk(name, count=1, session=None, agent=None):
    """
    Get the next N count integers in the sequence.  Persists sequence usage immediately, even in case of failure.
    """
    from sqlalchemy.sql import text

    if name in configs.get()["esp"]["config"]["sequences"]:
        query = """
            UPDATE esp_sequences SET seq_number = seq_number + :count
            WHERE seq_name = :seq_name
            RETURNING seq_number - :count as the_seq
        """
        bind_data = {"seq_name": name, "count": count}
        rows = session.execute(text(query), bind_data)
        row = rows.first()
        return row["the_seq"] if count == 1 else [row["the_seq"] + i for i in range(count)]
    else:
        raise ValueError(_("Unknown sequence: {}").format(name))