coolipy.async_client

 1"""Asynchronous coolipy client."""
 2
 3from __future__ import annotations
 4
 5import httpx
 6
 7from coolipy._http import AsyncTransport, build_base_url, parse_response
 8from coolipy._response import CoolipyAPIResponse
 9from coolipy.exceptions import CoolipyConfigError
10from coolipy.models.system import SystemMessage
11from coolipy.resources.applications import AsyncApplications
12from coolipy.resources.databases import AsyncDatabases
13from coolipy.resources.deployments import AsyncDeployments
14from coolipy.resources.projects import AsyncProjects
15from coolipy.resources.s3_storages import AsyncS3Storages
16from coolipy.resources.security import AsyncSecurity
17from coolipy.resources.servers import AsyncServers
18from coolipy.resources.services import AsyncServices
19from coolipy.resources.tags import AsyncTags
20from coolipy.resources.teams import AsyncTeams
21
22
23class AsyncCoolipy:
24    """Asynchronous client for the Coolify API.
25
26    Args:
27        coolify_api_key: Bearer token used to authenticate with the Coolify API.
28        coolify_endpoint: Hostname or IP of the Coolify instance.
29        coolify_port: Port of the Coolify instance (ignored when ``omit_port``).
30        omit_port: When ``True``, build the base URL without a port.
31        http_protocol: ``"http"`` or ``"https"``.
32        timeout: Request timeout in seconds.
33        transport: Optional httpx async transport override (mainly for testing).
34    """
35
36    def __init__(
37        self,
38        coolify_api_key: str,
39        coolify_endpoint: str,
40        coolify_port: int = 8000,
41        omit_port: bool = False,
42        http_protocol: str = "http",
43        timeout: float = 30.0,
44        transport: httpx.AsyncBaseTransport | None = None,
45    ) -> None:
46        if not coolify_api_key:
47            raise CoolipyConfigError("coolify_api_key must be a non-empty string.")
48        base_url = build_base_url(http_protocol, coolify_endpoint, coolify_port, omit_port)
49        self._transport = AsyncTransport(
50            base_url, coolify_api_key, timeout=timeout, transport=transport
51        )
52        self.applications = AsyncApplications(self._transport)
53        self.databases = AsyncDatabases(self._transport)
54        self.services = AsyncServices(self._transport)
55        self.servers = AsyncServers(self._transport)
56        self.projects = AsyncProjects(self._transport)
57        self.teams = AsyncTeams(self._transport)
58        self.deployments = AsyncDeployments(self._transport)
59        self.tags = AsyncTags(self._transport)
60        self.s3_storages = AsyncS3Storages(self._transport)
61        self.security = AsyncSecurity(self._transport)
62
63    async def version(self) -> CoolipyAPIResponse[str]:
64        """Get the Coolify version."""
65        return parse_response(await self._transport.request("GET", "/version"), str)
66
67    async def health(self) -> CoolipyAPIResponse[str]:
68        """Run the Coolify healthcheck."""
69        return parse_response(await self._transport.request("GET", "/health"), str)
70
71    async def enable_api(self) -> CoolipyAPIResponse[SystemMessage]:
72        """Enable the Coolify API (requires root permissions)."""
73        return parse_response(await self._transport.request("POST", "/enable"), SystemMessage)
74
75    async def disable_api(self) -> CoolipyAPIResponse[SystemMessage]:
76        """Disable the Coolify API (requires root permissions)."""
77        return parse_response(await self._transport.request("POST", "/disable"), SystemMessage)
78
79    async def enable_mcp(self) -> CoolipyAPIResponse[SystemMessage]:
80        """Enable the Coolify MCP server (requires root permissions)."""
81        return parse_response(await self._transport.request("POST", "/mcp/enable"), SystemMessage)
82
83    async def disable_mcp(self) -> CoolipyAPIResponse[SystemMessage]:
84        """Disable the Coolify MCP server (requires root permissions)."""
85        return parse_response(await self._transport.request("POST", "/mcp/disable"), SystemMessage)
86
87    async def close(self) -> None:
88        """Close the underlying HTTP client."""
89        await self._transport.close()
90
91    async def __aenter__(self) -> AsyncCoolipy:
92        return self
93
94    async def __aexit__(self, *exc: object) -> None:
95        await self.close()