services: add stub classes

invenio_checks/services/__init__.py

@@ -0,0 +1,18 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2025 CERN.
+#
+# Invenio-Checks is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Checks services."""
+
+from .config import ChecksConfigServiceConfig
+from .schema import CheckConfigSchema
+from .services import CheckConfigService
+
+__all__ = (
+    "CheckConfigSchema",
+    "CheckConfigService",
+    "ChecksConfigServiceConfig",
+)

invenio_checks/services/config.py

@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2025 CERN.
+#
+# Invenio-Checks is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Checks services config."""
+
+from invenio_i18n import gettext as _
+from invenio_records_resources.services.base import ServiceConfig
+from invenio_records_resources.services.base.config import ConfiguratorMixin, FromConfig
+from invenio_records_resources.services.records.config import (
+    SearchOptions as SearchOptionsBase,
+)
+from sqlalchemy import asc, desc
+
+from ..models import CheckConfig
+from . import results
+from .permissions import CheckConfigPermissionPolicy
+from .schema import CheckConfigSchema
+
+
+class CheckConfigSearchOptions(SearchOptionsBase):
+    """Check config search options."""
+
+    sort_default = "title"
+    sort_direction_default = "asc"
+    sort_direction_options = {
+        "asc": dict(title=_("Ascending"), fn=asc),
+        "desc": dict(title=_("Descending"), fn=desc),
+    }
+    sort_options = {"title": dict(title=_("Title"), fields=["title"])}
+
+    pagination_options = {"default_results_per_page": 25}
+
+
+class ChecksConfigServiceConfig(ServiceConfig, ConfiguratorMixin):
+    """Checks config service configuration."""
+
+    service_id = "checks-config"
+
+    record_cls = CheckConfig
+    search = CheckConfigSearchOptions
+    schema = CheckConfigSchema
+
+    permission_policy_cls = FromConfig(
+        "CHECKS_PERMISSION_POLICY",
+        default=CheckConfigPermissionPolicy,
+    )
+
+    result_item_cls = results.Item
+    result_list_cls = results.List
+
+    links_item = {}

invenio_checks/services/permissions.py

@@ -0,0 +1,33 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2025 CERN.
+#
+# Invenio-Checks is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Checks permissions."""
+
+from invenio_administration.generators import Administration
+from invenio_records_permissions.generators import SystemProcess
+from invenio_records_permissions.policies import BasePermissionPolicy
+
+
+class CheckConfigPermissionPolicy(BasePermissionPolicy):
+    """Access control configuration for check configurations."""
+
+    can_search = [Administration(), SystemProcess()]
+    can_create = [Administration(), SystemProcess()]
+    can_read = [Administration(), SystemProcess()]
+    can_update = [Administration(), SystemProcess()]
+    can_delete = [Administration(), SystemProcess()]
+
+
+class CheckRunPermissionPolicy(BasePermissionPolicy):
+    """Access control configuration for check runs."""
+
+    can_search = [Administration(), SystemProcess()]
+    can_create = [Administration(), SystemProcess()]
+    can_read = [Administration(), SystemProcess()]
+    can_update = [Administration(), SystemProcess()]
+    can_delete = [Administration(), SystemProcess()]
+    can_stop = [Administration(), SystemProcess()]

invenio_checks/services/results.py

