partially working id tracking

pyproject.toml

@@ -16,10 +16,9 @@ classifiers = [
 description = "IOC Builder for EPICS and Kubernetes"
 dependencies = [
     "typing-extensions",
-    "apischema>=0.15",
+    "pydantic",
     "typer",
     "ruamel.yaml",
-    "jsonschema",
     "jinja2",
     "typing-extensions;python_version<'3.8'",
 ] # Add project dependencies here, e.g. ["click", "numpy"]

src/ibek/__main__.py

@@ -1,12 +1,17 @@
+import json
 from pathlib import Path
 from typing import List, Optional
 
 import typer
 from ruamel.yaml import YAML
 
 from ._version import __version__
-from .gen_scripts import create_boot_script, create_db_script, ioc_deserialize
-from .ioc import make_entity_models, make_ioc_model
+from .gen_scripts import (
+    create_boot_script,
+    create_db_script,
+    ioc_create_model,
+    ioc_deserialize,
+)
 from .support import Support
 
 cli = typer.Typer()
@@ -46,27 +51,13 @@ def ioc_schema(
         ..., help="The filepath to a support module definition file"
     ),
     output: Path = typer.Argument(..., help="The filename to write the schema to"),
-    no_schema: bool = typer.Option(False, help="disable schema checking"),
 ):
     """
     Create a json schema from a <support_module>.ibek.support.yaml file
     """
 
-    entity_classes = []
-
-    for definition in definitions:
-        support_dict = YAML(typ="safe").load(definition)
-        if not no_schema:
-            # Verify the schema of the support module definition file
-            Support.model_validate(support_dict)
-
-        # deserialize the support module definition file
-        support = Support(**support_dict)
-        # make Entity classes described in the support module definition file
-        entity_classes += make_entity_models(support)
-
-    # Save the schema for IOC
-    schema = make_ioc_model(entity_classes)
+    ioc_model = ioc_create_model(definitions)
+    schema = json.dumps(ioc_model.model_json_schema(), indent=2)
     output.write_text(schema)
 
 

src/ibek/gen_scripts.py

@@ -4,12 +4,12 @@
 import logging
 import re
 from pathlib import Path
-from typing import List
+from typing import List, Type
 
 from jinja2 import Template
 from ruamel.yaml.main import YAML
 
-from .ioc import IOC, make_entity_classes
+from .ioc import IOC, clear_entity_model_ids, make_entity_models, make_ioc_model
 from .render import Render
 from .support import Support
 
@@ -21,20 +21,44 @@
 url_f = r"file://"
 
 
+def ioc_create_model(definitions: List[Path]) -> Type[IOC]:
+    """
+    Take a list of definitions YAML and create an IOC model from it
+    """
+    entity_models = []
+
+    clear_entity_model_ids()
+    for definition in definitions:
+        support_dict = YAML(typ="safe").load(definition)
+
+        Support.model_validate(support_dict)
+
+        # deserialize the support module definition file
+        support = Support(**support_dict)
+        # make Entity classes described in the support module definition file
+        entity_models += make_entity_models(support)
+
+    # Save the schema for IOC
+    model = make_ioc_model(entity_models)
+
+    return model
+
+
 def ioc_deserialize(ioc_instance_yaml: Path, definition_yaml: List[Path]) -> IOC:
     """
     Takes an ioc instance entities file, list of generic ioc definitions files.
 
-    Returns an in memory object graph of the resulting ioc instance
+    Returns a model of the resulting ioc instance
     """
+    ioc_model = ioc_create_model(definition_yaml)
+
+    ioc_instance = YAML(typ="safe").load(ioc_instance_yaml)
 
-    # Read and load the support module definitions
-    for yaml in definition_yaml:
-        support = Support.deserialize(YAML(typ="safe").load(yaml))
-        make_entity_classes(support)
+    clear_entity_model_ids()
+    ioc_model.model_validate(ioc_instance)
 
-    # Create an IOC instance from it
-    return IOC.deserialize(YAML(typ="safe").load(ioc_instance_yaml))
+    # Create an IOC instance from the entities file and the model
+    return ioc_model.model_construct(ioc_instance)
 
 
 def create_db_script(ioc_instance: IOC) -> str:

src/ibek/globals.py

@@ -1,24 +1,17 @@
 """
 A few global definitions
 """
-from typing import TypeVar
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict
 
