diff --git a/invenio_checks/services/__init__.py b/invenio_checks/services/__init__.py new file mode 100644 index 0000000..de7a9de --- /dev/null +++ b/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", +) diff --git a/invenio_checks/services/config.py b/invenio_checks/services/config.py new file mode 100644 index 0000000..9f86b60 --- /dev/null +++ b/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 = {} diff --git a/invenio_checks/services/permissions.py b/invenio_checks/services/permissions.py new file mode 100644 index 0000000..7dce926 --- /dev/null +++ b/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()] diff --git a/invenio_checks/services/results.py b/invenio_checks/services/results.py new file mode 100644 index 0000000..46e3337 --- /dev/null +++ b/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 diff --git a/invenio_checks/services/schema.py b/invenio_checks/services/schema.py new file mode 100644 index 0000000..ac8707b --- /dev/null +++ b/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) diff --git a/invenio_checks/services/services.py b/invenio_checks/services/services.py new file mode 100644 index 0000000..2ee0006 --- /dev/null +++ b/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()