Dev/access rights

.github/workflows/ci.yml

@@ -13,7 +13,7 @@ jobs:
         strategy:
             fail-fast: false
             matrix:
-                python-version: ['3.8', '3.9', '3.10', '3.11', '3.12']
+                python-version: ['3.10', '3.11', '3.12', '3.13', '3.14']
 
         steps:
 

.pre-commit-config.yaml

@@ -26,13 +26,13 @@ repos:
             args: [--profile, black, --filter-files]
 
     - repo: https://github.com/asottile/pyupgrade
-      rev: v3.3.1
+      rev: v3.21.2
       hooks:
           - id: pyupgrade
             args: [--py38-plus]
 
     - repo: https://github.com/PyCQA/flake8
-      rev: 6.0.0
+      rev: 7.3.0
       hooks:
           - id: flake8
             args: [--count, --show-source, --statistics, '--ignore', 'E501,E203,W503,E201,E202,E221,E222,E231,E241,E271,E272,E702,E713']
@@ -45,14 +45,11 @@ repos:
           - id: setup-cfg-fmt
 
     - repo: https://github.com/PyCQA/bandit
-      rev: 1.7.5
+      rev: 1.9.3
       hooks:
       - id: bandit
         args: ["-r"]
         files: ^(dsms)/.*
-        additional_dependencies:
-          - "pbr==2.0.0"
-          - setuptools
 
     - repo: local
       hooks:

dsms/core/configuration.py

@@ -43,7 +43,33 @@ class Loglevel(Enum):
     WARNING = logging.WARNING
 
 
-class Configuration(BaseSettings):
+class BaseConfiguration(BaseSettings):
+    """Base Configuration for DSMS-SDK"""
+
+    label_internally_public: str = Field(
+        "Internally Public",
+        description="Label to use for KItems marked as `internally_public`.",
+    )
+
+    label_externally_public: str = Field(
+        "Externally Public",
+        description="Label to use for KItems marked as `externally_public`.",
+    )
+
+    id_internally_public: str = Field(
+        "dsms:internally-public",
+        description="ID to use for KItems marked as `internally_public`.",
+    )
+
+    id_externally_public: str = Field(
+        "dsms:externally-public",
+        description="ID to use for KItems marked as `externally_public`.",
+    )
+
+    model_config = ConfigDict(use_enum_values=True)
+
+
+class Configuration(BaseConfiguration):
     """General config for DSMS-SDK"""
 
     host_url: AnyUrl = Field(

dsms/core/dsms.py

@@ -3,7 +3,7 @@
 import os
 import warnings
 from enum import Enum
-from typing import TYPE_CHECKING, Any, Dict, List, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
 from uuid import UUID
 
 from dotenv import load_dotenv
@@ -25,12 +25,13 @@
     _get_remote_ktypes,
     _get_process_schemas,
     _get_webform_schemas,
+    _get_user_groups,
+    _get_user_list,
 )
 
 if TYPE_CHECKING:
-    from typing import Optional
-
     from dsms.core.session import Buffers
+    from dsms.knowledge.groups import Group, User
     from dsms.knowledge.search import KItemListModel, SearchResult
 
 
@@ -302,7 +303,9 @@ def kitems(self) -> "KItemListModel":
         warnings.warn(message, DeprecationWarning)
         return _get_kitem_list(self)
 
-    def get_kitems(self, limit=10, offset=0) -> "KItemListModel":
+    def get_kitems(
+        self, user_id: Optional[str] = None, limit=10, offset=0
+    ) -> "KItemListModel":
         """
         Get all available KItems from the remote backend.
 
@@ -311,7 +314,9 @@ def get_kitems(self, limit=10, offset=0) -> "KItemListModel":
             offset (int): The offset in the list of KItems. Defaults to 0.
 
         """
-        return _get_kitem_list(self, limit=limit, offset=offset)
+        return _get_kitem_list(
+            self, user_id=user_id, limit=limit, offset=offset
+        )
 
     @property
     def app_configs(self) -> "List[AppConfig]":
@@ -331,6 +336,16 @@ def session(self) -> "Session":
         """Return DSMS session"""
         return self._session
 
+    @property
+    def user_groups(self) -> "List[Group]":
+        """Return user groups of the DSMS session"""
+        return _get_user_groups(self)
+
+    @property
+    def users(self) -> "List[User]":
+        """Return user list of the DSMS session"""
+        return _get_user_list(self)
+
     @classmethod
     def __get_pydantic_core_schema__(cls):
         """Get validator of the DSMS-object."""

dsms/knowledge/groups/__init__.py

@@ -0,0 +1,15 @@
+"""DSMS User Groups Module."""
+
+from .models import BaseGroup, Group, GroupList, GroupListBase, User, UserList
+from .public import EXTERNALLY_PUBLIC_GROUP, INTERNALLY_PUBLIC_GROUP
+
+__all__ = [
+    "Group",
+    "GroupList",
+    "GroupListBase",
+    "INTERNALLY_PUBLIC_GROUP",
+    "EXTERNALLY_PUBLIC_GROUP",
+    "User",
+    "BaseGroup",
+    "UserList",
+]

dsms/knowledge/groups/models.py

@@ -0,0 +1,148 @@
+"""DSMS User Groups Module."""
+
+from typing import List, Optional
+
+import yaml
+from pydantic import BaseModel, Field
+
+from dsms.core.session import Session
+
+
+class User(BaseModel):
+    """User Model"""
+
+    id: str = Field(..., description="The unique identifier of the user.")
+    username: str = Field(..., description="The username of the user.")
+    user_groups: Optional[List["BaseGroup"]] = Field(
+        None, description="A list of groups the user belongs to."
+    )
+
+    def __repr__(self) -> str:
+        """String representation of the GroupList."""
+        return str(self)
+
+    def __str__(self):
+        """Pretty print the User"""
+        from dsms.knowledge.utils import print_model
+
+        return print_model(
+            self,
+            "user",
+            exclude_extra=Session.dsms.config.hide_properties,
+        )
+
+
+class UserList(list):
+    """List of Users with utility methods."""
+
+    def __repr__(self) -> str:
+        """String representation of the GroupList."""
+        return str(self)
+
+    def __str__(self):
+        """Pretty print the UserList"""
+        from dsms.knowledge.utils import dump_model
+
+        return yaml.dump(
+            [
+                dump_model(
+                    connection,
+                    exclude_extra=Session.dsms.config.hide_properties,
+                )
+                for connection in self
+            ]
+        )
+
+    @property
+    def by_id(self) -> dict[str, User]:
+        """Return a dictionary of users indexed by their ID."""
+        return {user.id: user for user in self}
+
+    @property
+    def by_username(self) -> dict[str, User]:
+        """Return a dictionary of users indexed by their username."""
+        return {user.username: user for user in self}
+
+    @property
+    def by_name(self) -> dict[str, User]:
+        """Return a dictionary of users indexed by their username."""
+        return self.by_username
+
+    def __getitem__(self, user_id: str) -> User:
+        """Get a user by ID"""
+
+        return self.by_id[user_id]
+
+
+class BaseGroup(BaseModel):
+    """User Group Model"""
+
+    id: str = Field(..., description="The unique identifier of the group.")
+    name: str = Field(..., description="The name of the group.")
+
+
+class Group(BaseGroup):
+    """User Group Model with Subgroups"""
+
+    subgroups: Optional[List["Group"]] = Field(
+        None, description="A list of subgroups."
+    )
+
+
+class GroupListBase(list):
+    """Base class for GroupList with utility methods."""
+
+    def __repr__(self) -> str:
+        """String representation of the GroupList."""
+        return str(self)
+
+    def __str__(self):
+        """Pretty print the GroupList"""
+        from dsms.knowledge.utils import dump_model
+
+        return yaml.dump(
+            [
+                dump_model(
+                    connection,
+                    exclude_extra=Session.dsms.config.hide_properties,
+                )
+                for connection in self
+            ]
+        )
+
+
+class GroupList(list):
+    """List of Groups with utility methods."""
+
+    @property
+    def flat(self) -> List[Group]:
+        """Return a flat list of all groups and their subgroups."""
+        flat_list = []
+
+        def _flatten(groups: List[Group]):
+            for group in groups:
+                flat_list.append(
+                    BaseGroup(**group.model_dump(exclude={"subgroups"}))
+                )
+                if group.subgroups:
+                    _flatten(group.subgroups)
+
+        _flatten(self)
+        return GroupListBase(flat_list)
+
+    @property
+    def by_id(self) -> dict[str, Group]:
+        """Return a dictionary of groups indexed by their ID."""
+        return {group.id: group for group in self.flat}
+
+    @property
+    def by_name(self) -> dict[str, Group]:
+        """Return a dictionary of groups indexed by their name."""
+        return {group.name: group for group in self.flat}
+
+
+Group.model_rebuild()
+interally_public = Group(id="dsms:internally_public", name="Internally Public")
+externally_public = Group(
+    id="dsms:externally_public", name="Externally Public"
+)

