MON-199535 Factorize Tools Implementation

centreon_mcp/components/acknowledgement.py

@@ -3,13 +3,14 @@
 from fastmcp import FastMCP
 from pydantic import Field
 
+from centreon_mcp.components.base import _list
 from centreon_mcp.types.acknowledgement import (
     Acknowledgement,
     AcknowledgementParams,
     AcknowledgementResource,
 )
 from centreon_mcp.utils import logger
-from centreon_mcp.utils.base import BaseFilter, BaseOrder, _list
+from centreon_mcp.utils.base import BaseFilter, BaseOrder
 
 acknowledgement = FastMCP()
 
@@ -53,7 +54,7 @@ async def list_acknowledgements(
     List all acknowledgements in real-time monitoring.
     """
     logger.info("Executing tool list_acknowledgements")
-    return await _list(Acknowledgement, AcknowledgementOrder, filters, limit, page, order)
+    return await _list(Acknowledgement, filters, limit, page, order)
 
 
 @acknowledgement.tool(

centreon_mcp/components/base.py

@@ -0,0 +1,72 @@
+import asyncio
+import json
+from collections.abc import Sequence
+from typing import Any
+
+from pydantic import BaseModel
+
+from centreon_mcp.utils.base import BaseFilter, BaseOrder
+from centreon_mcp.utils.mixins import (
+    CreateMixin,
+    DeleteMixin,
+    ListMixin,
+    PatchMixin,
+    UpdateMixin,
+)
+
+
+async def _list[CentreonModel: ListMixin](
+    model: type[CentreonModel],
+    filters: Sequence[BaseFilter] | None = None,
+    limit: int = 10,
+    page: int = 1,
+    order: BaseOrder | None = None,
+    extras: dict[str, Any] | None = None,
+) -> list[CentreonModel]:
+    """
+    Generic function to list resources based on provided filters, pagination and order.
+    """
+    search = json.dumps(BaseFilter.join(filters))
+    sort_by = order.model_dump_json() if order else None
+    return await model.list(search, limit, page, sort_by, extras)
+
+
+async def _delete[CentreonModel: DeleteMixin](
+    model: type[CentreonModel], model_ids: list[int]
+) -> dict[int, bool | BaseException]:
+    """
+    Generic function to delete multiple resources based on their ids.
+    """
+    tasks = [asyncio.create_task(model.delete(model_id)) for model_id in model_ids]
+    results = await asyncio.gather(*tasks, return_exceptions=True)
+    return dict(zip(model_ids, results, strict=True))
+
+
+async def _create[CentreonModel: CreateMixin](
+    model: type[CentreonModel], params: BaseModel
+) -> bool:
+    """
+    Generic function to create a resource based on params.
+    """
+    return await model.create(params)
+
+
+async def _patch[CentreonModel: PatchMixin](
+    model: type[CentreonModel], model_id: int, params: BaseModel
+) -> bool:
+    """
+    Generic function to patch a resource based on params.
+    """
+    return await model.patch(model_id, params)
+
+
+async def _update[CentreonModel: UpdateMixin, FullParams: BaseModel](
+    model: type[CentreonModel], full_params_cls: type[FullParams], model_id: int, params: BaseModel
+) -> bool:
+    """
+    Generic function to update a resource from params.
+    """
+    current = await model.get(model_id)
+    data = current.model_dump(exclude={"id"}, exclude_none=True)  # type: ignore
+    data |= params.model_dump(exclude_none=True)
+    return await model.update(model_id, full_params_cls(**data))

centreon_mcp/components/command.py

@@ -3,9 +3,10 @@
 from fastmcp import FastMCP
 from pydantic import Field
 
+from centreon_mcp.components.base import _list
 from centreon_mcp.types.command import Command, CommandParams, CommandType
 from centreon_mcp.utils import logger
-from centreon_mcp.utils.base import BaseFilter, BaseOrder, _list
+from centreon_mcp.utils.base import BaseFilter, BaseOrder
 
 command = FastMCP()
 
@@ -42,7 +43,7 @@ async def list_commands(
     to avoid retrieving all commands except if explicitly intended.
     """
     logger.info("Executing tool list_commands")
-    return await _list(Command, CommandOrder, filters, limit, page, order)
+    return await _list(Command, filters, limit, page, order)
 
 
 @command.tool(

centreon_mcp/components/downtime.py

@@ -1,13 +1,13 @@
-import asyncio
 from typing import Annotated, Literal
 
 from fastmcp import FastMCP
 from pydantic import Field
 
+from centreon_mcp.components.base import _delete, _list
 from centreon_mcp.types.downtime import Downtime, DowntimeParams, DowntimeResource
 from centreon_mcp.types.host import HostState
 from centreon_mcp.utils import logger
-from centreon_mcp.utils.base import BaseFilter, BaseOrder, _list
+from centreon_mcp.utils.base import BaseFilter, BaseOrder
 
 downtime = FastMCP()
 
@@ -59,7 +59,7 @@ async def list_downtimes(
     to avoid retrieving all downtimes except if explicitly intended.
     """
     logger.info("Executing tool list_downtimes")
-    return await _list(Downtime, DowntimeOrder, filters, limit, page, order)
+    return await _list(Downtime, filters, limit, page, order)
 
 
 @downtime.tool(
@@ -95,6 +95,4 @@ async def cancel_downtimes(downtime_ids: list[int]) -> dict[int, bool | BaseExce
     Use tools `list_downtimes` first to get downtime IDs.
     """
     logger.info("Executing tool cancel_downtimes")
-    tasks = [asyncio.create_task(Downtime.delete(downtime_id)) for downtime_id in downtime_ids]
-    results = await asyncio.gather(*tasks, return_exceptions=True)
-    return dict(zip(downtime_ids, results, strict=True))
+    return await _delete(Downtime, downtime_ids)

centreon_mcp/components/host.py

@@ -1,10 +1,10 @@
-import asyncio
 import json
 from typing import Annotated, Literal
 
 from fastmcp import FastMCP
 from pydantic import Field
 
+from centreon_mcp.components.base import _create, _delete, _list, _patch
 from centreon_mcp.types.host import (
     Host,
     HostConfiguration,
@@ -13,7 +13,7 @@
     HostStatusCount,
 )
 from centreon_mcp.utils import logger
-from centreon_mcp.utils.base import BaseFilter, BaseOrder, _list
+from centreon_mcp.utils.base import BaseFilter, BaseOrder
 
 host = FastMCP()
 
@@ -87,7 +87,7 @@ async def list_host_configurations(
     to avoid retrieving all host configurations except if explicitly intended.
     """
     logger.info("Executing tool list_host_configurations")
-    return await _list(HostConfiguration, HostConfigurationOrder, filters, limit, page, order)
+    return await _list(HostConfiguration, filters, limit, page, order)
 
 
 @host.tool(
@@ -104,7 +104,7 @@ async def create_host_configuration(params: HostConfigurationFullParams) -> bool
     Create a host configuration from params.
     """
     logger.info("Executing tool create_host_configuration")
-    return await HostConfiguration.create(params)
+    return await _create(HostConfiguration, params)
 
 
 @host.tool(
@@ -121,7 +121,7 @@ async def update_host_configuration(host_id: int, params: HostConfigurationParti
     Update a host configuration from params.
     """
     logger.info("Executing tool update_host_configuration")
-    return await HostConfiguration.patch(host_id, params)
+    return await _patch(HostConfiguration, host_id, params)
 
 
 @host.tool(
@@ -138,6 +138,4 @@ async def delete_host_configurations(host_ids: list[int]) -> dict[int, bool | Ba
     Delete multiple host configurations.
     """
     logger.info("Executing tool delete_host_configurations")
-    tasks = [asyncio.create_task(HostConfiguration.delete(host_id)) for host_id in host_ids]
-    results = await asyncio.gather(*tasks, return_exceptions=True)
-    return dict(zip(host_ids, results, strict=True))
+    return await _delete(HostConfiguration, host_ids)

centreon_mcp/components/host_category.py

@@ -1,16 +1,16 @@
-import asyncio
 from typing import Annotated, Literal
 
 from fastmcp import FastMCP
 from pydantic import Field
 
+from centreon_mcp.components.base import _create, _delete, _list, _update
 from centreon_mcp.types.host_category import (
     HostCategoryConfiguration,
     HostCategoryConfigurationFullParams,
     HostCategoryConfigurationPartialParams,
 )
 from centreon_mcp.utils import logger
-from centreon_mcp.utils.base import BaseFilter, BaseOrder, _list
+from centreon_mcp.utils.base import BaseFilter, BaseOrder
 
 host_category = FastMCP()
 
@@ -47,9 +47,7 @@ async def list_host_category_configurations(
     to avoid retrieving all host categories except if explicitly intended.
     """
     logger.info("Executing tool list_host_category_configurations")
-    return await _list(
-        HostCategoryConfiguration, HostCategoryConfigurationOrder, filters, limit, page, order
-    )
+    return await _list(HostCategoryConfiguration, filters, limit, page, order)
 
 
 @host_category.tool(
@@ -66,7 +64,7 @@ async def create_host_category_configuration(params: HostCategoryConfigurationFu
     Create a host category configuration.
     """
     logger.info("Executing tool create_host_category_configuration")
-    return await HostCategoryConfiguration.create(params)
+    return await _create(HostCategoryConfiguration, params)
 
 
 @host_category.tool(
@@ -85,11 +83,8 @@ async def update_host_category_configuration(
     Update a host category from params.
     """
     logger.info("Executing tool update_host_category_configuration")
-    host_category = await HostCategoryConfiguration.get(host_category_id)
-    data = host_category.model_dump(exclude={"id"}, exclude_none=True)
-    data |= params.model_dump(exclude_none=True)
-    return await HostCategoryConfiguration.update(
-        host_category_id, HostCategoryConfigurationFullParams(**data)
+    return await _update(
+        HostCategoryConfiguration, HostCategoryConfigurationFullParams, host_category_id, params
     )
 
 
@@ -109,9 +104,4 @@ async def delete_host_category_configurations(
     Delete multiple host category configurations.
     """
     logger.info("Executing tool delete_host_category_configurations")
-    tasks = [
-        asyncio.create_task(HostCategoryConfiguration.delete(host_category_id))
-        for host_category_id in host_category_ids
-    ]
-    results = await asyncio.gather(*tasks, return_exceptions=True)
-    return dict(zip(host_category_ids, results, strict=True))
+    return await _delete(HostCategoryConfiguration, host_category_ids)

centreon_mcp/components/host_group.py

@@ -1,9 +1,9 @@
-import asyncio
 from typing import Annotated, Literal
 
 from fastmcp import FastMCP
 from pydantic import Field
 
+from centreon_mcp.components.base import _create, _delete, _list, _update
 from centreon_mcp.types.host import HostState
 from centreon_mcp.types.host_group import (
     HostGroup,
@@ -12,7 +12,7 @@
     HostGroupConfigurationPartialParams,
 )
 from centreon_mcp.utils import logger
-from centreon_mcp.utils.base import BaseFilter, BaseOrder, _list
+from centreon_mcp.utils.base import BaseFilter, BaseOrder
 
 host_group = FastMCP()
 
@@ -53,7 +53,7 @@ async def list_host_groups(
     to avoid retrieving all host groups except if explicitly intended.
     """
     logger.info("Executing tool list_host_groups")
-    return await _list(HostGroup, HostGroupOrder, filters, limit, page, order)
+    return await _list(HostGroup, filters, limit, page, order)
 
 
 class HostGroupConfigurationOrder(BaseOrder):
@@ -88,9 +88,7 @@ async def list_host_group_configurations(
     to avoid retrieving all host groups except if explicitly intended.
     """
     logger.info("Executing tool list_hostgroup_configurations")
-    return await _list(
-        HostGroupConfiguration, HostGroupConfigurationOrder, filters, limit, page, order
-    )
+    return await _list(HostGroupConfiguration, filters, limit, page, order)
 
 
 @host_group.tool(
@@ -107,7 +105,7 @@ async def create_host_group_configuration(params: HostGroupConfigurationFullPara
     Create a hostgroup.
     """
     logger.info("Executing tool create_hostgroup_configuration")
-    return await HostGroupConfiguration.create(params)
+    return await _create(HostGroupConfiguration, params)
 
 
 @host_group.tool(
@@ -123,17 +121,11 @@ async def update_host_group_configuration(
     host_group_id: int, params: HostGroupConfigurationPartialParams
 ) -> bool:
     """
-    Update a host group from params.
+    Update a host group from params. Just need to get host_group_id first.
     """
     logger.info("Executing tool update_host_group_configuration")
-    hostgroup = await HostGroupConfiguration.get(host_group_id)
-    data = hostgroup.model_dump(exclude={"id", "is_activated", "icon", "hosts"}, exclude_none=True)
-    data["icon_id"] = hostgroup.icon.id if hostgroup.icon else None
-    data["hosts"] = [host.id for host in hostgroup.hosts if host.id not in params.hosts_removed]
-    data["hosts"] += [host_id for host_id in params.hosts_added if host_id not in data["hosts"]]
-    data |= params.model_dump(exclude_none=True)
-    return await HostGroupConfiguration.update(
-        host_group_id, HostGroupConfigurationFullParams(**data)
+    return await _update(
+        HostGroupConfiguration, HostGroupConfigurationFullParams, host_group_id, params
     )
 
 
@@ -153,9 +145,4 @@ async def delete_host_group_configurations(
     Delete multiple host group configurations.
     """
     logger.info("Executing tool delete_host_group_configurations")
-    tasks = [
-        asyncio.create_task(HostGroupConfiguration.delete(hostgroup_id))
-        for hostgroup_id in hostgroup_ids
-    ]
-    results = await asyncio.gather(*tasks, return_exceptions=True)
-    return dict(zip(hostgroup_ids, results, strict=True))
+    return await _delete(HostGroupConfiguration, hostgroup_ids)