Final commit of scheduler module into openvair

.github/workflows/tg-notify.yml

@@ -0,0 +1,27 @@
+name: Notifications in Telegram
+
+on:
+  pull_request:
+    types:
+      - opened
+      - review_requested
+      - closed
+
+
+jobs:
+  notify:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Github Telegram Notifier
+        # You may pin to the exact commit or the version.
+        # uses: EverythingSuckz/github-telegram-notify@11cbbc4c9284459d289584e9da5c24e0bdfc07b9
+        uses: EverythingSuckz/github-telegram-notify@v1.1.2
+        with:
+          # A bot token for sending the github commit updates to a chat. You may create one my sending `/newbot` to [@BotFather](https://telegram.dog/BotFather)
+          bot_token: ${{ secrets.TG_NOTIFY_BOT_TOKEN }}
+          # The ID of the chat where you want the bot to send the message
+          chat_id: ${{ secrets.TG_NOTIFY_CHAT_ID }}
+          # The ID of the topic where you want to receive the notifications.
+          topic_id: ${{ secrets.TG_NOTIFY_TOPIC_ID }}
+

openvair/alembic/versions/99d3c25dbacc_add_scheduler_jobs.py

@@ -0,0 +1,76 @@
+"""add scheduler_jobs
+
+Revision ID: 99d3c25dbacc
+Revises: 1
+Create Date: 2026-02-23 21:35:28.751098
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = '99d3c25dbacc'
+down_revision = '1'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('scheduler_jobs',
+    sa.Column('id', sa.UUID(), nullable=False),
+    sa.Column('name', sa.String(length=255), nullable=False),
+    sa.Column('description', sa.Text(), nullable=True),
+    sa.Column('cron_schedule', sa.String(length=255), nullable=False),
+    sa.Column('command', sa.Text(), nullable=False),
+    sa.Column('enabled', sa.Boolean(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), nullable=False),
+    sa.Column('last_run', sa.DateTime(), nullable=True),
+    sa.Column('next_run', sa.DateTime(), nullable=True),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.alter_column('events', 'module',
+               existing_type=sa.VARCHAR(length=40),
+               nullable=False)
+    op.alter_column('events', 'object_id',
+               existing_type=sa.UUID(),
+               nullable=False)
+    op.alter_column('events', 'user_id',
+               existing_type=sa.UUID(),
+               nullable=False)
+    op.alter_column('events', 'event',
+               existing_type=sa.VARCHAR(length=50),
+               nullable=False)
+    op.alter_column('events', 'timestamp',
+               existing_type=postgresql.TIMESTAMP(),
+               nullable=False)
+    op.alter_column('events', 'information',
+               existing_type=sa.TEXT(),
+               nullable=False)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column('events', 'information',
+               existing_type=sa.TEXT(),
+               nullable=True)
+    op.alter_column('events', 'timestamp',
+               existing_type=postgresql.TIMESTAMP(),
+               nullable=True)
+    op.alter_column('events', 'event',
+               existing_type=sa.VARCHAR(length=50),
+               nullable=True)
+    op.alter_column('events', 'user_id',
+               existing_type=sa.UUID(),
+               nullable=True)
+    op.alter_column('events', 'object_id',
+               existing_type=sa.UUID(),
+               nullable=True)
+    op.alter_column('events', 'module',
+               existing_type=sa.VARCHAR(length=40),
+               nullable=True)
+    op.drop_table('scheduler_jobs')
+    # ### end Alembic commands ###

openvair/main.py

@@ -80,6 +80,9 @@
 from openvair.modules.storage.entrypoints.api import router as storage
 from openvair.modules.template.entrypoints.api import router as tempalte_router
 from openvair.modules.dashboard.entrypoints.api import router as dashboard
+from openvair.modules.scheduler.entrypoints.api import (
+    router as scheduler_router,
+)
 from openvair.modules.event_store.entrypoints.api import router as event_store
 from openvair.modules.block_device.entrypoints.api import router as block_router
 from openvair.modules.notification.entrypoints.api import (
@@ -105,6 +108,7 @@
 app.include_router(vn_router)
 app.include_router(backup_router)
 app.include_router(tempalte_router)
+app.include_router(scheduler_router)
 
 project_dir = Path(__file__).parent
 templates = Jinja2Templates(directory=project_dir / 'dist')

openvair/modules/scheduler/adapters/dto/internal/commands.py

@@ -0,0 +1,68 @@
+"""DTOs for internal service-layer command operations.
+
+This module provides command objects used to describe service and domain-level
+operations on schedulers`s jobs (create, edit, delete).
+"""
+from uuid import UUID
+from typing import Optional
+
+from openvair.common.base_pydantic_models import BaseDTOModel
+
+
+class GetJobServiceCommandDTO(BaseDTOModel):
+    """DTO for retrieving a job by its ID.
+
+    Attributes:
+        id (UUID): Unique identifier of the job.
+    """
+
+    id: UUID
+
+
+class CreateJobServiceCommandDTO(BaseDTOModel):
+    """DTO for creating a job at the service layer.
+
+    Contains metadata required to register a new job in the system.
+
+    Attributes:
+        name (str): Unique name of the job.
+        description (Optional[str]): Description of the job.
+        cron_schedule (str): CRON expression defining job schedule.
+        command (str): Command to execute.
+        enabled (bool): Indicates whether the job is active.
+    """
+
+    name: str
+    description: Optional[str] = None
+    cron_schedule: str
+    command: str
+    enabled: bool
+
+
+class UpdateJobServiceCommandDTO(BaseDTOModel):
+    """DTO for updating job fields at the service layer.
+
+    Attributes:
+        id (UUID) : UUID of the job.
+        name (Optional[str]): Updated name for the job.
+        description (Optional[str]): Updated description.
+        cron_schedule (Optional[str]): Updated CRON schedule.
+        command (Optional[str]): Updated command.
+        enabled (Optional[bool]): Indicates if the job should be active.
+    """
+    id: UUID
+    name: Optional[str] = None
+    description: Optional[str] = None
+    cron_schedule: Optional[str] = None
+    command: Optional[str] = None
+    enabled: Optional[bool] = None
+
+
+class DeleteJobServiceCommandDTO(BaseDTOModel):
+    """DTO for deleting a job at the service layer.
+
+    Attributes:
+        id (UUID): ID of the job to delete.
+    """
+
+    id: UUID

openvair/modules/scheduler/adapters/exceptions.py

@@ -0,0 +1,15 @@
+"""Adapter-layer exceptions for the scheduler module.
+
+This module defines exceptions related to scheduler operations at the adapter
+level.
+
+Classes:
+    - SchedulerJobNotFoundException: Exception raised when a job is not found
+        in the database.
+"""
+
+from openvair.abstracts.base_exception import BaseCustomException
+
+
+class JobNotFoundInDBException(BaseCustomException):
+    """Exception raised when a job is not found in the database."""

openvair/modules/scheduler/adapters/orm.py

@@ -0,0 +1,52 @@
+"""ORM models for scheduler module."""
+
+import uuid
+from typing import Union
+from datetime import datetime
+
+from sqlalchemy import Text, String, Boolean, DateTime
+from sqlalchemy.orm import Mapped, DeclarativeBase, mapped_column
+from sqlalchemy.dialects.postgresql import UUID
+
+
+class Base(DeclarativeBase):
+    """Base class for SQLAlchemy declarative models."""
+
+
+class SchedulerJob(Base):
+    """ORM class representing a job.
+
+    Attributes:
+        id: Unique identifier of the job.
+        name: Unique name of the job.
+        description: Optional description.
+        cron_schedule: Cron schedule expression.
+        command: Cron job command.
+        enabled: State of the job (enabled / disabled).
+        created_at: Timestamp when the job was created.
+        updated_at: Timestamp when the job was lastly updated.
+        last_run: Timestamp when the job ran last time.
+        next_run: Timestamp when the job is going to run next time.
+
+    """
+
+    __tablename__ = 'scheduler_jobs'
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    description: Mapped[Union[str, None]] = mapped_column(Text, nullable=True)
+    cron_schedule: Mapped[str] = mapped_column(String(255), nullable=False)
+    command: Mapped[str] = mapped_column(Text, nullable=False)
+    enabled: Mapped[bool] = mapped_column(Boolean, default=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, default=datetime.now, onupdate=datetime.now
+    )
+    last_run: Mapped[Union[datetime, None]] = mapped_column(
+        DateTime, nullable=True
+    )
+    next_run: Mapped[Union[datetime, None]] = mapped_column(
+        DateTime, nullable=True
+    )

openvair/modules/scheduler/adapters/repository.py

@@ -0,0 +1,58 @@
+"""SQLAlchemy repository for the scheduler module.
+
+This module implements the repository pattern to manage job entities
+in the database using SQLAlchemy.
+"""
+
+from uuid import UUID
+from typing import List, Optional
+
+from sqlalchemy.orm import Session
+
+from openvair.modules.scheduler.adapters.orm import SchedulerJob
+from openvair.common.repositories.base_sqlalchemy import (
+    BaseSqlAlchemyRepository,
+)
+
+# from openvair.modules.scheduler.adapters.exceptions import (
+#     JobNotFoundInDBException, # NEEDS_TO_BE_IMPLEMENTED!!!
+# )
+
+
+class SchedulerSqlAlchemyRepository(BaseSqlAlchemyRepository[SchedulerJob]):
+    """Repository for managing scheduler jobs entities.
+
+    This class provides CRUD operations for the jobs using SQLAlchemy.
+    """
+
+    def __init__(self, session: Session) -> None:
+        """Initialize repository with SQLAlchemy session."""
+        super().__init__(session, SchedulerJob)
+
+    def get_all(self) -> List[SchedulerJob]:
+        """Retrieve all scheduler jobs."""
+        return self.session.query(SchedulerJob).all()
+
+    def get_by_id(self, job_id: UUID) -> Optional[SchedulerJob]:
+        """Retrieve scheduler job by its ID."""
+        return (
+            self.session.query(SchedulerJob)
+            .filter(SchedulerJob.id == job_id)
+            .first()
+        )
+
+    def get_by_name(self, job_name: str) -> Optional[SchedulerJob]:
+        """Retrieve scheduler job by its name."""
+        return (
+            self.session.query(SchedulerJob)
+            .filter(SchedulerJob.name == job_name)
+            .first()
+        )
+
+    def add(self, job: SchedulerJob) -> None:
+        """Add a new job to session."""
+        self.session.add(job)
+
+    def delete(self, job: SchedulerJob) -> None:
+        """Delete job from session."""
+        self.session.delete(job)

openvair/modules/scheduler/adapters/serializer.py

@@ -0,0 +1,56 @@
+"""Serializers for the Scheduler module.
+
+Provides conversion logic between ORM models and DTOs used at API and domain
+layers.
+
+Classes: TODO: NEED TO CONVERT TO DTO!!!
+    - ApiSerializer: ORM <-> API DTO
+    - DomainSerializer: ORM <-> Domain DTO
+    - CreateSerializer: Create DTO -> ORM
+"""
+
+from typing import Any, Dict
+
+from openvair.modules.scheduler.adapters.orm import SchedulerJob
+
+
+class SchedulerJobSerializer:
+    """Base class for serializing methods."""
+
+    @staticmethod
+    def to_web(db_model: SchedulerJob) -> Dict[str, Any]:
+        """Gathering db attributes and converting them to a json format."""
+        return {
+            'id': str(db_model.id),
+            'name': db_model.name,
+            'description': db_model.description,
+            'cron_schedule': db_model.cron_schedule,
+            'command': db_model.command,
+            'enabled': db_model.enabled,
+            'created_at': db_model.created_at.isoformat() if db_model.created_at
+                                                            else None,
+            'updated_at': db_model.updated_at.isoformat() if db_model.updated_at
+                                                            else None,
+            'last_run': db_model.last_run.isoformat() if db_model.last_run
+                                                            else None,
+            'next_run': db_model.next_run.isoformat() if db_model.next_run
+                                                            else None,
+        }
+
+
+    @staticmethod
+    def to_db(data: Dict[str, Any]) -> SchedulerJob:
+        """Parsing json into values suitable for database attributes."""
+        return SchedulerJob(**data)
+
+    @staticmethod
+    def to_domain(db_model: SchedulerJob) -> Dict[str, Any]:
+        """Gathering attributes to send to domain layer for job creation."""
+        return {
+            'id': str(db_model.id),
+            'name': db_model.name,
+            'description': db_model.description,
+            'cron_schedule': db_model.cron_schedule,
+            'command': db_model.command,
+            'enabled': db_model.enabled,
+        }