dsms/knowledge/groups/public.py

@@ -0,0 +1,31 @@
+"""DSMS Public User Groups Module."""
+
+from dsms.core.configuration import BaseConfiguration
+from dsms.core.session import Session
+
+from .models import Group
+
+if not Session.dsms:
+    config = BaseConfiguration()
+else:
+    config = Session.dsms.config
+
+# The internally/externally public group objects will generally
+# be served by the user-service, but we define them here for uniquely
+# setting the ids and names in a common place.
+# They can be adapted through environment variables anyway.
+# A common place is needed because the group objects are used in various places
+# such as internally within the knowledge service, the user service, and the SDK itself.
+# If the IDs and names of these public groups are only delivered in the user service,
+# we cannot distinguish them from the ones which come from keycloak
+# - indicating only organizational groups.
+
+INTERNALLY_PUBLIC_GROUP = Group(
+    id=config.id_internally_public,
+    name=config.label_internally_public,
+)
+
+EXTERNALLY_PUBLIC_GROUP = Group(
+    id=config.id_externally_public,
+    name=config.label_externally_public,
+)

dsms/knowledge/kitem.py

@@ -44,10 +44,9 @@
     KItemRelationshipModel,
     LinkedKItemsList,
     Summary,
-    UserGroup,
+    KItemAccessProperties,
 )
 
-
 from dsms.knowledge.ktype import KType  # isort:skip
 
 from dsms.knowledge.utils import (  # isort:skip
@@ -172,10 +171,6 @@ class KItem(KItemCompactedModel):
     summary: Optional[Union[str, Summary]] = Field(
         None, description="Human readable summary text of the KItem."
     )
-    user_groups: List[UserGroup] = Field(
-        [],
-        description="User groups able to access the KItem.",
-    )
     custom_properties: Optional[Union[KItemCustomPropertiesModel]] = Field(
         None, description="Custom properties associated to the KItem"
     )
@@ -192,6 +187,10 @@ class KItem(KItemCompactedModel):
         default_factory=Avatar, description="KItem avatar interface"
     )
 
+    access_properties: Optional[KItemAccessProperties] = Field(
+        None, description="Access properties of the KItem"
+    )
+
     contexts: List[
         Union["KItem", KItemCompactedModel, KItemBaseModel]
     ] = Field(
@@ -287,6 +286,17 @@ def validate_apps(cls, value: List[App], info: ValidationInfo) -> AppList:
                 app.id = kitem_id
         return AppList(value)
 
+    @field_validator("avatar", mode="after")
+    @classmethod
+    def validate_avatar(cls, value: Avatar, info: ValidationInfo) -> Avatar:
+        """
+        Validate avatar Field
+        """
+        kitem_id = info.data.get("id")
+        if value:
+            value.id = kitem_id
+        return value
+
     @field_validator("linked_kitems", mode="before")
     @classmethod
     def validate_linked_kitems_list(