Source code for ace.centralize.models.scheduler

"""
    Scheduler models for Centralize

    Models for scheduled jobs, business events, and job processing.

    Based on the Scheduler GraphQL schema (CreateScheduledJobParamsInput,
    CreateBusinessEventParamsInput, JobProcessedParamsInput, etc.)
"""
from datetime import date, datetime
from typing import List, Optional, Any
from uuid import UUID

from ace.centralize.models.base import ModelBase


# ================================================== KEY VALUE =========================================================



class KeyValueInput(ModelBase):
    """Maps to KeyValueInput in GraphQL schema."""


    def __init__(self, key: str, value: Any):
        self.key = key
        self.value = value




# Backward-compatible alias
SchedulerParameter = KeyValueInput


# ============================================== SCHEDULED JOB =========================================================



class CreateScheduledJobParamsInput(ModelBase):
    """Maps to CreateScheduledJobParamsInput in GraphQL schema."""


    def __init__(
            self,
            effective_date: date,
            entity_id: int,
            entity_type_id: str,
            business_event_id: Optional[int] = None,
            entity_type_name: Optional[str] = None,
            execution_datetime: Optional[datetime] = None,
            job_id: Optional[int] = None,
            job_name: Optional[str] = None,
            parameters: Optional[List[KeyValueInput]] = None,
            payload: Optional[str] = None,
            user_id: Optional[int] = None,
    ):
        self.effectiveDate = effective_date
        self.entityId = entity_id
        self.entityTypeId = entity_type_id
        if business_event_id is not None:
            self.businessEventId = business_event_id
        if entity_type_name is not None:
            self.entityTypeName = entity_type_name
        if execution_datetime is not None:
            self.executionDateTime = execution_datetime
        if job_id is not None:
            self.jobId = job_id
        if job_name is not None:
            self.jobName = job_name
        if parameters is not None:
            self.parameters = parameters
        if payload is not None:
            self.payload = payload
        if user_id is not None:
            self.userId = user_id




# Backward-compatible alias
CreateScheduledJobObject = CreateScheduledJobParamsInput




class CancelScheduledJobParamsInput(ModelBase):
    """Maps to CancelScheduledJobParamsInput in GraphQL schema."""


    def __init__(
            self,
            entity_id: Optional[int] = None,
            entity_type_id: Optional[str] = None,
            job_id: Optional[int] = None,
            next_execution_date_time_utc: Optional[datetime] = None,
            user_id: Optional[int] = None,
    ):
        if entity_id is not None:
            self.entityId = entity_id
        if entity_type_id is not None:
            self.entityTypeId = entity_type_id
        if job_id is not None:
            self.jobId = job_id
        if next_execution_date_time_utc is not None:
            self.nextExecutionDateTimeUTC = next_execution_date_time_utc
        if user_id is not None:
            self.userId = user_id






class RunScheduledJobParamsInput(ModelBase):
    """Maps to RunScheduledJobParamsInput in GraphQL schema."""


    def __init__(
            self,
            entity_id: Optional[int] = None,
            entity_type_id: Optional[str] = None,
            job_id: Optional[int] = None,
            job_name: Optional[str] = None,
            next_execution_date_time_utc: Optional[datetime] = None,
            user_id: Optional[int] = None,
    ):
        if entity_id is not None:
            self.entityId = entity_id
        if entity_type_id is not None:
            self.entityTypeId = entity_type_id
        if job_id is not None:
            self.jobId = job_id
        if job_name is not None:
            self.jobName = job_name
        if next_execution_date_time_utc is not None:
            self.nextExecutionDateTimeUTC = next_execution_date_time_utc
        if user_id is not None:
            self.userId = user_id




# ============================================= BUSINESS EVENT =========================================================



class CreateBusinessEventParamsInput(ModelBase):
    """Maps to CreateBusinessEventParamsInput in GraphQL schema."""


    def __init__(
            self,
            creation_user_id: int,
            effective_date: date,
            entity_id: int,
            entity_type_id: str,
            entity_type_name: Optional[str] = None,
            event_datetime: Optional[datetime] = None,
            event_type_id: Optional[int] = None,
            event_type_name: Optional[str] = None,
            parameters: Optional[List[KeyValueInput]] = None,
    ):
        self.creationUserId = creation_user_id
        self.effectiveDate = effective_date
        self.entityId = entity_id
        self.entityTypeId = entity_type_id
        if entity_type_name is not None:
            self.entityTypeName = entity_type_name
        if event_datetime is not None:
            self.eventDateTime = event_datetime
        if event_type_id is not None:
            self.eventTypeId = event_type_id
        if event_type_name is not None:
            self.eventTypeName = event_type_name
        if parameters is not None:
            self.parameters = parameters




# Backward-compatible alias
CreateBusinessEventObject = CreateBusinessEventParamsInput




class CancelBusinessEventParamsInput(ModelBase):
    """Maps to CancelBusinessEventParamsInput in GraphQL schema."""


    def __init__(
            self,
            id: Optional[int] = None,
            entity_id: Optional[int] = None,
            entity_type_id: Optional[str] = None,
            event_datetime: Optional[datetime] = None,
            event_type_id: Optional[int] = None,
    ):
        if id is not None:
            self.id = id
        if entity_id is not None:
            self.entityId = entity_id
        if entity_type_id is not None:
            self.entityTypeId = entity_type_id
        if event_datetime is not None:
            self.eventDateTime = event_datetime
        if event_type_id is not None:
            self.eventTypeId = event_type_id






class UpdateBusinessEventDateTimeInput(ModelBase):
    """Maps to UpdateBusinessEventDateTimeInput in GraphQL schema."""


    def __init__(
            self,
            id: int,
            event_datetime: Optional[datetime] = None,
    ):
        self.id = id
        if event_datetime is not None:
            self.eventDateTime = event_datetime




# ============================================= JOB PROCESSED =========================================================



class JobProcessedParamsInput(ModelBase):
    """Maps to JobProcessedParamsInput in GraphQL schema."""


    def __init__(
            self,
            scheduled_job_id: int,
            status_id: int,
            end_date_time_utc: Optional[datetime] = None,
            error_code: Optional[int] = None,
            error_description: Optional[str] = None,
            request_guid: Optional[UUID] = None,
            response_payload: Optional[str] = None,
            user_id: Optional[int] = None,
    ):
        self.scheduledJobId = scheduled_job_id
        self.statusId = status_id
        if end_date_time_utc is not None:
            self.endDateTimeUTC = end_date_time_utc
        if error_code is not None:
            self.errorCode = error_code
        if error_description is not None:
            self.errorDescription = error_description
        if request_guid is not None:
            self.requestGuid = request_guid
        if response_payload is not None:
            self.responsePayload = response_payload
        if user_id is not None:
            self.userId = user_id