feat: make BaseService methods async

src/aignostics/CLAUDE.md

@@ -294,10 +294,10 @@ For detailed information about each module, see:
  from aignostics.utils import BaseService
 
  class Service(BaseService):
- def health(self) -> Health:
+async def health(self) -> Health:
  return Health(status=Health.Code.UP)
 
- def info(self, mask_secrets=True) -> dict:
+async def info(self, mask_secrets=True) -> dict:
  return {"version": "1.0.0"}
  ```
 

src/aignostics/application/_cli.py

@@ -1,5 +1,6 @@
 """CLI of application module."""
 
+import asyncio
 import json
 import sys
 import time
@@ -129,7 +130,7 @@
 
 
 def _abort_if_system_unhealthy() -> None:
- health = SystemService.health_static()
+ health = asyncio.run(SystemService.health_static())
  if not health:
  logger.error(f"Platform is not healthy: {health.reason}. Aborting.")
  console.print(f"[error]Error:[/error] Platform is not healthy: {health.reason}. Aborting.")

src/aignostics/application/_gui/_page_application_describe.py

@@ -107,6 +107,7 @@ async def _page_application_describe(application_id: str) -> None: # noqa: C901
  type="positive",
  )
  spinner.set_visibility(False)
+ system_healthy = bool(await SystemService.health_static())
 
  if application is None:
  await _frame(
@@ -301,8 +302,6 @@ def _add_application_version_selection_section() -> None:
  with ui.column(), ui.button(icon="info", on_click=info_dialog.open):
  ui.tooltip("Show changes and input/ouput schema of this application version.")
  with ui.stepper_navigation():
- # Check system health and determine if force option should be available
- system_healthy = bool(SystemService.health_static())
  unhealthy_tooltip = None
  with ui.button(
  "Next",

src/aignostics/application/_service.py

@@ -73,7 +73,7 @@ def __init__(self) -> None:
  """Initialize service."""
  super().__init__(Settings) # automatically loads and validates the settings
 
- def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PLR6301
+async def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PLR6301
  """Determine info of this service.
 
  Args:
@@ -84,7 +84,7 @@ def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PL
  """
  return {}
 
- def health(self) -> Health: # noqa: PLR6301
+async def health(self) -> Health: # noqa: PLR6301
  """Determine health of this service.
 
  Returns:

src/aignostics/bucket/_service.py

@@ -93,7 +93,7 @@ def __init__(self) -> None:
  super().__init__(Settings)
  self._platform_service = PlatformService()
 
- def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PLR6301
+async def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PLR6301
  """Determine info of this service.
 
  Args:
@@ -104,7 +104,7 @@ def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PL
  """
  return {}
 
- def health(self) -> Health: # noqa: PLR6301
+async def health(self) -> Health: # noqa: PLR6301
  """Determine health of this service.
 
  Returns:

src/aignostics/dataset/_service.py

@@ -62,7 +62,7 @@ def _cleanup_processes() -> None:
 class Service(BaseService):
  """Service of the IDC module."""
 
- def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PLR6301
+async def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PLR6301
  """Determine info of this service.
 
  Args:
@@ -73,7 +73,7 @@ def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PL
  """
  return {}
 
- def health(self) -> Health: # noqa: PLR6301
+async def health(self) -> Health: # noqa: PLR6301
  """Determine health of hello service.
 
  Returns:

src/aignostics/gui/_frame.py

@@ -204,7 +204,7 @@ def health_link() -> None:
  async def _health_load_and_render() -> None:
  nonlocal launchpad_healthy
  with contextlib.suppress(Exception):
- launchpad_healthy = bool(await run.cpu_bound(SystemService.health_static))
+ launchpad_healthy = bool(await SystemService.health_static())
  health_icon.refresh()
  health_link.refresh()
 

src/aignostics/notebook/_service.py

@@ -245,7 +245,7 @@ def _get_runner() -> _Runner:
 class Service(BaseService):
  """Service of the Marimo module."""
 
- def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PLR6301
+async def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PLR6301
  """Determine info of this service.
 
  Args:
@@ -256,7 +256,7 @@ def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PL
  """
  return {}
 
- def health(self) -> Health: # noqa: PLR6301
+async def health(self) -> Health: # noqa: PLR6301
  """Determine health of hello service.
 
  Returns:

src/aignostics/platform/_service.py

@@ -6,7 +6,7 @@
 from http import HTTPStatus
 from typing import Any
 
-import urllib3
+import httpx
 from aignx.codegen.models import MeReadResponse as Me
 from aignx.codegen.models import OrganizationReadResponse as Organization
 from aignx.codegen.models import UserReadResponse as User
@@ -149,29 +149,12 @@ class Service(BaseService):
  """Service of the application module."""
 
  _settings: Settings
- _http_pool: urllib3.PoolManager | None = None
 
  def __init__(self) -> None:
  """Initialize service."""
  super().__init__(Settings) # automatically loads and validates the settings
 
- @classmethod
- def _get_http_pool(cls) -> urllib3.PoolManager:
- """Get or create the shared HTTP pool manager.
-
- All service instances share the same urllib3.PoolManager for efficient connection reuse.
-
- Returns:
- urllib3.PoolManager: Shared connection pool manager.
- """
- if cls._http_pool is None:
- cls._http_pool = urllib3.PoolManager(
- maxsize=10, # Max connections per host
- block=False, # Don't block if pool is full
- )
- return cls._http_pool
-
- def info(self, mask_secrets: bool = True) -> dict[str, Any]:
+ async def info(self, mask_secrets: bool = True) -> dict[str, Any]:
  """Determine info of this service.
 
  Args:
@@ -193,26 +176,26 @@ def info(self, mask_secrets: bool = True) -> dict[str, Any]:
  }
 
  @staticmethod
- def _health_from_response(response: urllib3.BaseHTTPResponse) -> Health:
+ def _health_from_response(response: httpx.Response) -> Health:
  """Map a PAPI health response to a Health status.
 
  Handles non-200 status codes, unparseable bodies, and the three recognised
  ``status`` values (``"UP"``, ``"DEGRADED"``, ``"DOWN"``).
 
  Args:
- response: urllib3 response from the ``/health`` endpoint.
+ response: httpx response from the ``/health`` endpoint.
 
  Returns:
  Health: ``UP``, ``DEGRADED``, or ``DOWN`` derived from the response.
  """
- if response.status != HTTPStatus.OK:
- logger.error("Aignostics Platform API returned '{}'", response.status)
+ if response.status_code != HTTPStatus.OK:
+ logger.error("Aignostics Platform API returned '{}'", response.status_code)
  return Health(
- status=Health.Code.DOWN, reason=f"Aignostics Platform API returned status '{response.status}'"
+ status=Health.Code.DOWN, reason=f"Aignostics Platform API returned status '{response.status_code}'"
  )
 
  try:
- body = json.loads(response.data)
+ body = response.json()
  except Exception:
  return Health(status=Health.Code.DOWN, reason="Aignostics Platform API returned unparseable response")
 
@@ -228,57 +211,53 @@ def _health_from_response(response: urllib3.BaseHTTPResponse) -> Health:
  reason=f"Aignostics Platform API returned unknown status '{api_status}'",
  )
 
- def _determine_api_public_health(self) -> Health:
+async def _determine_api_public_health(self) -> Health:
  """Determine healthiness and reachability of Aignostics Platform API.
 
  - Checks if health endpoint is reachable and returns 200 OK
  - Parses the response body to detect DEGRADED status
- - Uses urllib3 for a direct connection check without authentication
+ - Uses httpx for a direct connection check without authentication
 
  Returns:
  Health: The healthiness of the Aignostics Platform API via basic unauthenticated request.
  """
  try:
- http = self._get_http_pool()
- response = http.request(
- method="GET",
- url=f"{self._settings.api_root}/health",
- headers={"User-Agent": user_agent()},
- timeout=urllib3.Timeout(total=self._settings.health_timeout),
- )
+ async with httpx.AsyncClient() as client:
+ response = await client.get(
+ f"{self._settings.api_root}/health",
+ headers={"User-Agent": user_agent()},
+ timeout=self._settings.health_timeout,
+ )
  return self._health_from_response(response)
  except Exception as e:
  logger.exception("Issue with Aignostics Platform API")
  return Health(status=Health.Code.DOWN, reason=f"Issue with Aignostics Platform API: '{e}'")
 
- def _determine_api_authenticated_health(self) -> Health:
+async def _determine_api_authenticated_health(self) -> Health:
  """Determine healthiness and reachability of Aignostics Platform API via authenticated request.
 
- Uses a dedicated HTTP pool (separate from the API client's connection pool) to prevent
- connection-level cross-contamination between health checks and API calls.
  Parses the response body to detect DEGRADED status.
 
  Returns:
  Health: The healthiness of the Aignostics Platform API when trying to reach via authenticated request.
  """
  try:
  token = get_token(use_cache=True)
- http = self._get_http_pool()
- response = http.request(
- method="GET",
- url=f"{self._settings.api_root}/health",
- headers={
- "User-Agent": user_agent(),
- "Authorization": f"Bearer {token}",
- },
- timeout=urllib3.Timeout(total=self._settings.health_timeout),
- )
+ async with httpx.AsyncClient() as client:
+ response = await client.get(
+ f"{self._settings.api_root}/health",
+ headers={
+ "User-Agent": user_agent(),
+ "Authorization": f"Bearer {token}",
+ },
+ timeout=self._settings.health_timeout,
+ )
  return self._health_from_response(response)
  except Exception as e:
  logger.exception("Issue with Aignostics Platform API")
  return Health(status=Health.Code.DOWN, reason=f"Issue with Aignostics Platform API: '{e}'")
 
- def health(self) -> Health:
+async def health(self) -> Health:
  """Determine health of this service.
 
  Returns:
@@ -287,8 +266,8 @@ def health(self) -> Health:
  return Health(
  status=Health.Code.UP,
  components={
- "api_public": self._determine_api_public_health(),
- "api_authenticated": self._determine_api_authenticated_health(),
+ "api_public": await self._determine_api_public_health(),
+ "api_authenticated": await self._determine_api_authenticated_health(),
  },
  )
 

src/aignostics/qupath/_service.py

@@ -209,7 +209,7 @@ def __init__(self) -> None:
  """Initialize service."""
  super().__init__(Settings)
 
- def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PLR6301
+async def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PLR6301
  """Determine info of this service.
 
  Args:
@@ -228,7 +228,7 @@ def info(self, mask_secrets: bool = True) -> dict[str, Any]: # noqa: ARG002, PL
  }
  }
 
- def health(self) -> Health: # noqa: PLR6301
+async def health(self) -> Health: # noqa: PLR6301
  """Determine health of this service.
 
  Returns:

src/aignostics/system/_cli.py

@@ -1,5 +1,6 @@
 """System CLI commands."""
 
+import asyncio
 import json
 import sys
 from enum import StrEnum
@@ -51,7 +52,7 @@ def health(
  Args:
  output_format (OutputFormat): Output format (JSON or YAML).
  """
- health = _service.health()
+ health = asyncio.run(_service.health())
  match output_format:
  case OutputFormat.JSON:
  console.print_json(data=health.model_dump())
@@ -79,7 +80,7 @@ def info(
  mask_secrets (bool): Mask values for variables identified as secrets.
  output_format (OutputFormat): Output format (JSON or YAML).
  """
- info = _service.info(include_environ=include_environ, mask_secrets=mask_secrets)
+ info = asyncio.run(_service.info(include_environ=include_environ, mask_secrets=mask_secrets))
  match output_format:
  case OutputFormat.JSON:
  console.print_json(data=info)

src/aignostics/system/_gui.py

@@ -12,7 +12,7 @@
 class PageBuilder(BasePageBuilder):
  @staticmethod
  def register_pages() -> None: # noqa: PLR0915
- from nicegui import app, run, ui # noqa: PLC0415
+ from nicegui import app, ui # noqa: PLC0415
 
  locate_subclasses(BaseService) # Ensure settings are loaded
  app.add_static_files("/system_assets", Path(__file__).parent / "assets")
@@ -49,7 +49,7 @@ async def page_system() -> None: # noqa: PLR0915
  }
  editor = ui.json_editor(properties).style("width: 100%").mark("JSON_EDITOR_HEALTH")
  editor.set_visibility(False)
- health = await run.cpu_bound(Service.health_static)
+ health = await Service.health_static()
  if health is None:
  properties["content"] = {"json": "Health check failed."} # type: ignore[unreachable]
  else:
@@ -84,9 +84,7 @@ async def load_info(mask_secrets: bool = True) -> None:
  editor.set_visibility(False)
  spinner.set_visibility(True)
  mask_secrets_switch.set_visibility(False)
- info = await run.cpu_bound(
- Service.info, include_environ=True, mask_secrets=mask_secrets
- )
+ info = await Service.info(include_environ=True, mask_secrets=mask_secrets)
  if info is None:
  properties["content"] = {"json": "Info retrieval failed."} # type: ignore[unreachable]
  else: