update committing apps

dsms/__init__.py

@@ -1,9 +1,10 @@
 """DSMS top module"""
 
+from dsms.apps import App
 from dsms.core.configuration import Configuration
 from dsms.core.context import Context
 from dsms.core.dsms import DSMS
 from dsms.knowledge.kitem import KItem
 from dsms.knowledge.ktype import KType
 
-__all__ = ["DSMS", "Configuration", "Context", "KItem", "KType"]
+__all__ = ["DSMS", "Configuration", "Context", "KItem", "KType", "App"]

dsms/apps/apps.py

@@ -1,34 +1,140 @@
 """DSMS apps models"""
 
+import logging
 import urllib.parse
+from typing import TYPE_CHECKING, Optional
 
-from pydantic import BaseModel, Field
+from pydantic import (  # isort:skip
+    BaseModel,
+    ConfigDict,
+    Field,
+    field_validator,
+    model_validator,
+)
 
-from dsms.core.utils import _perform_request
+from dsms.apps.utils import (  # isort:skip
+    _get_app_specification,
+    _get_available_apps,
+)
+
+
+from dsms.core.logging import handler  # isort:skip
+
+
+logger = logging.getLogger(__name__)
+logger.addHandler(handler)
+logger.propagate = False
+
+if TYPE_CHECKING:
+    from typing import Any
+
+    from dsms import DSMS, Context
 
 
 class App(BaseModel):
     """KItem app list"""
 
-    filename: str = Field(
-        ..., description="File name of the notebook in the DSMS."
+    filename: Optional[str] = Field(
+        "", description="File name of the app in the DSMS."
     )
-    basename: str = Field(
-        ..., description="Base name of the notebook in the DSMS."
+    basename: str = Field(..., description="Base name of the app in the DSMS.")
+    folder: Optional[str] = Field(
+        "", description="Directory of the app in the DSMS."
     )
-    folder: str = Field(
-        ..., description="Directory of the notebook in the DSMS."
+
+    specification: Optional[str] = Field(
+        None,
+        description="File path for content of YAML Specification of the app",
+    )
+
+    model_config = ConfigDict(
+        extra="forbid",
+        validate_assignment=True,
+        validate_default=True,
+        arbitrary_types_allowed=True,
     )
 
-    def get(self, as_html: bool = False) -> str:
-        """Download the jupyter notebook"""
-        safe_filename = urllib.parse.quote_plus(self.filename)
-        response = _perform_request(
-            f"knowledge/api/apps/{safe_filename}",
-            "get",
-            params={"as_html": as_html},
+    def __init__(self, **kwargs: "Any") -> None:
+        """Initialize the KItem"""
+        from dsms import DSMS
+
+        logger.debug("Initialize KItem with model data: %s", kwargs)
+
+        # set dsms instance if not already done
+        if not self.dsms:
+            self.dsms = DSMS()
+
+        # initialize the app
+        super().__init__(**kwargs)
+
+        # add app to buffer
+        if (
+            not self.in_backend
+            and self.basename not in self.context.buffers.created
+        ):
+            logger.debug(
+                "Marking App with name `%s` as created and updated during App initialization.",
+                self.basename,
+            )
+            self.context.buffers.created.update({self.basename: self})
+            self.context.buffers.updated.update({self.basename: self})
+
+        logger.debug("App initialization successful.")
+
+    def __setattr__(self, name, value) -> None:
+        """Add app to updated-buffer if an attribute is set"""
+        super().__setattr__(name, value)
+        logger.debug(
+            "Setting property with key `%s` on KItem level: %s.", name, value
         )
-        if not response.ok:
-            message = f"Something went wrong downloading app `{self.filename}`: {response.text}"
-            raise RuntimeError(message)
-        return response.text
+        if self.basename not in self.context.buffers.updated:
+            logger.debug(
+                "Setting App with name `%s` as updated during App.__setattr__",
+                self.id,
+            )
+            self.context.buffers.updated.update({self.basename: self})
+
+    @field_validator("basename")
+    @classmethod
+    def validate_basename(cls, value: str) -> str:
+        """Check whether the basename of the app contains invalid characters."""
+        new_value = urllib.parse.quote_plus(value)
+        if not new_value == value:
+            raise ValueError(f"Basename contains invalid characters: {value}")
+
+    @model_validator(mode="after")
+    @classmethod
+    def validate_app(cls, self: "App") -> "App":
+        """Validate app definition."""
+        if self.in_backend and not self.specification:
+            self.specification = _get_app_specification(self.filename)
+            logger.info(
+                "App already exists in backend. Fetched the YAML specification."
+            )
+        return self
+
+    @property
+    def in_backend(self) -> bool:
+        """Checks whether the app already exists."""
+        return self.basename in [
+            app.get("basename") for app in _get_available_apps()
+        ]
+
+    @property
+    def context(cls) -> "Context":
+        """Getter for Context"""
+        from dsms import (  # isort:skip
+            Context,
+        )
+
+        return Context
+
+    @property
+    def dsms(self) -> "DSMS":
+        """DSMS context getter"""
+        return self.context.dsms
+
+    @dsms.setter
+    def dsms(self, value: "DSMS") -> None:
+        """DSMS context setter"""
+        self.context.dsms = value

dsms/apps/utils.py

@@ -1,19 +1,31 @@
 """KItem Apps utils"""
 
+import urllib.parse
 from typing import TYPE_CHECKING
 
-from dsms.apps.apps import App
 from dsms.core.utils import _perform_request
 
 if TYPE_CHECKING:
-    from typing import List
+    from typing import Any, Dict, List
 
 
-def _get_available_apps() -> "List[App]":
+def _get_available_apps() -> "List[Dict[str, Any]]":
     """Get available KItem app."""
     response = _perform_request("api/knowledge/apps", "get")
     if not response.ok:
         message = f"""Something went wrong fetching the available app
         list in the DSMS: {response.text}"""
         raise RuntimeError(message)
-    return [App(**app) for app in response.json()]
+    return response.json()
+
+
+def _get_app_specification(appname) -> str:
+    safe_filename = urllib.parse.quote_plus(appname)
+    response = _perform_request(
+        f"knowledge/api/apps/{safe_filename}",
+        "get",
+    )
+    if not response.ok:
+        message = f"Something went wrong downloading app `{appname}`: {response.text}"
+        raise RuntimeError(message)
+    return response.text

dsms/core/dsms.py

@@ -182,7 +182,9 @@ def kitems(cls) -> "List[KItem]":
     @property
     def apps(cls) -> "List[App]":
         """Return available KItem apps in the DSMS"""
-        return _get_available_apps()
+        from dsms.apps import App
+
+        return [App(**app) for app in _get_available_apps()]
 
     @property
     def buffers(cls) -> "Buffers":

dsms/knowledge/utils.py

@@ -2,6 +2,7 @@
 import base64
 import io
 import logging
+import os
 import re
 import warnings
 from enum import Enum
@@ -33,6 +34,7 @@
 from dsms.knowledge.search import SearchResult  # isort:skip
 
 if TYPE_CHECKING:
+    from dsms.apps import App
     from dsms.core.context import Buffers
     from dsms.knowledge import KItem, KType
 
@@ -492,55 +494,99 @@ def _commit(buffers: "Buffers") -> None:
     logger.debug("Committing successful, clearing buffers.")
 
 
-def _commit_created(buffer: "Dict[str, KItem]") -> dict:
+def _commit_created(buffer: "Dict[str, Union[KItem, KType, App]]") -> dict:
     """Commit the buffer for the `created` buffers"""
-    for kitem in buffer.values():
-        _create_new_kitem(kitem)
+    from dsms import App, KItem, KType
+
+    for obj in buffer.values():
+        if isinstance(obj, KItem):
+            _create_new_kitem(obj)
+        elif isinstance(obj, App):
+            _create_or_update_app(obj)
+        elif isinstance(obj, KType):
+            raise NotImplementedError(
+                "Committing of KTypes not implemented yet."
+            )
+        else:
+            raise TypeError(
+                f"Object `{obj}` of type {type(obj)} cannot be committed."
+            )
 
 
-def _commit_updated(buffer: "Dict[str, KItem]") -> None:
+def _commit_updated(buffer: "Dict[str, Union[KItem, App, KType]]") -> None:
     """Commit the buffer for the `updated` buffers"""
-    for new_kitem in buffer.values():
-        old_kitem = _get_kitem(new_kitem.id, as_json=True)
+    from dsms import App, KItem, KType
+
+    for obj in buffer.values():
+        if isinstance(obj, KItem):
+            _commit_updated_kitem(obj)
+        elif isinstance(obj, App):
+            _create_or_update_app(obj, overwrite=True)
+        elif isinstance(obj, KType):
+            raise NotImplementedError(
+                "Committing of KTypes not implemented yet."
+            )
+        else:
+            raise TypeError(
+                f"Object `{obj}` of type {type(obj)} cannot be committed."
+            )
+
+
+def _commit_updated_kitem(new_kitem: "KItem") -> None:
+    """Commit the updated KItems"""
+    old_kitem = _get_kitem(new_kitem.id, as_json=True)
+    logger.debug(
+        "Fetched data from old KItem with id `%s`: %s",
+        new_kitem.id,
+        old_kitem,
+    )
+    if old_kitem:
+        if isinstance(new_kitem.dataframe, pd.DataFrame):
+            logger.debug(
+                "New KItem data has `pd.DataFrame`. Will push as dataframe."
+            )
+            _update_dataframe(new_kitem.id, new_kitem.dataframe)
+            new_kitem.dataframe = _inspect_dataframe(new_kitem.id)
+        elif isinstance(
+            new_kitem.dataframe, type(None)
+        ) and _inspect_dataframe(new_kitem.id):
+            _delete_dataframe(new_kitem.id)
+        _update_kitem(new_kitem, old_kitem)
+        _update_attachments(new_kitem, old_kitem)
+        if new_kitem.avatar.file or new_kitem.avatar.include_qr:
+            _commit_avatar(new_kitem)
+        new_kitem.in_backend = True
         logger.debug(
-            "Fetched data from old KItem with id `%s`: %s",
-            new_kitem.id,
-            old_kitem,
+            "Fetching updated KItem from remote backend: %s", new_kitem.id
         )
-        if old_kitem:
-            if isinstance(new_kitem.dataframe, pd.DataFrame):
-                logger.debug(
-                    "New KItem data has `pd.DataFrame`. Will push as dataframe."
-                )
-                _update_dataframe(new_kitem.id, new_kitem.dataframe)
-                new_kitem.dataframe = _inspect_dataframe(new_kitem.id)
-            elif isinstance(
-                new_kitem.dataframe, type(None)
-            ) and _inspect_dataframe(new_kitem.id):
-                _delete_dataframe(new_kitem.id)
-            _update_kitem(new_kitem, old_kitem)
-            _update_attachments(new_kitem, old_kitem)
-            if new_kitem.avatar.file or new_kitem.avatar.include_qr:
-                _commit_avatar(new_kitem)
-            new_kitem.in_backend = True
+        for key, value in _get_kitem(new_kitem.id, as_json=True).items():
             logger.debug(
-                "Fetching updated KItem from remote backend: %s", new_kitem.id
+                "Set updated property `%s` for KItem with id `%s` after commiting: %s",
+                key,
+                new_kitem.id,
+                value,
             )
-            for key, value in _get_kitem(new_kitem.id, as_json=True).items():
-                logger.debug(
-                    "Set updated property `%s` for KItem with id `%s` after commiting: %s",
-                    key,
-                    new_kitem.id,
-                    value,
-                )
-                setattr(new_kitem, key, value)
+            setattr(new_kitem, key, value)
 
 
-def _commit_deleted(buffer: "Dict[str, KItem]") -> None:
+def _commit_deleted(buffer: "Dict[str, Union[KItem, KType, App]]") -> None:
     """Commit the buffer for the `deleted` buffers"""
-    for kitem in buffer.values():
-        _delete_dataframe(kitem.id)
-        _delete_kitem(kitem)
+    from dsms import App, KItem, KType
+
+    for obj in buffer.values():
+        if isinstance(obj, KItem):
+            _delete_dataframe(obj.id)
+            _delete_kitem(obj)
+        elif isinstance(obj, App):
+            raise NotImplementedError("Deletion of Apps not implemented yet.")
+        elif isinstance(obj, KType):
+            raise NotImplementedError(
+                "Deletion of KTypes not implemented yet."
+            )
+        else:
+            raise TypeError(
+                f"Object `{obj}` of type {type(obj)} cannot be committed or deleted."
+            )
 
 
 def _split_iri(iri: str) -> List[str]:
@@ -728,3 +774,21 @@ def _get_avatar(kitem: "KItem") -> Image.Image:
     response = _perform_request(f"api/knowledge/avatar/{kitem.id}", "get")
     buffer = io.BytesIO(response.content)
     return Image.open(buffer)
+
+
+def _create_or_update_app(app: "App", overwrite=False) -> None:
+    if os.path.exists(app.specification):
+        with open(app.specification, mode="rb") as file:
+            upload_file = {"dataFile": file}
+    else:
+        upload_file = {"dataFile": io.StringIO(app.specification)}
+    response = _perform_request(
+        f"knowledge/api/apps/argo/{app.basename}",
+        "put",
+        files=upload_file,
+        params={"overwrite": overwrite},
+    )
+    if not response.ok:
+        message = f"Something went wrong uploading app with name `{app.basename}`: {response.text}"
+        raise RuntimeError(message)
+    return response.text