diff --git a/examples/domains/delete_domain.py b/examples/domains/delete_domain.py deleted file mode 100644 index e69de29..0000000 diff --git a/examples/domains/get_domain.py b/examples/domains/get_domain.py deleted file mode 100644 index e69de29..0000000 diff --git a/examples/domains/list_domains.py b/examples/domains/list_domains.py deleted file mode 100644 index e69de29..0000000 diff --git a/examples/domains/update_domain.py b/examples/domains/update_domain.py deleted file mode 100644 index e69de29..0000000 diff --git a/examples/domains/update_workspace_connections.py b/examples/domains/update_workspace_connections.py deleted file mode 100644 index e69de29..0000000 diff --git a/examples/teams/domains/create_domain.py b/examples/teams/domains/create_domain.py new file mode 100644 index 0000000..7527dc1 --- /dev/null +++ b/examples/teams/domains/create_domain.py @@ -0,0 +1,18 @@ +import asyncio +import logging +from codesphere import CodesphereSDK + +logging.basicConfig(level=logging.INFO) + + +async def main(): + async with CodesphereSDK() as sdk: + team = await sdk.teams.get(team_id=35663) + domain = await team.domains.create(domain_name="test.com") + + print(f"Domain created: {domain.name}") + print(domain.model_dump_json(indent=2)) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/teams/domains/delete_domain.py b/examples/teams/domains/delete_domain.py new file mode 100644 index 0000000..717be45 --- /dev/null +++ b/examples/teams/domains/delete_domain.py @@ -0,0 +1,18 @@ +import asyncio +import logging +from codesphere import CodesphereSDK + +logging.basicConfig(level=logging.INFO) + + +async def main(): + async with CodesphereSDK() as sdk: + team = await sdk.teams.get(team_id=35663) + domain = await team.domains.delete(domain_name="test.com") + + print(f"Domain created: {domain.name}") + print(domain.model_dump_json(indent=2)) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/teams/domains/get_domain.py b/examples/teams/domains/get_domain.py new file mode 100644 index 0000000..2f82f8d --- /dev/null +++ b/examples/teams/domains/get_domain.py @@ -0,0 +1,18 @@ +import asyncio +import logging +from codesphere import CodesphereSDK + +logging.basicConfig(level=logging.INFO) + + +async def main(): + async with CodesphereSDK() as sdk: + team = await sdk.teams.get(team_id=35663) + logging.info(f"Working with team: {team.name}") + domain = await team.domains.get("test.com") + + print(domain.model_dump_json(indent=2)) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/teams/domains/list_domains.py b/examples/teams/domains/list_domains.py new file mode 100644 index 0000000..eb0705c --- /dev/null +++ b/examples/teams/domains/list_domains.py @@ -0,0 +1,19 @@ +import asyncio +import logging +from codesphere import CodesphereSDK + +logging.basicConfig(level=logging.INFO) + + +async def main(): + async with CodesphereSDK() as sdk: + team = await sdk.teams.get(team_id=35663) + domains = await team.domains.list() + + for domain in domains: + print(f"Domain: {domain.name}") + print(domain.model_dump_json(indent=2)) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/teams/domains/update_domain.py b/examples/teams/domains/update_domain.py new file mode 100644 index 0000000..6694b3c --- /dev/null +++ b/examples/teams/domains/update_domain.py @@ -0,0 +1,20 @@ +import asyncio +import logging +from codesphere import CodesphereSDK, CustomDomainConfig + +logging.basicConfig(level=logging.INFO) + + +async def main(): + async with CodesphereSDK() as sdk: + team = await sdk.teams.get(team_id=35663) + domain = await team.domains.get(domain_name="test.com") + + new_config_data = CustomDomainConfig( + max_body_size_mb=24, max_connection_timeout_s=500, use_regex=False + ) + await domain.update(new_config_data) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/teams/domains/update_workspace_connections.py b/examples/teams/domains/update_workspace_connections.py new file mode 100644 index 0000000..f698562 --- /dev/null +++ b/examples/teams/domains/update_workspace_connections.py @@ -0,0 +1,24 @@ +import asyncio +import logging +from codesphere import CodesphereSDK, DomainRouting + +logging.basicConfig(level=logging.INFO) + + +async def main(): + async with CodesphereSDK() as sdk: + team = await sdk.teams.get(team_id=35663) + domainBuilder = DomainRouting() + + routing = ( + domainBuilder.add("/", [74861]).add("/api", [74868]).add("/test", [74868]) + ) + + domain = await team.domains.update_workspace_connections( + name="test.com", connections=routing + ) + print(f"Current routing: {domain.workspaces}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/domains/verify_domain.py b/examples/teams/domains/verify_domain.py similarity index 100% rename from examples/domains/verify_domain.py rename to examples/teams/domains/verify_domain.py diff --git a/examples/workspaces/execute_command.py b/examples/workspaces/execute_command.py index fdae358..a3e63e1 100644 --- a/examples/workspaces/execute_command.py +++ b/examples/workspaces/execute_command.py @@ -2,13 +2,7 @@ import logging from codesphere import CodesphereSDK -# --- Logging-Konfiguration --- logging.basicConfig(level=logging.INFO) -# (Optionale Logger stummschalten) -logging.getLogger("codesphere.http_client").setLevel(logging.WARNING) -logging.getLogger("httpx").setLevel(logging.WARNING) - -log = logging.getLogger(__name__) async def main(): diff --git a/src/codesphere/__init__.py b/src/codesphere/__init__.py index a09318a..44b6606 100644 --- a/src/codesphere/__init__.py +++ b/src/codesphere/__init__.py @@ -23,7 +23,16 @@ from .client import CodesphereSDK from .exceptions import CodesphereError, AuthenticationError -from .resources.team import Team, TeamCreate, TeamBase +from .resources.team import ( + Team, + TeamCreate, + TeamBase, + Domain, + CustomDomainConfig, + DomainVerificationStatus, + DomainBase, + DomainRouting, +) from .resources.workspace import ( Workspace, WorkspaceCreate, @@ -51,4 +60,10 @@ "Characteristic", "WsPlan", "Image", + "Domain", + "CustomDomainConfig", + "DomainVerificationStatus", + "DomainBase", + "DomainsResource", + "DomainRouting", ] diff --git a/src/codesphere/client.py b/src/codesphere/client.py index d076df4..d46b68f 100644 --- a/src/codesphere/client.py +++ b/src/codesphere/client.py @@ -1,9 +1,3 @@ -""" -Codesphere SDK Client - -This module provides the main client class, CodesphereSDK. -""" - from .http_client import APIHttpClient from .resources.metadata import MetadataResource from .resources.team import TeamsResource @@ -11,38 +5,9 @@ class CodesphereSDK: - """The main entrypoint for interacting with the `Codesphere Public API `_. - - This class manages the HTTP client, its lifecycle, - and provides access to the various API resources. - - Primary usage is via an asynchronous context manager: - - Usage: - >>> import asyncio - >>> from codesphere import CodesphereSDK - >>> - >>> async def main(): - >>> async with CodesphereSDK() as sdk: - >>> teams = await sdk.teams.list() - >>> print(teams) - >>> - >>> asyncio.run(main()) - - Attributes: - teams (TeamsResource): Access to Team API operations. - workspaces (WorkspacesResource): Access to Workspace API operations. - metadata (MetadataResource): Access to Metadata API operations. - """ - teams: TeamsResource - """Access to the Team API. (e.g., `sdk.teams.list()`)""" - workspaces: WorkspacesResource - """Access to the Workspace API. (e.g., `sdk.workspaces.list()`)""" - metadata: MetadataResource - """Access to the Metadata API. (e.g., `sdk.metadata.list_plans()`)""" def __init__(self): self._http_client = APIHttpClient() @@ -51,20 +16,9 @@ def __init__(self): self.metadata = MetadataResource(self._http_client) async def open(self): - """Manually opens the underlying HTTP client session. - - Required for manual lifecycle control when not using `async with`. - - Usage: - >>> sdk = CodesphereSDK() - >>> await sdk.open() - >>> # ... API calls ... - >>> await sdk.close() - """ await self._http_client.open() async def close(self): - """Manually closes the underlying HTTP client session.""" await self._http_client.close() async def __aenter__(self): diff --git a/src/codesphere/config.py b/src/codesphere/config.py index 5e5559e..cc2ad15 100644 --- a/src/codesphere/config.py +++ b/src/codesphere/config.py @@ -3,10 +3,6 @@ class Settings(BaseSettings): - """ - API Client Settings - """ - model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", env_prefix="CS_" ) diff --git a/src/codesphere/core/base.py b/src/codesphere/core/base.py index 07d9452..14a6882 100644 --- a/src/codesphere/core/base.py +++ b/src/codesphere/core/base.py @@ -1,7 +1,33 @@ +from typing import Generic, List, TypeVar +from pydantic import BaseModel, ConfigDict, RootModel +from pydantic.alias_generators import to_camel + from ..http_client import APIHttpClient from .handler import _APIOperationExecutor +ModelT = TypeVar("ModelT") + class ResourceBase(_APIOperationExecutor): def __init__(self, http_client: APIHttpClient): self._http_client = http_client + + +class CamelModel(BaseModel): + model_config = ConfigDict( + alias_generator=to_camel, + populate_by_name=True, + ) + + +class ResourceList(RootModel[List[ModelT]], Generic[ModelT]): + root: List[ModelT] + + def __iter__(self): + return iter(self.root) + + def __getitem__(self, item): + return self.root[item] + + def __len__(self): + return len(self.root) diff --git a/src/codesphere/core/handler.py b/src/codesphere/core/handler.py index 5d7f31e..cd2eb7f 100644 --- a/src/codesphere/core/handler.py +++ b/src/codesphere/core/handler.py @@ -72,6 +72,10 @@ def _prepare_request_args(self) -> tuple[str, dict]: else: payload = json_data_obj + if payload is not None: + log.info(f"PAYLOAD TYPE: {type(payload)}") + log.info(f"PAYLOAD CONTENT: {payload}") + request_kwargs = {"params": self.kwargs.get("params"), "json": payload} return endpoint, {k: v for k, v in request_kwargs.items() if v is not None} diff --git a/src/codesphere/core/operations.py b/src/codesphere/core/operations.py index 2f6b137..bf8ee3f 100644 --- a/src/codesphere/core/operations.py +++ b/src/codesphere/core/operations.py @@ -1,22 +1,17 @@ -from typing import Callable, Awaitable, List, Optional, Type, TypeAlias, TypeVar +from typing import Callable, Awaitable, Generic, Optional, Type, TypeAlias, TypeVar from pydantic import BaseModel _T = TypeVar("_T") +ResponseT = TypeVar("ResponseT") +InputT = TypeVar("InputT") AsyncCallable: TypeAlias = Callable[[], Awaitable[_T]] -class APIOperation: - def __init__( - self, - method: str, - endpoint_template: str, - response_model: Type[BaseModel] | Type[List[BaseModel]], - input_model: Optional[Type[BaseModel]] = None, - ): - self.method = method - self.endpoint_template = endpoint_template - self.response_model = response_model - self.input_model = input_model +class APIOperation(BaseModel, Generic[ResponseT, InputT]): + method: str + endpoint_template: str + response_model: Type[ResponseT] + input_model: Optional[Type[InputT]] = None diff --git a/src/codesphere/resources/domain/models.py b/src/codesphere/resources/domain/models.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/codesphere/resources/domain/resources.py b/src/codesphere/resources/domain/resources.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/codesphere/resources/metadata/__init__.py b/src/codesphere/resources/metadata/__init__.py index b0c28cf..1f4cd88 100644 --- a/src/codesphere/resources/metadata/__init__.py +++ b/src/codesphere/resources/metadata/__init__.py @@ -1,6 +1,6 @@ """Metadata Resource & Models""" -from .models import Datacenter, Characteristic, WsPlan, Image +from .schemas import Datacenter, Characteristic, WsPlan, Image from .resources import MetadataResource __all__ = ["Datacenter", "Characteristic", "WsPlan", "Image", "MetadataResource"] diff --git a/src/codesphere/resources/metadata/operations.py b/src/codesphere/resources/metadata/operations.py new file mode 100644 index 0000000..a31383b --- /dev/null +++ b/src/codesphere/resources/metadata/operations.py @@ -0,0 +1,24 @@ +from ...core.base import ResourceList +from ...core.operations import APIOperation +from .schemas import Datacenter, Image, WsPlan + +_LIST_DC_OP = APIOperation( + method="GET", + endpoint_template="/metadata/datacenters", + input_model=type(None), + response_model=ResourceList[Datacenter], +) + +_LIST_PLANS_OP = APIOperation( + method="GET", + endpoint_template="/metadata/workspace-plans", + input_model=type(None), + response_model=ResourceList[WsPlan], +) + +_LIST_IMAGES_OP = APIOperation( + method="GET", + endpoint_template="/metadata/workspace-base-images", + input_model=type(None), + response_model=ResourceList[Image], +) diff --git a/src/codesphere/resources/metadata/resources.py b/src/codesphere/resources/metadata/resources.py index a514e4b..2fa30eb 100644 --- a/src/codesphere/resources/metadata/resources.py +++ b/src/codesphere/resources/metadata/resources.py @@ -1,37 +1,31 @@ -""" -Defines the resource class for the Metadata API endpoints. -""" - from typing import List -from ...core import APIOperation, AsyncCallable +from pydantic import Field +from ...core.base import ResourceList +from ...core import AsyncCallable from ...core import ResourceBase -from .models import Datacenter, WsPlan, Image +from .operations import _LIST_DC_OP, _LIST_IMAGES_OP, _LIST_PLANS_OP +from .schemas import Datacenter, WsPlan, Image class MetadataResource(ResourceBase): - list_datacenters: AsyncCallable[List[Datacenter]] - """Fetches a list of all available data centers.""" - list_datacenters = APIOperation( - method="GET", - endpoint_template="/metadata/datacenters", - input_model=None, - response_model=List[Datacenter], + list_datacenters_op: AsyncCallable[ResourceList[Datacenter]] = Field( + default=_LIST_DC_OP, exclude=True ) - - list_plans: AsyncCallable[List[WsPlan]] - """Fetches a list of all available workspace plans.""" - list_plans = APIOperation( - method="GET", - endpoint_template="/metadata/workspace-plans", - input_model=None, - response_model=List[WsPlan], + list_plans_op: AsyncCallable[ResourceList[WsPlan]] = Field( + default=_LIST_PLANS_OP, exclude=True ) - - list_images: AsyncCallable[List[Image]] - """Fetches a list of all available workspace base images.""" - list_images = APIOperation( - method="GET", - endpoint_template="/metadata/workspace-base-images", - input_model=None, - response_model=List[Image], + list_images_op: AsyncCallable[ResourceList[Image]] = Field( + default=_LIST_IMAGES_OP, exclude=True ) + + async def list_datacenters(self) -> List[Datacenter]: + result = await self.list_datacenters_op() + return result.root + + async def list_plans(self) -> List[WsPlan]: + result = await self.list_plans_op() + return result.root + + async def list_images(self) -> List[Image]: + result = await self.list_images_op() + return result.root diff --git a/src/codesphere/resources/metadata/models.py b/src/codesphere/resources/metadata/schemas.py similarity index 56% rename from src/codesphere/resources/metadata/models.py rename to src/codesphere/resources/metadata/schemas.py index 1791e4f..51e0232 100644 --- a/src/codesphere/resources/metadata/models.py +++ b/src/codesphere/resources/metadata/schemas.py @@ -1,45 +1,46 @@ from __future__ import annotations -from pydantic import BaseModel import datetime +from ...core.base import CamelModel -class Datacenter(BaseModel): + +class Datacenter(CamelModel): """Represents a physical data center location.""" id: int name: str city: str - countryCode: str + country_code: str -class Characteristic(BaseModel): +class Characteristic(CamelModel): """Defines the resource specifications for a WsPlan.""" id: int - CPU: float - GPU: int - RAM: int - SSD: int - TempStorage: int - onDemand: bool + cpu: float + gpu: int + ram: int + ssd: int + temp_storage: int + on_demand: bool -class WsPlan(BaseModel): +class WsPlan(CamelModel): """ Represents a purchasable workspace plan. """ id: int - priceUsd: int + price_usd: int title: str deprecated: bool characteristics: Characteristic - maxReplicas: int + max_replicas: int -class Image(BaseModel): +class Image(CamelModel): """Represents a runnable workspace base image.""" id: str name: str - supportedUntil: datetime.datetime + supported_until: datetime.datetime diff --git a/src/codesphere/resources/team/__init__.py b/src/codesphere/resources/team/__init__.py index 85e8cf6..deb4b45 100644 --- a/src/codesphere/resources/team/__init__.py +++ b/src/codesphere/resources/team/__init__.py @@ -1,4 +1,22 @@ -from .models import Team, TeamCreate, TeamBase +from .schemas import Team, TeamCreate, TeamBase from .resources import TeamsResource +from .domain.resources import ( + Domain, + CustomDomainConfig, + DomainVerificationStatus, + DomainBase, + DomainRouting, +) -__all__ = ["Team", "TeamCreate", "TeamBase", "TeamsResource"] + +__all__ = [ + "Team", + "TeamCreate", + "TeamBase", + "TeamsResource", + "Domain", + "CustomDomainConfig", + "DomainVerificationStatus", + "DomainBase", + "DomainRouting", +] diff --git a/src/codesphere/resources/domain/__init__.py b/src/codesphere/resources/team/domain/__init__.py similarity index 100% rename from src/codesphere/resources/domain/__init__.py rename to src/codesphere/resources/team/domain/__init__.py diff --git a/src/codesphere/resources/team/domain/manager.py b/src/codesphere/resources/team/domain/manager.py new file mode 100644 index 0000000..7f54163 --- /dev/null +++ b/src/codesphere/resources/team/domain/manager.py @@ -0,0 +1,67 @@ +from typing import List, Union +from pydantic import Field + +from ....core.base import ResourceList +from ....core.handler import _APIOperationExecutor +from ....core.operations import AsyncCallable +from ....http_client import APIHttpClient +from .schemas import CustomDomainConfig, DomainRouting, RoutingMap +from .resources import Domain +from .operations import _CREATE_OP, _GET_OP, _LIST_OP, _UPDATE_OP, _UPDATE_WS_OP + + +class TeamDomainManager(_APIOperationExecutor): + """ + Verwaltet Domains im Kontext eines spezifischen Teams. + Zugriff typischerweise über 'team.domains'. + """ + + def __init__(self, http_client: APIHttpClient, team_id: int): + self._http_client = http_client + self.team_id = team_id + + list_op: AsyncCallable[ResourceList[Domain]] = Field( + default=_LIST_OP.model_copy(update={"response_model": ResourceList[Domain]}), + exclude=True, + ) + + async def list(self) -> List[Domain]: + result = await self.list_op() + return result.root + + get_op: AsyncCallable[Domain] = Field( + default=_GET_OP.model_copy(update={"response_model": Domain}), + exclude=True, + ) + + async def get(self, name: str) -> Domain: + return await self.get_op(name=name) + + create_op: AsyncCallable[Domain] = Field( + default=_CREATE_OP.model_copy(update={"response_model": Domain}), + exclude=True, + ) + + async def create(self, name: str) -> Domain: + return await self.create_op(name=name) + + update_op: AsyncCallable[Domain] = Field( + default=_UPDATE_OP.model_copy(update={"response_model": Domain}), + exclude=True, + ) + + async def update(self, name: str, config: CustomDomainConfig) -> Domain: + return await self.update_op(name=name, data=config) + + update_ws_op: AsyncCallable[Domain] = Field( + default=_UPDATE_WS_OP.model_copy(update={"response_model": Domain}), + exclude=True, + ) + + async def update_workspace_connections( + self, name: str, connections: Union[DomainRouting, RoutingMap] + ) -> Domain: + payload = ( + connections.root if isinstance(connections, DomainRouting) else connections + ) + return await self.update_ws_op(name=name, data=payload) diff --git a/src/codesphere/resources/team/domain/operations.py b/src/codesphere/resources/team/domain/operations.py new file mode 100644 index 0000000..a3c07b2 --- /dev/null +++ b/src/codesphere/resources/team/domain/operations.py @@ -0,0 +1,51 @@ +from ....core.base import ResourceList +from ....core.operations import APIOperation +from .schemas import ( + CustomDomainConfig, + DomainBase, + DomainVerificationStatus, +) + +_LIST_OP = APIOperation( + method="GET", + endpoint_template="/domains/team/{team_id}", + response_model=ResourceList[DomainBase], +) + +_GET_OP = APIOperation( + method="GET", + endpoint_template="/domains/team/{team_id}/domain/{name}", + response_model=DomainBase, +) + +_CREATE_OP = APIOperation( + method="POST", + endpoint_template="/domains/team/{team_id}/domain/{name}", + response_model=DomainBase, +) + +_UPDATE_OP = APIOperation( + method="PATCH", + endpoint_template="/domains/team/{team_id}/domain/{name}", + input_model=CustomDomainConfig, + response_model=DomainBase, +) + +_UPDATE_WS_OP = APIOperation( + method="PUT", + endpoint_template="/domains/team/{team_id}/domain/{name}/workspace-connections", + input_model=None, + response_model=DomainBase, +) + +_VERIFY_OP = APIOperation( + method="POST", + endpoint_template="/domains/team/{team_id}/domain/{name}/verify", + response_model=DomainVerificationStatus, +) + +_DELETE_OP = APIOperation( + method="DELETE", + endpoint_template="/domains/team/{team_id}/domain/{name}", + response_model=DomainBase, +) diff --git a/src/codesphere/resources/team/domain/resources.py b/src/codesphere/resources/team/domain/resources.py new file mode 100644 index 0000000..f212293 --- /dev/null +++ b/src/codesphere/resources/team/domain/resources.py @@ -0,0 +1,58 @@ +from __future__ import annotations +import logging +from typing import Union +from pydantic import Field + +from .operations import ( + _DELETE_OP, + _UPDATE_OP, + _UPDATE_WS_OP, + _VERIFY_OP, +) +from .schemas import ( + CustomDomainConfig, + DomainBase, + DomainRouting, + DomainVerificationStatus, + RoutingMap, +) + +from ....core.handler import _APIOperationExecutor +from ....core.operations import AsyncCallable +from ....utils import update_model_fields + + +log = logging.getLogger(__name__) + + +class Domain(DomainBase, _APIOperationExecutor): + update_op: AsyncCallable[None] = Field(default=_UPDATE_OP, exclude=True) + update_workspace_connections_op: AsyncCallable[None] = Field( + default=_UPDATE_WS_OP, exclude=True + ) + verify_domain_op: AsyncCallable[None] = Field(default=_VERIFY_OP, exclude=True) + delete_domain_op: AsyncCallable[None] = Field(default=_DELETE_OP, exclude=True) + + async def update(self, data: CustomDomainConfig) -> Domain: + payload = data.model_dump(exclude_unset=True, by_alias=True) + response = await self.update_op(data=payload) + update_model_fields(target=self, source=response) + return response + + async def update_workspace_connections( + self, connections: Union[DomainRouting, RoutingMap] + ) -> Domain: + payload = ( + connections.root if isinstance(connections, DomainRouting) else connections + ) + response = await self.update_workspace_connections_op(data=payload) + update_model_fields(target=self, source=response) + return response + + async def verify_status(self) -> DomainVerificationStatus: + response = await self.verify_domain_op() + update_model_fields(target=self.domain_verification_status, source=response) + return response + + async def delete(self) -> None: + await self.delete_domain_op() diff --git a/src/codesphere/resources/team/domain/schemas.py b/src/codesphere/resources/team/domain/schemas.py new file mode 100644 index 0000000..c2f86a0 --- /dev/null +++ b/src/codesphere/resources/team/domain/schemas.py @@ -0,0 +1,54 @@ +from __future__ import annotations +from typing import Dict, List, Optional, TypeAlias +from pydantic import Field, RootModel + +from ....core.base import CamelModel + +RoutingMap: TypeAlias = Dict[str, List[int]] + + +class CertificateRequestStatus(CamelModel): + issued: bool + reason: Optional[str] = None + + +class DNSEntries(CamelModel): + a: str + cname: str + txt: str + + +class DomainVerificationStatus(CamelModel): + verified: bool + reason: Optional[str] = None + + +class CustomDomainConfig(CamelModel): + restricted: Optional[bool] = None + max_body_size_mb: Optional[int] = None + max_connection_timeout_s: Optional[int] = None + use_regex: Optional[bool] = None + + +class DomainRouting(RootModel): + """ + Helper class to build the routing configuration. + """ + + root: RoutingMap = Field(default_factory=dict) + + def add(self, path: str, workspace_ids: List[int]) -> DomainRouting: + self.root[path] = workspace_ids + return self + + +class DomainBase(CamelModel): + name: str + team_id: int + data_center_id: int + workspaces: RoutingMap + certificate_request_status: CertificateRequestStatus + dns_entries: DNSEntries + domain_verification_status: DomainVerificationStatus + custom_config_revision: Optional[int] = None + custom_config: Optional[CustomDomainConfig] = None diff --git a/src/codesphere/resources/team/models.py b/src/codesphere/resources/team/models.py deleted file mode 100644 index 4004935..0000000 --- a/src/codesphere/resources/team/models.py +++ /dev/null @@ -1,60 +0,0 @@ -""" -Pydantic models for the Team resource. - -Includes the 'active' Team model (with API methods) -and data-only models for creation payloads. -""" - -from __future__ import annotations -from pydantic import BaseModel, Field -from typing import Optional - -from ...core import _APIOperationExecutor, APIOperation, AsyncCallable - - -class TeamCreate(BaseModel): - """Data payload required for creating a new team.""" - - name: str - dc: int - - -class TeamBase(BaseModel): - id: int - name: str - description: Optional[str] = None - avatarId: Optional[str] = None - avatarUrl: Optional[str] = None - isFirst: bool - defaultDataCenterId: int - role: Optional[int] = None - - -class Team(TeamBase, _APIOperationExecutor): - """ - Represents a complete, 'active' team object returned from the API. - - This model includes methods to interact with the resource directly. - - Usage: - >>> # Get a team instance - >>> team = await sdk.teams.get(team_id=123) - >>> # Call methods on it - >>> await team.delete() - """ - - delete: AsyncCallable[None] - """ - Deletes this team via the API. - - .. warning:: - This is a destructive operation and cannot be undone. - """ - delete = Field( - default=APIOperation( - method="DELETE", - endpoint_template="/teams/{id}", - response_model=None, - ), - exclude=True, - ) diff --git a/src/codesphere/resources/team/operations.py b/src/codesphere/resources/team/operations.py new file mode 100644 index 0000000..ca56871 --- /dev/null +++ b/src/codesphere/resources/team/operations.py @@ -0,0 +1,24 @@ +from .schemas import Team, TeamCreate +from ...core.base import ResourceList +from ...core.operations import APIOperation + +_LIST_TEAMS_OP = APIOperation( + method="GET", + endpoint_template="/teams", + input_model=type(None), + response_model=ResourceList[Team], +) + +_GET_TEAM_OP = APIOperation( + method="GET", + endpoint_template="/teams/{team_id}", + input_model=type(None), + response_model=Team, +) + +_CREATE_TEAM_OP = APIOperation( + method="POST", + endpoint_template="/teams", + input_model=TeamCreate, + response_model=Team, +) diff --git a/src/codesphere/resources/team/resources.py b/src/codesphere/resources/team/resources.py index 730a072..7c9fb3c 100644 --- a/src/codesphere/resources/team/resources.py +++ b/src/codesphere/resources/team/resources.py @@ -1,78 +1,33 @@ -""" -Defines the resource class for the Team API endpoints. -""" +from typing import List -from typing import List, Protocol +from pydantic import Field -from ...core import APIOperation, AsyncCallable, ResourceBase -from .models import Team, TeamCreate +from .operations import ( + _CREATE_TEAM_OP, + _GET_TEAM_OP, + _LIST_TEAMS_OP, +) - -class GetTeamCallable(Protocol): - async def __call__(self, *, team_id: int) -> Team: ... - - -class CreateTeamCallable(Protocol): - async def __call__(self, *, data: TeamCreate) -> Team: ... +from ...core.base import ResourceList +from ...core import AsyncCallable, ResourceBase +from .schemas import Team, TeamCreate class TeamsResource(ResourceBase): - """ - Provides access to the Team API operations. + list_team_op: AsyncCallable[ResourceList[Team]] = Field( + default=_LIST_TEAMS_OP, exclude=True + ) - Usage: - >>> # Access via the main SDK client - >>> async with CodesphereSDK() as sdk: - >>> new_team_data = TeamCreate(name="My Team", dc=1) - >>> new_team = await sdk.teams.create(data=new_team_data) - >>> team = await sdk.teams.get(team_id=new_team.id) - """ + async def list(self) -> List[Team]: + result = await self.list_team_op() + return result.root - list: AsyncCallable[List[Team]] - """ - Fetches a list of all teams the user belongs to. - - Returns: - List[Team]: A list of Team objects. - """ - list = APIOperation( - method="GET", - endpoint_template="/teams", - input_model=None, - response_model=List[Team], - ) + get_team_op: AsyncCallable[Team] = Field(default=_GET_TEAM_OP, exclude=True) - get: GetTeamCallable - """ - Fetches a single team by its ID. - - Args: - team_id (int): The unique identifier for the team. - - Returns: - Team: The requested Team object. - """ - get = APIOperation( - method="GET", - endpoint_template="/teams/{team_id}", - input_model=None, - response_model=Team, - ) + async def get(self, team_id: int) -> Team: + return await self.get_team_op(data=team_id) - create: CreateTeamCallable - """ - Creates a new team. - - Args: - data (TeamCreate): A :class:`~.models.TeamCreate` object - containing the new team's information. - - Returns: - Team: The newly created Team object. - """ - create = APIOperation( - method="POST", - endpoint_template="/teams", - input_model=TeamCreate, - response_model=Team, - ) + create_team_op: AsyncCallable[Team] = Field(default=_CREATE_TEAM_OP, exclude=True) + + async def create(self, payload: TeamCreate) -> Team: + return await self.create_team_op(data=payload) diff --git a/src/codesphere/resources/team/schemas.py b/src/codesphere/resources/team/schemas.py new file mode 100644 index 0000000..ca8b659 --- /dev/null +++ b/src/codesphere/resources/team/schemas.py @@ -0,0 +1,46 @@ +from __future__ import annotations +from functools import cached_property +from pydantic import Field +from typing import TYPE_CHECKING, Optional + +from .domain.manager import TeamDomainManager +from ...core.base import CamelModel +from ...core import _APIOperationExecutor, APIOperation, AsyncCallable + +if TYPE_CHECKING: + pass + + +class TeamCreate(CamelModel): + name: str + dc: int + + +class TeamBase(CamelModel): + id: int + name: str + description: Optional[str] = None + avatar_id: Optional[str] = None + avatar_url: Optional[str] = None + is_first: bool + default_data_center_id: int + role: Optional[int] = None + + +class Team(TeamBase, _APIOperationExecutor): + delete: AsyncCallable[None] + delete = Field( + default=APIOperation( + method="DELETE", + endpoint_template="/teams/{id}", + response_model=type(None), + ), + exclude=True, + ) + + @cached_property + def domains(self) -> TeamDomainManager: + if not self._http_client: + raise RuntimeError("Cannot access 'domains' on a detached model.") + + return TeamDomainManager(http_client=self._http_client, team_id=self.id) diff --git a/examples/domains/create_domain.py b/src/codesphere/resources/team/uasage/.gitkeep similarity index 100% rename from examples/domains/create_domain.py rename to src/codesphere/resources/team/uasage/.gitkeep diff --git a/src/codesphere/resources/workspace/__init__.py b/src/codesphere/resources/workspace/__init__.py index af4e319..0bb117c 100644 --- a/src/codesphere/resources/workspace/__init__.py +++ b/src/codesphere/resources/workspace/__init__.py @@ -1,4 +1,4 @@ -from .models import Workspace, WorkspaceCreate, WorkspaceUpdate, WorkspaceStatus +from .schemas import Workspace, WorkspaceCreate, WorkspaceUpdate, WorkspaceStatus from .resources import WorkspacesResource __all__ = [ diff --git a/src/codesphere/resources/workspace/envVars/models.py b/src/codesphere/resources/workspace/envVars/models.py index 9f6b408..d48f883 100644 --- a/src/codesphere/resources/workspace/envVars/models.py +++ b/src/codesphere/resources/workspace/envVars/models.py @@ -3,9 +3,11 @@ from typing import Dict, List, Union from pydantic import BaseModel, Field -from codesphere.core.handler import _APIOperationExecutor -from codesphere.core.operations import APIOperation, AsyncCallable -from codesphere.http_client import APIHttpClient +from ....core.base import ResourceList +from ....core.handler import _APIOperationExecutor +from ....core.operations import AsyncCallable +from ....http_client import APIHttpClient +from .operations import _BULK_DELETE_OP, _BULK_SET_OP, _GET_OP log = logging.getLogger(__name__) @@ -16,66 +18,48 @@ class EnvVar(BaseModel): class WorkspaceEnvVarManager(_APIOperationExecutor): - """ - Verwaltet die Env Vars für einen *bestimmten* Workspace. - Wird typischerweise über 'workspace.env_vars' aufgerufen. - """ - def __init__(self, http_client: APIHttpClient, workspace_id: int): self._http_client = http_client self._workspace_id = workspace_id self.id = workspace_id - get_all_op: AsyncCallable[List[EnvVar]] = Field( - default=APIOperation( - method="GET", - endpoint_template="/workspaces/{id}/env-vars", - response_model=List[EnvVar], - ), + get_all_op: AsyncCallable[ResourceList[EnvVar]] = Field( + default=_GET_OP, exclude=True, ) + async def get(self) -> List[EnvVar]: + return await self.get_all_op() + bulk_set_op: AsyncCallable[None] = Field( - default=APIOperation( - method="PUT", - endpoint_template="/workspaces/{id}/env-vars", - response_model=None, - ), + default=_BULK_SET_OP, exclude=True, ) + async def set( + self, env_vars: Union[ResourceList[EnvVar], List[Dict[str, str]]] + ) -> None: + payload = ResourceList[EnvVar].model_validate(env_vars) + await self.bulk_set_op(data=payload.model_dump()) + bulk_delete_op: AsyncCallable[None] = Field( - default=APIOperation( - method="DELETE", - endpoint_template="/workspaces/{id}/env-vars", - response_model=None, - ), + default=_BULK_DELETE_OP, exclude=True, ) - async def get(self) -> List[EnvVar]: - """Fetches all environment variables for this workspace.""" - env_vars = await self.get_all_op() - return env_vars - - async def set(self, env_vars: Union[List[EnvVar], List[Dict[str, str]]]) -> None: - """Sets or updates environment variables for this workspace.""" - json_payload = [] - if env_vars: - if isinstance(env_vars[0], EnvVar): - json_payload = [var.model_dump() for var in env_vars] - else: - json_payload = env_vars + async def delete(self, items: Union[List[str], ResourceList[EnvVar]]) -> None: + if not items: + return - await self.bulk_set_op(data=json_payload) + payload: List[str] = [] - async def delete(self, var_names: Union[List[str], List[EnvVar]]) -> None: - """Deletes specific environment variables from this workspace.""" - payload = [] - if var_names: - if isinstance(var_names[0], EnvVar): - payload = [var.name for var in var_names] - else: - payload = var_names + for item in items: + if isinstance(item, str): + payload.append(item) + elif hasattr(item, "name"): + payload.append(item.name) + elif isinstance(item, dict) and "name" in item: + payload.append(item["name"]) - await self.bulk_delete_op(data=payload) + if payload: + await self.bulk_delete_op(data=payload) diff --git a/src/codesphere/resources/workspace/envVars/operations.py b/src/codesphere/resources/workspace/envVars/operations.py new file mode 100644 index 0000000..d5e1be6 --- /dev/null +++ b/src/codesphere/resources/workspace/envVars/operations.py @@ -0,0 +1,41 @@ +from .models import EnvVar +from ...workspace.schemas import CommandInput, CommandOutput, WorkspaceStatus +from ....core.base import ResourceList +from ....core.operations import APIOperation + +_GET_OP = APIOperation( + method="GET", + endpoint_template="/workspaces/{id}/env-vars", + response_model=ResourceList[EnvVar], +) + +_BULK_SET_OP = APIOperation( + method="PUT", + endpoint_template="/workspaces/{id}/env-vars", + response_model=type(None), +) + +_BULK_DELETE_OP = APIOperation( + method="DELETE", + endpoint_template="/workspaces/{id}/env-vars", + response_model=type(None), +) + +_DELETE_OP = APIOperation( + method="DELETE", + endpoint_template="/workspaces/{id}", + response_model=type(None), +) + +_GET_STATUS_OP = APIOperation( + method="GET", + endpoint_template="/workspaces/{id}/status", + response_model=WorkspaceStatus, +) + +_EXECUTE_COMMAND_OP = APIOperation( + method="POST", + endpoint_template="/workspaces/{id}/execute", + input_model=CommandInput, + response_model=CommandOutput, +) diff --git a/src/codesphere/resources/workspace/models.py b/src/codesphere/resources/workspace/models.py deleted file mode 100644 index deb8534..0000000 --- a/src/codesphere/resources/workspace/models.py +++ /dev/null @@ -1,163 +0,0 @@ -from __future__ import annotations -from functools import cached_property -import logging -from pydantic import BaseModel, Field -from typing import Dict, Optional, List - -from ...core import _APIOperationExecutor, APIOperation, AsyncCallable -from .envVars import EnvVar, WorkspaceEnvVarManager - -log = logging.getLogger(__name__) - - -class WorkspaceCreate(BaseModel): - teamId: int - name: str - planId: int - baseImage: Optional[str] = None - isPrivateRepo: bool = True - replicas: int = 1 - gitUrl: Optional[str] = None - initialBranch: Optional[str] = None - cloneDepth: Optional[int] = None - sourceWorkspaceId: Optional[int] = None - welcomeMessage: Optional[str] = None - vpnConfig: Optional[str] = None - restricted: Optional[bool] = None - env: Optional[List[EnvVar]] = None - - -class WorkspaceBase(BaseModel): - id: int - teamId: int - name: str - planId: int - isPrivateRepo: bool - replicas: int - baseImage: Optional[str] = None - dataCenterId: int - userId: int - gitUrl: Optional[str] = None - initialBranch: Optional[str] = None - sourceWorkspaceId: Optional[int] = None - welcomeMessage: Optional[str] = None - vpnConfig: Optional[str] = None - restricted: bool - - -class WorkspaceUpdate(BaseModel): - planId: Optional[int] = None - baseImage: Optional[str] = None - name: Optional[str] = None - replicas: Optional[int] = None - vpnConfig: Optional[str] = None - restricted: Optional[bool] = None - - -class WorkspaceStatus(BaseModel): - isRunning: bool - - -class CommandInput(BaseModel): - command: str - env: Optional[Dict[str, str]] = None - - -class CommandOutput(BaseModel): - command: str - workingDir: str - output: str - error: str - - -class Workspace(WorkspaceBase, _APIOperationExecutor): - update_op: AsyncCallable[None] = Field( - default=APIOperation( - method="PATCH", - endpoint_template="/workspaces/{id}", - response_model=None, - ), - exclude=True, - ) - - delete_op: AsyncCallable[None] = Field( - default=APIOperation( - method="DELETE", - endpoint_template="/workspaces/{id}", - response_model=None, - ), - exclude=True, - ) - - get_status_op: AsyncCallable[WorkspaceStatus] = Field( - default=APIOperation( - method="GET", - endpoint_template="/workspaces/{id}/status", - response_model=WorkspaceStatus, - ), - exclude=True, - ) - - execute_command_op: AsyncCallable[CommandOutput] = Field( - default=APIOperation( - method="POST", - endpoint_template="/workspaces/{id}/execute", - input_model=CommandInput, - response_model=CommandOutput, - ), - exclude=True, - ) - - async def update(self, data: WorkspaceUpdate) -> None: - """ - Updates this workspace with new data and refreshes the - local object state. - - Args: - data (WorkspaceUpdate): The payload with fields to update. - """ - await self.update_op(data=data) - update_data = data.model_dump(exclude_unset=True) - log.debug(f"Updating local workspace state (id={self.id}) with: {update_data}") - for key, value in update_data.items(): - setattr(self, key, value) - - async def delete(self) -> None: - """Deletes this workspace.""" - await self.delete_op() - - async def get_status(self) -> WorkspaceStatus: - """Gets the running status of this workspace.""" - return await self.get_status_op() - - async def execute_command( - self, command: str, env: Optional[Dict[str, str]] = None - ) -> CommandOutput: - """ - Führt einen Befehl in diesem Workspace aus. - - Args: - command (str): Der Bash-Befehl (z.B. "ls -la"). - env (Dict[str, str], optional): Env Vars, die nur - für diesen Befehl gesetzt werden. - - Returns: - CommandOutput: Ein Objekt mit stdout und stderr. - """ - command_data = CommandInput(command=command, env=env) - return await self.execute_command_op(data=command_data) - - @cached_property - def env_vars(self) -> WorkspaceEnvVarManager: - """ - Provides access to the Environment Variable manager for this workspace. - - Usage: - >>> await workspace.env_vars.get() - >>> await workspace.env_vars.set([{"name": "KEY", "value": "VALUE"}]) - """ - if not self._http_client: - raise RuntimeError("Cannot access 'env_vars' on a detached model.") - return WorkspaceEnvVarManager( - http_client=self._http_client, workspace_id=self.id - ) diff --git a/src/codesphere/resources/workspace/operations.py b/src/codesphere/resources/workspace/operations.py new file mode 100644 index 0000000..0c0ab84 --- /dev/null +++ b/src/codesphere/resources/workspace/operations.py @@ -0,0 +1,53 @@ +from .schemas import ( + CommandInput, + CommandOutput, + Workspace, + WorkspaceCreate, + WorkspaceStatus, +) +from ...core.base import ResourceList +from ...core.operations import APIOperation + +_LIST_BY_TEAM_OP = APIOperation( + method="GET", + endpoint_template="/workspaces/team/{team_id}", + response_model=ResourceList[Workspace], +) + +_GET_OP = APIOperation( + method="GET", + endpoint_template="/workspaces/{workspace_id}", + response_model=Workspace, +) + +_CREATE_OP = APIOperation( + method="POST", + endpoint_template="/workspaces", + input_model=WorkspaceCreate, + response_model=Workspace, +) + +_UPDATE_OP = APIOperation( + method="PATCH", + endpoint_template="/workspaces/{id}", + response_model=type(None), +) + +_DELETE_OP = APIOperation( + method="DELETE", + endpoint_template="/workspaces/{id}", + response_model=type(None), +) + +_GET_STATUS_OP = APIOperation( + method="GET", + endpoint_template="/workspaces/{id}/status", + response_model=WorkspaceStatus, +) + +_EXECUTE_COMMAND_OP = APIOperation( + method="POST", + endpoint_template="/workspaces/{id}/execute", + input_model=CommandInput, + response_model=CommandOutput, +) diff --git a/src/codesphere/resources/workspace/resources.py b/src/codesphere/resources/workspace/resources.py index cba51b0..1662a51 100644 --- a/src/codesphere/resources/workspace/resources.py +++ b/src/codesphere/resources/workspace/resources.py @@ -1,69 +1,33 @@ -from typing import List, Protocol +from typing import List +from pydantic import Field -from ...core import APIOperation, ResourceBase -from .models import Workspace, WorkspaceCreate +from ...core.base import ResourceList +from ...core.operations import AsyncCallable +from .operations import ( + _CREATE_OP, + _GET_OP, + _LIST_BY_TEAM_OP, +) - -class ListWorkspacesCallable(Protocol): - async def __call__(self, *, team_id: int) -> List[Workspace]: ... - - -class GetWorkspaceCallable(Protocol): - async def __call__(self, *, workspace_id: int) -> Workspace: ... - - -class CreateWorkspaceCallable(Protocol): - async def __call__(self, *, data: WorkspaceCreate) -> Workspace: ... +from ...core import ResourceBase +from .schemas import Workspace, WorkspaceCreate class WorkspacesResource(ResourceBase): - """Manages all API operations for the Workspace resource.""" - - list_by_team: ListWorkspacesCallable - """ - Lists all workspaces for a specific team. - - Args: - team_id (int): The unique identifier for the team. - - Returns: - List[Workspace]: A list of Workspace objects associated with the team. - """ - list_by_team = APIOperation( - method="GET", - endpoint_template="/workspaces/team/{team_id}", - response_model=List[Workspace], + list_by_team_op: AsyncCallable[ResourceList[Workspace]] = Field( + default=_LIST_BY_TEAM_OP, exclude=True ) - get: GetWorkspaceCallable - """ - Fetches a single workspace by its ID. + async def list(self, team_id: int) -> List[Workspace]: + result = await self.list_by_team_op(data=team_id) + return result.root - Args: - workspace_id (int): The unique identifier for the workspace. + get_op: AsyncCallable[Workspace] = Field(default=_GET_OP, exclude=True) - Returns: - Workspace: The requested Workspace object. - """ - get = APIOperation( - method="GET", - endpoint_template="/workspaces/{workspace_id}", - response_model=Workspace, - ) + async def get(self, workspace_id: int) -> Workspace: + return await self.get_op(data=workspace_id) - create: CreateWorkspaceCallable - """ - Creates a new workspace. + create_op: AsyncCallable[Workspace] = Field(default=_CREATE_OP, exclude=True) - Args: - data (WorkspaceCreate): The data payload for the new workspace. - - Returns: - Workspace: The newly created Workspace object. - """ - create = APIOperation( - method="POST", - endpoint_template="/workspaces", - input_model=WorkspaceCreate, - response_model=Workspace, - ) + async def create(self, payload=WorkspaceCreate) -> Workspace: + return await self.create_op(data=payload) diff --git a/src/codesphere/resources/workspace/schemas.py b/src/codesphere/resources/workspace/schemas.py new file mode 100644 index 0000000..de17628 --- /dev/null +++ b/src/codesphere/resources/workspace/schemas.py @@ -0,0 +1,119 @@ +from __future__ import annotations +from functools import cached_property +import logging +from pydantic import Field +from typing import Dict, Optional, List + +from .operations import _DELETE_OP, _EXECUTE_COMMAND_OP, _GET_STATUS_OP, _UPDATE_OP +from .envVars import EnvVar, WorkspaceEnvVarManager +from ...core.base import CamelModel +from ...core import _APIOperationExecutor, AsyncCallable +from ...utils import update_model_fields + +log = logging.getLogger(__name__) + + +class WorkspaceCreate(CamelModel): + team_id: int + name: str + plan_id: int + base_image: Optional[str] = None + is_private_repo: bool = True + replicas: int = 1 + git_url: Optional[str] = None + initial_branch: Optional[str] = None + clone_depth: Optional[int] = None + source_workspace_id: Optional[int] = None + welcome_message: Optional[str] = None + vpn_config: Optional[str] = None + restricted: Optional[bool] = None + env: Optional[List[EnvVar]] = None + + +class WorkspaceBase(CamelModel): + id: int + team_id: int + name: str + plan_id: int + is_private_repo: bool + replicas: int + base_image: Optional[str] = None + data_center_id: int + user_id: int + git_url: Optional[str] = None + initial_branch: Optional[str] = None + source_workspace_id: Optional[int] = None + welcome_message: Optional[str] = None + vpn_config: Optional[str] = None + restricted: bool + + +class WorkspaceUpdate(CamelModel): + plan_id: Optional[int] = None + base_image: Optional[str] = None + name: Optional[str] = None + replicas: Optional[int] = None + vpn_config: Optional[str] = None + restricted: Optional[bool] = None + + +class WorkspaceStatus(CamelModel): + is_running: bool + + +class CommandInput(CamelModel): + command: str + env: Optional[Dict[str, str]] = None + + +class CommandOutput(CamelModel): + command: str + working_dir: str + output: str + error: str + + +class Workspace(WorkspaceBase, _APIOperationExecutor): + update_op: AsyncCallable[None] = Field( + default=_UPDATE_OP, + exclude=True, + ) + + async def update(self, data: WorkspaceUpdate) -> None: + await self.update_op(data=data) + update_model_fields(target=self, source=data) + + delete_op: AsyncCallable[None] = Field( + default=_DELETE_OP, + exclude=True, + ) + + async def delete(self) -> None: + await self.delete_op() + + get_status_op: AsyncCallable[WorkspaceStatus] = Field( + default=_GET_STATUS_OP, + exclude=True, + ) + + async def get_status(self) -> WorkspaceStatus: + return await self.get_status_op() + + execute_command_op: AsyncCallable[CommandOutput] = Field( + default=_EXECUTE_COMMAND_OP, + exclude=True, + ) + + async def execute_command( + self, command: str, env: Optional[Dict[str, str]] = None + ) -> CommandOutput: + command_data = CommandInput(command=command, env=env) + return await self.execute_command_op(data=command_data) + + @cached_property + def env_vars(self) -> WorkspaceEnvVarManager: + if not self._http_client: + raise RuntimeError("Cannot access 'env_vars' on a detached model.") + return WorkspaceEnvVarManager( + http_client=self._http_client, workspace_id=self.id + ) diff --git a/src/codesphere/utils.py b/src/codesphere/utils.py new file mode 100644 index 0000000..baa6024 --- /dev/null +++ b/src/codesphere/utils.py @@ -0,0 +1,45 @@ +import logging +from pydantic import BaseModel +from typing import Any, Dict, List, Type, TypeVar + +log = logging.getLogger(__name__) + +T = TypeVar("T", bound=BaseModel) + + +def update_model_fields(target: BaseModel, source: BaseModel) -> None: + if log.isEnabledFor(logging.DEBUG): + debug_dump = source.model_dump(exclude_unset=True) + log.debug(f"Updating {target.__class__.__name__} with data: {debug_dump}") + + for field_name in source.model_fields_set: + value = getattr(source, field_name) + setattr(target, field_name, value) + + +def dict_to_model_list( + data: Dict[Any, Any], + model_cls: Type[T], + key_field: str = None, + value_field: str = None, +) -> List[T]: + if key_field is None or value_field is None: + for name, field_info in model_cls.model_fields.items(): + if field_info.json_schema_extra: + if field_info.json_schema_extra.get("is_dict_key"): + key_field = name + elif field_info.json_schema_extra.get("is_dict_value"): + value_field = name + + if not key_field or not value_field: + raise ValueError( + f"Could not determine key/value mapping for {model_cls.__name__}. " + "Please explicitly pass key_field/value_field OR mark fields in the model." + ) + + items = [] + for key, value in data.items(): + item = model_cls(**{key_field: key, value_field: value}) + items.append(item) + + return items