New Blue Prism Queue Item

blueprism_queue_item.py

from __future__ import annotations
from typing import TYPE_CHECKING

from dataclasses import dataclass, fields
from datetime import datetime, timezone
from enum import Enum
import uuid

import pandas as pd  # External package, not in standard library.
from sqlengine import build_engine  # External package, not in standard library.

if TYPE_CHECKING:
    from uuid import UUID
    from typing import Optional


class BluePrismDatabase(Enum):
    SCHEMA: str = "dbo"
    DEV_DB: str = "BP_DEV"
    UAT_DB: str = "BP_UAT"
    PRD_DB: str = "BP_PRD"
    DEV_HOST: str = "LGCDVDBZ500"
    UAT_HOST: str = "UATLGDB1"
    PROD_HOST: str = "PRDLGDB2"
    QUEUE_ITEMS_TABLE: str = "BPAWorkQueueItem"


class Status(Enum):
    PENDING = None
    BUSINESS_EXCEPTION = "Business Exception"
    SYSTEM_EXCEPTION = "System Exception"
    COMPLETED = "Completed"


@dataclass
class NewQueueItem:
    queueid: UUID
    keyvalue: str
    attempt: int
    queueident: int
    lockid: Optional[UUID] = None
    locktime: Optional[datetime] = None
    encryptid: Optional[str] = None
    prevworktime: Optional[int] = 0
    priority: Optional[int] = 0
    sessionid: Optional[UUID] = None
    data: Optional[str] = None
    worktime: Optional[int] = 0
    deferred: Optional[datetime] = None
    exceptionreason: Optional[str] = None
    exception: Optional[datetime] = None
    completed: Optional[datetime] = None
    loaded: Optional[datetime] = datetime.now(timezone.utc).replace(tzinfo=None)
    status: Optional[str] = Status.PENDING.value
    id: Optional[UUID] = uuid.uuid4()  # If the DB is hosted on MS SQL Server, this may not adhere to the 'uniqueidentifier' type.

    def __post_init__(self):
        if isinstance(self.queueid, str):
            self.queueid = uuid.UUID(self.queueid)
        if isinstance(self.lockid, str):
            self.lockid = uuid.UUID(self.lockid)
        if isinstance(self.sessionid, str):
            self.sessionid = uuid.UUID(self.sessionid)
        if isinstance(self.id, str):
            self.id = uuid.UUID(self.id)
        for f in fields(self):
            value = getattr(self, f.name)
            if isinstance(value, datetime):
                setattr(self, f.name, value.replace(tzinfo=None))


engine = build_engine(
    driver="mssql",
    host=BluePrismDatabase.UAT_HOST.value,
    database=BluePrismDatabase.UAT_DB.value
)
# Create NewQueueItem objects en masse, from a dict or something... write it yourself...

queue_items: list[NewQueueItem] = []  # Put the NewQueueItem objects here.
queue_items_df: pd.DataFrame = pd.DataFrame(queue_items)  # Convert the list of NewQueueItem objects to a DataFrame.
queue_items_df.to_sql(  # Insert DataFrame into SQL Server database.
    name=BluePrismDatabase.QUEUE_ITEMS_TABLE.value,
    con=engine,
    schema=BluePrismDatabase.SCHEMA.value,
    if_exists="append",
    index=False
)