-#: A generic Type for use in type hints
-T = TypeVar("T")
-
-
-def desc(description: str):
-    """a description Annotation to add to our Entity derived Types"""
-    return Field(description=description)
+# pydantic model configuration
+model_config = ConfigDict(
+    # arbitrary_types_allowed=True,
+    extra="forbid",
+)
 
 
 class BaseSettings(BaseModel):
     """A Base class for setting Pydantic model configuration"""
 
-    # Pydantic model configuration
-    model_config = ConfigDict(
-        # arbitrary_types_allowed=True,
-        extra="forbid",
-    )
+    model_config = model_config

src/ibek/ioc.py

@@ -5,21 +5,22 @@
 from __future__ import annotations
 
 import builtins
-import json
-from typing import Any, Dict, Sequence, Tuple, Type, Union
+from typing import Any, Dict, Literal, Sequence, Tuple, Type, Union
 
 from jinja2 import Template
-from pydantic import Field, ValidationError, create_model
+from pydantic import Field, create_model, field_validator
 from pydantic.fields import FieldInfo
-from typing_extensions import Literal
 
-from .globals import BaseSettings, model_config
+from .globals import BaseSettings
 from .support import Definition, IdArg, ObjectArg, Support
 from .utils import UTILS
 
 # A base class for applying settings to all serializable classes
 
 
+id_to_entity: Dict[str, Entity] = {}
+
+
 class Entity(BaseSettings):
     """
     A baseclass for all generated Entity classes. Provides the
@@ -33,36 +34,24 @@ class Entity(BaseSettings):
         description="enable or disable this entity instance", default=True
     )
 
-    def __post_init__(self: "Entity"):
-        # If there is an argument which is an id then allow deserialization by that
-        args = self.__definition__.args
-        ids = set(a.name for a in args if isinstance(a, IdArg))
-        assert len(ids) <= 1, f"Multiple id args {list(ids)} defined in {args}"
-        if ids:
-            # A string id, use that
-            inst_id = getattr(self, ids.pop())
-            assert inst_id not in id_to_entity, f"Already got an instance {inst_id}"
-            id_to_entity[inst_id] = self
-
-            # TODO - not working as printing own ID
-            setattr(self, "__str__", inst_id)
+    # @model_validator(mode="before")  # type: ignore
+    def add_ibek_attributes(cls, entity: Dict):
+        """Add attributes used by ibek"""
 
         # add in the global __utils__ object for state sharing
-        self.__utils__ = UTILS
+        entity["__utils__"] = UTILS
 
         # copy 'values' from the definition into the Entity
-        for value in self.__definition__.values:
-            setattr(self, value.name, value.value)
+
+        # if hasattr(entity, "__definition__"):
+        #     entity.update(entity.__definition__.values)
 
         # Jinja expansion of any string args/values in the Entity's attributes
-        for arg, value in self.__dict__.items():
+        for arg, value in entity.items():
             if isinstance(value, str):
                 jinja_template = Template(value)
-                rendered = jinja_template.render(self.__dict__)
-                setattr(self, arg, rendered)
-
-
-id_to_entity: Dict[str, Entity] = {}
+                entity[arg] = jinja_template.render(entity)
+        return entity
 
 
 def make_entity_model(definition: Definition, support: Support) -> Type[Entity]:
@@ -76,50 +65,64 @@ def make_entity_model(definition: Definition, support: Support) -> Type[Entity]:
     """
 
     def add_entity(name, typ, description, default):
-        entities[name] = (typ, FieldInfo(description=description, default=default))
+        entities[name] = (
+            typ,
+            FieldInfo(
+                description=description,
+                default=default,
+            ),
+        )
 
     entities: Dict[str, Tuple[type, Any]] = {}
+    validators: Dict[str, Any] = {}
+
+    # fully qualified name of the Entity class including support module
+    full_name = f"{support.module}.{definition.name}"
 
     # add in each of the arguments as a Field in the Entity
     for arg in definition.args:
-        arg_type: Type
+        full_arg_name = f"{full_name}.{arg.name}"
 
         if isinstance(arg, ObjectArg):
-            pass
 
-            def lookup_instance(id):
+            @field_validator(arg.name)
+            def lookup_instance(cls, id):
                 try:
                     return id_to_entity[id]
                 except KeyError:
-                    raise ValidationError(f"{id} is not in {list(id_to_entity)}")
+                    raise KeyError(f"object id {id} not in {list(id_to_entity)}")
+
+            validators[full_arg_name] = lookup_instance
+            arg_type = str
 
