59 lines
1.6 KiB
Python
59 lines
1.6 KiB
Python
import enum
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from app.brm.brn import BRN
|
|
from app.extensions import db
|
|
from app.models import AbstractConfiguration, AbstractResource
|
|
from app.models.types import AwareDateTime
|
|
|
|
|
|
class AutomationState(enum.Enum):
|
|
IDLE = 0
|
|
RUNNING = 1
|
|
ERROR = 3
|
|
|
|
|
|
class Automation(AbstractConfiguration):
|
|
short_name: Mapped[str]
|
|
state: Mapped[AutomationState] = mapped_column(default=AutomationState.IDLE)
|
|
enabled: Mapped[bool]
|
|
last_run: Mapped[Optional[datetime]] = mapped_column(AwareDateTime())
|
|
next_run: Mapped[Optional[datetime]] = mapped_column(AwareDateTime())
|
|
next_is_full: Mapped[bool]
|
|
|
|
logs = db.relationship("AutomationLogs", back_populates="automation")
|
|
|
|
@property
|
|
def brn(self) -> BRN:
|
|
return BRN(
|
|
group_id=0,
|
|
product="core",
|
|
provider="",
|
|
resource_type="automation",
|
|
resource_id=self.short_name,
|
|
)
|
|
|
|
def kick(self) -> None:
|
|
self.enabled = True
|
|
self.next_run = datetime.now(tz=timezone.utc)
|
|
self.updated = datetime.now(tz=timezone.utc)
|
|
|
|
|
|
class AutomationLogs(AbstractResource):
|
|
automation_id: Mapped[int] = mapped_column(db.ForeignKey("automation.id"))
|
|
logs: Mapped[str]
|
|
|
|
automation = db.relationship("Automation", back_populates="logs")
|
|
|
|
@property
|
|
def brn(self) -> BRN:
|
|
return BRN(
|
|
group_id=0,
|
|
product="core",
|
|
provider="",
|
|
resource_type="automationlog",
|
|
resource_id=str(self.id),
|
|
)
|