Add support for attachments

docs/_templates/scitacean-class-template.rst

@@ -6,6 +6,7 @@
                        "OrigDatablockProxy": ["__init__", "from_download_model"],
                        "PID": ["__init__", "parse"],
                        "ScicatClient": ["from_credentials", "from_token", "without_login"],
+                       "Thumbnail": ["__init__", "load_file", "parse"],
                       } %}
 {% set regular_methods = methods | reject("in", constructors.get(name, []) + ["__init__"]) | list %}
 

docs/reference/index.rst

@@ -43,7 +43,8 @@ Auxiliary classes
    dataset.DatablockUploadModels
    PID
    RemotePath
-   model.DatasetType
+   Thumbnail
+   DatasetType
 
 Exceptions
 ~~~~~~~~~~

docs/release-notes.rst

@@ -44,6 +44,7 @@ Features
 * File downloads now use the checksum algorithm stored in SciCat when possible.
   So it is no longer necessary to specify it by hand in many cases.
 * Added ``SFTPFileTransfer`` which is similar to ``SSHFileTransfer`` but relies only on SFTP.
+* Added support for attachments.
 
 Breaking changes
 ~~~~~~~~~~~~~~~~

docs/user-guide/uploading.ipynb

@@ -308,6 +308,79 @@
     "Or by looking at each file individually as shown in the section above."
    ]
   },