-            # metadata = schema(extra={"vscode_ibek_plugin_type": "type_object"})
-            # metadata = conversion(
-            #     deserialization=Conversion(lookup_instance, str, Entity)
-            # ) | schema(extra={"vscode_ibek_plugin_type": "type_object"})
-            arg_type = Entity
         elif isinstance(arg, IdArg):
+
+            @field_validator(arg.name)
+            def save_instance(cls, id):
+                if id in id_to_entity:
+                    # TODO we are getting multiple registers of same Arg
+                    pass  # raise KeyError(f"id {id} already defined in {list(id_to_entity)}")
+                id_to_entity[id] = cls
+                return id
+
+            validators[full_arg_name] = save_instance
             arg_type = str
-            # metadata = schema(extra={"vscode_ibek_plugin_type": "type_id"})
+
         else:
             # arg.type is str, int, float, etc.
             arg_type = getattr(builtins, arg.type)
 
         default = getattr(arg, "default", None)
         add_entity(arg.name, arg_type, arg.description, default)
 
-    # type is a unique key for each of the entity types we may instantiate
-    full_name = f"{support.module}.{definition.name}"
     typ = Literal[full_name]  # type: ignore
     add_entity("type", typ, "The type of this entity", full_name)
 
-    # entity_enabled controls rendering of the entity without having to delete it
-    add_entity("entity_enabled", bool, "enable or disable entity", True)
-
     entity_cls = create_model(
-        "definitions",
+        full_name.replace(".", "_"),
         **entities,
-        __config__=model_config,
+        __validators__=validators,
+        # __base__=Entity,
     )  # type: ignore
 
     # add a link back to the Definition Object that generated this Entity Class
@@ -136,44 +139,40 @@ def make_entity_models(support: Support):
     set to a Union of all the Entity subclasses created."""
 
     entity_models = []
+    entity_names = []
 
     for definition in support.defs:
         entity_models.append(make_entity_model(definition, support))
+        if definition.name in entity_names:
+            raise ValueError(f"Duplicate entity name {definition.name}")
+        entity_names.append(definition.name)
 
     return entity_models
 
 
-def clear_entity_classes():
-    """Reset the modules namespaces, deserializers and caches of defined Entity
-    subclasses"""
-
-    # TODO: do we need this for Pydantic?
-
-
-def make_ioc_model(entity_classes: Sequence[Type[Entity]]) -> str:
+def make_ioc_model(entity_models: Sequence[Type[Entity]]) -> Type[IOC]:
     class NewIOC(IOC):
-        entities: Sequence[Union[tuple(entity_classes)]] = Field(  # type: ignore
+        entities: Sequence[Union[tuple(entity_models)]] = Field(  # type: ignore
             description="List of entities this IOC instantiates", default=()
         )
 
-    return json.dumps(NewIOC.model_json_schema(), indent=2)
+    return NewIOC
+
+
+def clear_entity_model_ids():
+    """Resets the global id_to_entity dict."""
+    global id_to_entity
+
+    id_to_entity.clear()
 
 
 class IOC(BaseSettings):
     """
-    Used to load an IOC instance entities yaml file into memory.
-
-    This is the base class that is adjusted at runtime by updating the
-    type of its entities attribute to be a union of all of the subclasses of Entity
-    provided by the support module definitions used by the current IOC
+    Used to load an IOC instance entities yaml file into a Pydantic Model.
     """
 
     ioc_name: str = Field(description="Name of IOC instance")
     description: str = Field(description="Description of what the IOC does")
     generic_ioc_image: str = Field(
         description="The generic IOC container image registry URL"
     )
-    # placeholder for the entities attribute - updated at runtime
-    entities: Sequence[Entity] = Field(
-        description="List of entities this IOC instantiates"
-    )

src/ibek/support.py

@@ -29,7 +29,6 @@ class Arg(BaseSettings):
     description: str = Field(
         description="Description of what the argument will be used for"
     )
-    # __discriminator__ = "type"
 
 
 # FloatArg must be defined before StrArg, otherwise we get:
@@ -40,7 +39,7 @@ class Arg(BaseSettings):
 # have a trailing 'f'. It is due to the order of declaration of subclasses of
 # Arg. When StrArg is before FloatArg, apischema attempts to deserialize as a
 # string first. The coercion from str to number requires a trailing f if there
-# is a decimal.
+# is a decimal. TODO is this still an issue with Pydantic?
 class FloatArg(Arg):
     """An argument with a float value"""