Feature/teams

src/albert/albert.py

@@ -31,6 +31,7 @@
 from albert.collections.substance import SubstanceCollection
 from albert.collections.tags import TagCollection
 from albert.collections.tasks import TaskCollection
+from albert.collections.teams import TeamsCollection
 from albert.collections.un_numbers import UnNumberCollection
 from albert.collections.units import UnitCollection
 from albert.collections.users import UserCollection
@@ -203,6 +204,10 @@ def storage_locations(self) -> StorageLocationsCollection:
     def pricings(self) -> PricingCollection:
         return PricingCollection(session=self.session)
 
+    @property
+    def teams(self) -> TeamsCollection:
+        return TeamsCollection(session=self.session)
+
     @property
     def files(self) -> FileCollection:
         return FileCollection(session=self.session)

src/albert/collections/teams.py

@@ -0,0 +1,158 @@
+from collections.abc import Generator, Iterator
+
+from albert.collections.base import BaseCollection
+from albert.resources.teams import Team, TeamRole
+from albert.resources.users import User
+from albert.session import AlbertSession
+
+
+class TeamsCollection(BaseCollection):
+    """
+    TeamsCollection is a collection class for managing teams entities.
+
+    Parameters
+    ----------
+    session : AlbertSession
+        The Albert session instance.
+
+    Attributes
+    ----------
+    base_path : str
+        The base URL for project API requests.
+
+    Methods
+    -------
+    list(params: dict) -> Iterator
+        Lists teams with optional filters.
+    """
+
+    _api_version = "v3"
+
+    def __init__(self, *, session: AlbertSession):
+        """
+        Initialize a TeamsCollection object.
+
+        Parameters
+        ----------
+        session : AlbertSession
+            The Albert session instance.
+        """
+        super().__init__(session=session)
+        self.base_path = f"/api/{TeamsCollection._api_version}/teams"
+
+    def add_users_to_team(
+        self, *, team: Team, users: list[User], team_role: TeamRole = TeamRole.TEAM_VIEWER
+    ) -> bool:
+        """
+        add users to a team
+        """
+        # build payload
+        newValue = []
+        for _u in users:
+            newValue.append({"id": _u.id, "fgc": team_role})
+        payload = [
+            {
+                "id": team.id,
+                "data": [{"operation": "add", "attribute": "ACL", "newValue": newValue}],
+            }
+        ]
+        # run request
+        self.session.patch(self.base_path, json=payload)
+        return True
+
+    def remove_users_from_team(self, *, team: Team, users: list[User]) -> bool:
+        """
+        add users to a team
+        """
+        # build payload
+        payload = [
+            {
+                "id": team.id,
+                "data": [
+                    {
+                        "operation": "delete",
+                        "attribute": "ACL",
+                        "oldValue": [{"id": _u.id} for _u in users],
+                    }
+                ],
+            }
+        ]
+        # run request
+        self.session.patch(self.base_path, json=payload)
+        return True
+
+    def _list_generator(
+        self,
+        *,
+        limit: int = 100,
+        # order_by: OrderBy = OrderBy.DESCENDING,
+        name: list[str] = None,
+        # exact_match: bool = True,
+        # start_key: str | None = None,
+    ) -> Generator[Team, None, None]:
+        """
+        Lists team entities with optional filters.
+
+        Parameters
+        ----------
+        limit : int, optional
+            The maximum number of teams to return, by default 100.
+
+        Returns
+        -------
+        Generator
+            A generator of Team objects.
+        """
+        params = {"limit": limit}  # , "orderBy": order_by.value}
+        if name:
+            params["name"] = name if isinstance(name, list) else [name]
+        #     params["exactMatch"] = str(exact_match).lower()
+        # if start_key:  # pragma: no cover
+        #     params["startKey"] = start_key
+
+        while True:
+            response = self.session.get(self.base_path, params=params)
+            teams_data = response.json().get("Items", [])
+            if not teams_data or teams_data == []:
+                break
+            for t in teams_data:
+                this_team = Team(**t)
+                yield this_team
+            start_key = response.json().get("lastKey")
+            if not start_key:
+                break
+            params["startKey"] = start_key
+
+    def list(
+        self,
+        # *,
+        # order_by: OrderBy = OrderBy.DESCENDING,
+        name: str | list[str] = None,
+        # exact_match: bool = True
+    ) -> Iterator[Team]:
+        """Lists the available Teams
+
+        Parameters
+        ----------
+        params : dict, optional
+            _description_, by default {}
+
+        Returns
+        -------
+        List
+            List of available Roles
+        """
+        return self._list_generator(name=name)
+
+    def create(self, *, team: Team) -> Team:
+        """ """
+        response = self.session.post(
+            self.base_path, json=team.model_dump(by_alias=True, exclude_none=True)
+        )
+        return Team(**response.json())
+
+    def delete(self, *, team_id: str) -> bool:
+        """ """
+        url = f"{self.base_path}/{team_id}"
+        self.session.delete(url)
+        return True

src/albert/resources/teams.py

@@ -0,0 +1,20 @@
+from enum import Enum
+
+from pydantic import Field
+
+from albert.resources.acls import ACL
+from albert.resources.base import BaseResource, SecurityClass
+from albert.resources.users import User
+
+
+class TeamRole(str, Enum):
+    TEAM_OWNER = "TeamOwner"
+    TEAM_VIEWER = "TeamViewer"
+
+
+class Team(BaseResource):
+    id: str | None = Field(default=None, alias="albertId")
+    name: str = Field(min_length=1, max_length=255)
+    team_class: SecurityClass | None = Field(default=None, alias="class")
+    user: list[User] | None = Field(default=None)
+    acl: list[ACL] | None = Field(default=None)

tests/collections/test_teams.py

@@ -0,0 +1,22 @@
+from albert.albert import Albert
+from albert.resources.teams import Team
+
+
+def test_create_and_delete(client: Albert, seeded_teams: list[Team]):
+    # get frist team from seeding list
+    t = seeded_teams[0]
+    # create
+    ret_c = client.teams.create(team=t)
+    # assert
+    assert isinstance(ret_c, Team)
+    # delete
+    ret_d = client.teams.delete(team_id=ret_c.t)
+    # assert
+    assert ret_d == True
+
+
+def test_list(client: Albert):
+    # get team
+    t = next(client.teams.list())
+    # check type
+    assert isinstance(t, Team)

tests/conftest.py

@@ -57,6 +57,7 @@
     generate_storage_location_seeds,
     generate_tag_seeds,
     generate_task_seeds,
+    generate_teams_seeds,
     generate_unit_seeds,
     generate_workflow_seeds,
 )
@@ -506,6 +507,18 @@ def seeded_pricings(client: Albert, seed_prefix: str, seeded_inventory, seeded_l
             client.pricings.delete(id=p.id)
 
 
+@pytest.fixture(scope="session")
+def seeded_teams(client: Albert):
+    seeded = []  # init list
+    all_teams = generate_teams_seeds()  # get list of teams from "seeding.py"
+    for t in all_teams:
+        seeded.append(client.teams.create(team=t))  # create all teams -> "setup"
+    yield seeded
+    for t in seeded:
+        with suppress(NotFoundError):
+            client.teams.delete(team_id=t.id)  # delete all teams -> "cleanup"
+
+
 @pytest.fixture(scope="session")
 def seeded_workflows(
     client: Albert,
@@ -589,6 +602,7 @@ def seeded_tasks(
     for t in all_tasks:
         seeded.append(client.tasks.create(task=t))
     yield seeded
+    # workflows cannot be deleted
     for t in seeded:
         with suppress(NotFoundError, BadRequestError):
             client.tasks.delete(id=t.id)

tests/seeding.py

@@ -74,6 +74,7 @@
     TaskCategory,
     TaskPriority,
 )
+from albert.resources.teams import Team
 from albert.resources.units import Unit, UnitCategory
 from albert.resources.users import User
 from albert.resources.workflows import (
@@ -1036,6 +1037,10 @@ def generate_pricing_seeds(
     ]
 
 
+def generate_teams_seeds() -> list[Team]:
+    return [Team(name="TEST -- Team A (SDK)"), Team(name="TEST -- Team B (SDK)")]
+
+
 def generate_workflow_seeds(
     seed_prefix: str,
     seeded_parameter_groups: list[ParameterGroup],