coolipy.resources.teams

  1"""Teams resource clients (sync + async)."""
  2
  3from __future__ import annotations
  4
  5from typing import Any
  6
  7from coolipy._base import AsyncResourceBase, ResourceBase
  8from coolipy._response import CoolipyAPIResponse
  9from coolipy.models.common import (
 10    EnvironmentVariable,
 11    EnvironmentVariableCreate,
 12    EnvironmentVariableUpdate,
 13    MessageResponse,
 14    UUIDResponse,
 15)
 16from coolipy.models.teams import TeamModel, UserModel
 17
 18TeamList = list[TeamModel]
 19UserList = list[UserModel]
 20EnvironmentVariableList = list[EnvironmentVariable]
 21
 22
 23def _dump(model: Any) -> dict[str, Any]:
 24    return model.model_dump(mode="json", exclude_none=True)
 25
 26
 27class Teams(ResourceBase):
 28    """Synchronous client for Coolify teams and shared envs."""
 29
 30    def list(self) -> CoolipyAPIResponse[TeamList]:
 31        """List all teams."""
 32        return self._get("/teams", response_model=list[TeamModel])
 33
 34    def get(self, team_id: int) -> CoolipyAPIResponse[TeamModel]:
 35        """Get a team by ID."""
 36        return self._get(f"/teams/{team_id}", response_model=TeamModel)
 37
 38    def members(self, team_id: int) -> CoolipyAPIResponse[UserList]:
 39        """List members of a team."""
 40        return self._get(f"/teams/{team_id}/members", response_model=list[UserModel])
 41
 42    def current(self) -> CoolipyAPIResponse[TeamModel]:
 43        """Get the authenticated team."""
 44        return self._get("/team", response_model=TeamModel)
 45
 46    def current_members(self) -> CoolipyAPIResponse[UserList]:
 47        """List members of the authenticated team."""
 48        return self._get("/team/members", response_model=list[UserModel])
 49
 50    def envs(self) -> CoolipyAPIResponse[EnvironmentVariableList]:
 51        """List shared environment variables."""
 52        return self._get("/team/envs", response_model=list[EnvironmentVariable])
 53
 54    def create_env(self, model: EnvironmentVariableCreate) -> CoolipyAPIResponse[UUIDResponse]:
 55        """Create a shared environment variable."""
 56        return self._post("/team/envs", json=_dump(model), response_model=UUIDResponse)
 57
 58    def update_env(
 59        self, model: EnvironmentVariableUpdate
 60    ) -> CoolipyAPIResponse[EnvironmentVariable]:
 61        """Update a shared environment variable."""
 62        return self._patch("/team/envs", json=_dump(model), response_model=EnvironmentVariable)
 63
 64    def delete_env(self, env_id: str) -> CoolipyAPIResponse[MessageResponse]:
 65        """Delete a shared environment variable."""
 66        return self._delete(f"/team/envs/{env_id}", response_model=MessageResponse)
 67
 68
 69class AsyncTeams(AsyncResourceBase):
 70    """Asynchronous client for Coolify teams and shared envs."""
 71
 72    async def list(self) -> CoolipyAPIResponse[TeamList]:
 73        """List all teams."""
 74        return await self._get("/teams", response_model=list[TeamModel])
 75
 76    async def get(self, team_id: int) -> CoolipyAPIResponse[TeamModel]:
 77        """Get a team by ID."""
 78        return await self._get(f"/teams/{team_id}", response_model=TeamModel)
 79
 80    async def members(self, team_id: int) -> CoolipyAPIResponse[UserList]:
 81        """List members of a team."""
 82        return await self._get(f"/teams/{team_id}/members", response_model=list[UserModel])
 83
 84    async def current(self) -> CoolipyAPIResponse[TeamModel]:
 85        """Get the authenticated team."""
 86        return await self._get("/team", response_model=TeamModel)
 87
 88    async def current_members(self) -> CoolipyAPIResponse[UserList]:
 89        """List members of the authenticated team."""
 90        return await self._get("/team/members", response_model=list[UserModel])
 91
 92    async def envs(self) -> CoolipyAPIResponse[EnvironmentVariableList]:
 93        """List shared environment variables."""
 94        return await self._get("/team/envs", response_model=list[EnvironmentVariable])
 95
 96    async def create_env(
 97        self, model: EnvironmentVariableCreate
 98    ) -> CoolipyAPIResponse[UUIDResponse]:
 99        """Create a shared environment variable."""
100        return await self._post("/team/envs", json=_dump(model), response_model=UUIDResponse)
101
102    async def update_env(
103        self, model: EnvironmentVariableUpdate
104    ) -> CoolipyAPIResponse[EnvironmentVariable]:
105        """Update a shared environment variable."""
106        return await self._patch(
107            "/team/envs", json=_dump(model), response_model=EnvironmentVariable
108        )
109
110    async def delete_env(self, env_id: str) -> CoolipyAPIResponse[MessageResponse]:
111        """Delete a shared environment variable."""
112        return await self._delete(f"/team/envs/{env_id}", response_model=MessageResponse)