@@ -0,0 +1,78 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2025 CERN.
+#
+# Invenio-Checks is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Checks service results."""
+
+from collections.abc import Iterable, Sized
+
+from flask_sqlalchemy.pagination import Pagination
+from invenio_records_resources.services.records.results import (
+    RecordItem,
+    RecordList,
+)
+
+
+class Item(RecordItem):
+    """Single item result."""
+
+    @property
+    def id(self):
+        """Get the result id."""
+        return str(self._record.id)
+
+
+class List(RecordList):
+    """List result."""
+
+    @property
+    def items(self):
+        """Iterator over the items."""
+        if isinstance(self._results, Pagination):
+            return self._results.items
+        elif isinstance(self._results, Iterable):
+            return self._results
+        return self._results
+
+    @property
+    def total(self):
+        """Get total number of hits."""
+        if hasattr(self._results, "hits"):
+            return self._results.hits.total["value"]
+        if isinstance(self._results, Pagination):
+            return self._results.total
+        elif isinstance(self._results, Sized):
+            return len(self._results)
+        else:
+            return None
+
+    # TODO: See if we need to override this
+    @property
+    def aggregations(self):
+        """Get the search result aggregations."""
+        try:
+            return self._results.labelled_facets.to_dict()
+        except AttributeError:
+            return None
+
+    @property
+    def hits(self):
+        """Iterator over the hits."""
+        for hit in self.items:
+            # Project the hit
+            hit_dict = hit.dump()
+            hit_record = AttrDict(hit_dict)
+            projection = self._schema.dump(
+                hit_record,
+                context=dict(identity=self._identity, record=hit),
+            )
+            if self._links_item_tpl:
+                projection["links"] = self._links_item_tpl.expand(self._identity, hit)
+            if self._nested_links_item:
+                for link in self._nested_links_item:
+                    link.expand(self._identity, hit, projection)
+
+            yield projection

invenio_checks/services/schema.py

@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2025 CERN.
+#
+# Invenio-Checks is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Service schemas."""
+
+from datetime import timezone
+
+from marshmallow import EXCLUDE, Schema, fields
+from marshmallow_utils.fields import SanitizedUnicode, TZDateTime
+from marshmallow_utils.permissions import FieldPermissionsMixin
+
+from ..models import CheckRunStatus
+
+
+class CheckConfigSchema(Schema, FieldPermissionsMixin):
+    """Base schema for a check configuration."""
+
+    class Meta:
+        """Meta attributes for the schema."""
+
+        unknown = EXCLUDE
+
+    id = fields.UUID(dump_only=True)
+
+    created = TZDateTime(timezone=timezone.utc, format="iso", dump_only=True)
+    updated = TZDateTime(timezone=timezone.utc, format="iso", dump_only=True)
+
+    title = SanitizedUnicode(required=True)
+    description = SanitizedUnicode()
+
+    active = fields.Boolean(load_default=True)
+
+
+class CheckRunSchema(Schema, FieldPermissionsMixin):
+    """Base schema for a check run."""
+
+    class Meta:
+        """Meta attributes for the schema."""
+
+        unknown = EXCLUDE
+
+    id = fields.UUID(dump_only=True)
+
+    created = TZDateTime(timezone=timezone.utc, format="iso", dump_only=True)
+    updated = TZDateTime(timezone=timezone.utc, format="iso", dump_only=True)
+
+    started_at = TZDateTime(timezone=timezone.utc, format="iso", dump_only=True)
+    finished_at = TZDateTime(timezone=timezone.utc, format="iso", dump_only=True)
+
+    status = fields.Enum(CheckRunStatus, dump_only=True)
+    message = SanitizedUnicode(dump_only=True)

invenio_checks/services/services.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2025 CERN.
+#
+# Invenio-Checks is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Checks services."""
+
+from invenio_records_resources.services.records import RecordService
+
+
+class BaseClass(RecordService):
+    """Base service class for DB-backed services.
+
+    NOTE: See https://github.com/inveniosoftware/invenio-records-resources/issues/583
+    for future directions.
+    TODO: This has to be addressed now, since we're at 4+ cases that need a DB service.
+    """
+
+    def rebuild_index(self, identity, uow=None):
+        """Raise error since services are not backed by search indices."""
+        raise NotImplementedError()
+
+
+class CheckConfigService(RecordService):
+    """Service for managing and check configurations."""
+
+    def read(self, identity, id_, **kwargs):
+        """Read a check configuration."""
+        raise NotImplementedError()
+
+    def search(self, identity, params=None, **kwargs):
+        """Search for check configurations."""
+        raise NotImplementedError()
+
+    def create(self, identity, data, uow=None, **kwargs):
+        """Create a check configuration."""
+        raise NotImplementedError()
+
+    def update(self, identity, id_, data, revision_id=None, uow=None, **kwargs):
+        """Update a check configuration."""
+        raise NotImplementedError()
+
+    def delete(self, identity, id_, revision_id=None, uow=None, **kwargs):
+        """Delete a check configuration."""
+        raise NotImplementedError()