+  {
+   "cell_type": "markdown",
+   "id": "ebbfc735-e8eb-4ceb-b5fb-ca252dd9ee6f",
+   "metadata": {},
+   "source": [
+    "## Attaching images to datasets\n",
+    "\n",
+    "It is possible to attach *small* images to datasets.\n",
+    "In SciCat, this is done by creating 'attachment' objects which contain the image.\n",
+    "Scitacean handles those via the `attachments` property of `Dataset`.\n",
+    "For our locally created dataset, the property is an empty list and we can add an attachment like this:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "5f15eedf-ab14-4f1b-8659-d8f62c6975bc",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from scitacean import Attachment, Thumbnail\n",
+    "\n",
+    "dset.attachments.append(\n",
+    "    Attachment(\n",
+    "        caption=\"Scitacean logo\",\n",
+    "        owner_group=dset.owner_group,\n",
+    "        thumbnail=Thumbnail.load_file(\"./logo.png\"),\n",
+    "    )\n",
+    ")\n",
+    "dset.attachments[0]"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "986b6271-4faf-4b2c-9283-872f6fdf8281",
+   "metadata": {},
+   "source": [
+    "We used `Thumbnail.load_file` because it properly encodes the file for SciCat.\n",
+    "\n",
+    "When we then upload the dataset, the client automatically uploads all attachments as well.\n",
+    "Note that this creates a new dataset in SciCat.\n",
+    "If you want to add attachments to an existing dataset after upload, you need to use the lower-level API through `client.scicat.create_attachment_for_dataset` or the web interface directly."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "f82de9eb-2a34-436c-8694-1a13af82cb66",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "finalized = client.upload_new_dataset_now(dset)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "7f686546-da06-4c15-b357-b44da8bfc328",
+   "metadata": {},
+   "source": [
+    "In order to download the attachments again, we can pass `attachments=True` when downloading the dataset:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "b8a673c7-20b2-4e0e-a003-016227821651",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "downloaded = client.get_dataset(finalized.pid, attachments=True)\n",
+    "downloaded.attachments[0]"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,

src/scitacean/__init__.py

@@ -14,11 +14,13 @@
 from .error import FileUploadError, IntegrityError, ScicatCommError, ScicatLoginError
 from .file import File
 from .filesystem import RemotePath
-from .model import DatasetType
+from .model import Attachment, DatasetType
 from .pid import PID
+from .thumbnail import Thumbnail
 from .warning import VisibleDeprecationWarning
 
 __all__ = (
+    "Attachment",
     "Client",
     "Dataset",
     "DatasetType",
@@ -30,5 +32,6 @@
     "RemotePath",
     "ScicatCommError",
     "ScicatLoginError",
+    "Thumbnail",
     "VisibleDeprecationWarning",
 )

src/scitacean/_base_model.py

@@ -6,7 +6,19 @@
 
 import dataclasses
 from datetime import datetime
-from typing import Any, ClassVar, Dict, Iterable, Optional, Tuple, Type, TypeVar, Union
+from typing import (
+    Any,
+    ClassVar,
+    Dict,
+    Iterable,
+    List,
+    Optional,
+    Tuple,
+    Type,
+    TypeVar,
+    Union,
+    overload,
+)
 
 import pydantic
 from dateutil.parser import parse as parse_datetime
@@ -16,6 +28,7 @@
 from .filesystem import RemotePath
 from .logging import get_logger
 from .pid import PID
+from .thumbnail import Thumbnail
 
 try:
     # Python 3.11+
@@ -51,6 +64,7 @@ class Config:
             json_encoders = {  # noqa: RUF012
                 PID: lambda v: str(v),
                 RemotePath: lambda v: v.posix,
+                Thumbnail: lambda v: v.serialize(),
             }
 
     else:
@@ -89,12 +103,17 @@ def _delete_ignored_args(self, args: Dict[str, Any]) -> None:
     # The mask is cached afterward.
     @classmethod
     def _init_mask(cls: Type[ModelType], instance: ModelType) -> None:
-        field_names = {field.alias for field in instance.get_model_fields().values()}
+        def get_name(name: str, field: Any) -> Any:
+            return field.alias if field.alias is not None else name
+
+        field_names = {
+            get_name(name, field) for name, field in instance.get_model_fields().items()
+        }
         default_mask = tuple(key for key in _IGNORED_KWARGS if key not in field_names)
         cls._masked_fields = cls._user_mask + default_mask
 
     @classmethod
-    def user_model_type(cls) -> Optional[type]:
+    def user_model_type(cls) -> Optional[Type[BaseUserModel]]:
         """Return the user model type for this model.
 
         Returns ``None`` if there is no user model, e.g., for ``Dataset``
@@ -103,15 +122,15 @@ def user_model_type(cls) -> Optional[type]:
         return None
 
     @classmethod
-    def upload_model_type(cls) -> Optional[type]:
+    def upload_model_type(cls) -> Optional[Type[BaseModel]]:
         """Return the upload model type for this model.
 
         Returns ``None`` if the model cannot be uploaded or this is an upload model.
         """
         return None
 
     @classmethod
-    def download_model_type(cls) -> Optional[type]:
+    def download_model_type(cls) -> Optional[Type[BaseModel]]:
         """Return the download model type for this model.
 
         Returns ``None`` if this is a download model.
@@ -177,22 +196,31 @@ def _upload_model_dict(self) -> Dict[str, Any]:
     def from_download_model(cls, download_model: Any) -> BaseUserModel:
         raise NotImplementedError("Function does not exist for BaseUserModel")
 
+    def make_upload_model(self) -> BaseModel:
+        raise NotImplementedError("Function does not exist for BaseUserModel")
+
     @classmethod
-    def upload_model_type(cls) -> Optional[type]:
+    def upload_model_type(cls) -> Optional[Type[BaseModel]]:
         """Return the upload model type for this user model.
 
         Returns ``None`` if the model cannot be uploaded.
         """
         return None
 
     @classmethod
-    def download_model_type(cls) -> type:
+    def download_model_type(cls) -> Type[BaseModel]:
         """Return the download model type for this user model."""
         # There is no sensible default value here as there always exists a download
         # model.
         # All child classes must implement this function.
         raise NotImplementedError("Function does not exist for BaseUserModel")
 
+    def _repr_html_(self) -> Optional[str]:
+        """Return an HTML representation of the model if possible."""
+        from ._html_repr import user_model_html_repr
+
+        return user_model_html_repr(self)
+
 
 def construct(
     model: Type[PydanticModelType],
@@ -281,6 +309,64 @@ def validate_orcids(value: Optional[str]) -> Optional[str]:
     )
 
 
+@overload
+def convert_download_to_user_model(download_model: None) -> None:
+    ...
+
+
+@overload
+def convert_download_to_user_model(download_model: BaseModel) -> BaseUserModel:
+    ...
+
+
+@overload
+def convert_download_to_user_model(
+    download_model: Iterable[BaseModel],
+) -> List[BaseUserModel]:
+    ...
+
+
+def convert_download_to_user_model(
+    download_model: Optional[Union[BaseModel, Iterable[BaseModel]]]
+) -> Optional[Union[BaseUserModel, List[BaseUserModel]]]:
+    """Construct user models from download models."""
+    if download_model is None:
+        return download_model
+    if isinstance(download_model, BaseModel):
+        if (user_type := download_model.user_model_type()) is None:
+            raise TypeError("Cannot convert to user model in this way.")
+        return user_type.from_download_model(download_model)
+    return list(map(convert_download_to_user_model, download_model))
+
+
+@overload
+def convert_user_to_upload_model(user_model: None) -> None:
+    ...
+
+
+@overload
+def convert_user_to_upload_model(user_model: BaseUserModel) -> BaseModel:
+    ...
+
+
+@overload
+def convert_user_to_upload_model(
+    user_model: Iterable[BaseUserModel],
+) -> List[BaseModel]:
+    ...
+
+
+def convert_user_to_upload_model(
+    user_model: Optional[Union[BaseUserModel, Iterable[BaseUserModel]]]
+) -> Optional[Union[BaseModel, List[BaseModel]]]:
+    """Construct upload models from user models."""
+    if user_model is None:
+        return None
+    if isinstance(user_model, BaseUserModel):
+        return user_model.make_upload_model()
+    return list(map(convert_user_to_upload_model, user_model))
+
+
 def _model_field_name_of(cls_name: str, name: str) -> str:
     """Convert a user model field name to a SciCat model field name.
 

src/scitacean/_dataset_fields.py

@@ -22,6 +22,7 @@
 from .filesystem import RemotePath
 from .model import (
     construct,
+    Attachment,
     BaseModel,
     BaseUserModel,
     DownloadDataset,
@@ -565,6 +566,7 @@ def used_by(self, dataset_type: DatasetType) -> bool:
         "_type",
         "_default_checksum_algorithm",
         "_orig_datablocks",
+        "_attachments",
     )
 
     def __init__(
@@ -655,6 +657,7 @@ def __init__(
             checksum_algorithm
         )
         self._orig_datablocks: List[OrigDatablock] = []
+        self._attachments: Optional[List[Attachment]] = []
 
     @property
     def access_groups(self) -> Optional[List[str]]: