Job objects

Wordcab API Job object.

BaseJob dataclass

Wordcab API BaseJob object.

Source code in src/wordcab/core_objects/job.py

@dataclass
class BaseJob:
    """Wordcab API BaseJob object."""

    display_name: str
    job_name: str
    source: str
    audio_duration: int = field(default=0)
    job_status: Optional[str] = field(default="Pending")
    metadata: Optional[Dict[str, str]] = field(default=None)
    settings: Optional[JobSettings] = field(default=None)
    source_lang: Optional[str] = field(default=None)
    target_lang: Optional[str] = field(default=None)
    tags: Optional[List[str]] = field(default=None)
    time_started: Optional[str] = field(default=None)
    time_completed: Optional[str] = field(default=None)
    transcript_details: Optional[Dict[str, str]] = field(default=None)
    transcript_id: Optional[str] = field(default=None)

    def __post_init__(self) -> None:
        """Post-init method."""
        logger.info(f"Job {self.job_name} created.")
        if self.source not in SOURCE_OBJECT_MAPPING.keys():
            raise ValueError(f"""
                Source {self.source} is not a valid source. Valid sources are {SOURCE_OBJECT_MAPPING.keys()}.
            """)

    def job_update(self, parameters: Dict[str, str]) -> None:
        """Update the job attributes."""
        for key, value in parameters.items():
            if key in self.__dict__:
                if getattr(self, key) != value:
                    setattr(self, key, value)
                    logger.info(f"Job {self.job_name} updated: {key} = {value}")
                else:
                    logger.info(f"Job {self.job_name} not updated: {key} = {value}")
            else:
                logger.warning(
                    f"Cannot update {key} in {self.job_name}, not a valid attribute."
                )

__post_init__()

Post-init method.

Source code in src/wordcab/core_objects/job.py

def __post_init__(self) -> None:
    """Post-init method."""
    logger.info(f"Job {self.job_name} created.")
    if self.source not in SOURCE_OBJECT_MAPPING.keys():
        raise ValueError(f"""
            Source {self.source} is not a valid source. Valid sources are {SOURCE_OBJECT_MAPPING.keys()}.
        """)

job_update(parameters)

Update the job attributes.

Source code in src/wordcab/core_objects/job.py

def job_update(self, parameters: Dict[str, str]) -> None:
    """Update the job attributes."""
    for key, value in parameters.items():
        if key in self.__dict__:
            if getattr(self, key) != value:
                setattr(self, key, value)
                logger.info(f"Job {self.job_name} updated: {key} = {value}")
            else:
                logger.info(f"Job {self.job_name} not updated: {key} = {value}")
        else:
            logger.warning(
                f"Cannot update {key} in {self.job_name}, not a valid attribute."
            )

ExtractJob dataclass

Bases: BaseJob

Wordcab API ExtractJob object.

Source code in src/wordcab/core_objects/job.py

@dataclass
class ExtractJob(BaseJob):
    """Wordcab API ExtractJob object."""

    def __post_init__(self) -> None:
        """Post-init."""
        super().__post_init__()
        self._job_type = "ExtractJob"
        self.available_status = EXTRACT_AVAILABLE_STATUS

__post_init__()

Post-init.

Source code in src/wordcab/core_objects/job.py

def __post_init__(self) -> None:
    """Post-init."""
    super().__post_init__()
    self._job_type = "ExtractJob"
    self.available_status = EXTRACT_AVAILABLE_STATUS

JobSettings dataclass

Wordcab API Job Settings object.

Source code in src/wordcab/core_objects/job.py

@dataclass
class JobSettings:
    """Wordcab API Job Settings object."""

    ephemeral_data: Optional[bool] = field(default=False)
    pipeline: str = field(default="default")
    only_api: Optional[bool] = field(default=True)
    split_long_utterances: Optional[bool] = field(default=False)

    def __post_init__(self) -> None:
        """Post init."""
        if self.pipeline == "default":
            raise ValueError("Pipeline must be set to a valid pipeline name")

__post_init__()

Post init.

Source code in src/wordcab/core_objects/job.py

def __post_init__(self) -> None:
    """Post init."""
    if self.pipeline == "default":
        raise ValueError("Pipeline must be set to a valid pipeline name")

ListJobs dataclass

Wordcab API ListJobs object.

Source code in src/wordcab/core_objects/job.py

@dataclass
class ListJobs:
    """Wordcab API ListJobs object."""

    page_count: int
    next_page: str
    results: List[Union[ExtractJob, SummarizeJob]]

SummarizeJob dataclass

Bases: BaseJob

Wordcab API SummarizeJob object.

Source code in src/wordcab/core_objects/job.py

@dataclass
class SummarizeJob(BaseJob):
    """Wordcab API SummarizeJob object."""

    summary_details: Optional[Dict[str, str]] = field(default=None)

    def __post_init__(self) -> None:
        """Post-init."""
        super().__post_init__()
        self._job_type = "SummarizeJob"
        self.available_status = SUMMARIZE_AVAILABLE_STATUS

__post_init__()

Post-init.

Source code in src/wordcab/core_objects/job.py

def __post_init__(self) -> None:
    """Post-init."""
    super().__post_init__()
    self._job_type = "SummarizeJob"
    self.available_status = SUMMARIZE_AVAILABLE_STATUS

TranscribeJob dataclass

Bases: BaseJob

Wordcab API TranscribeJob object.

Source code in src/wordcab/core_objects/job.py

@dataclass
class TranscribeJob(BaseJob):
    """Wordcab API TranscribeJob object."""

    def __post_init__(self) -> None:
        """Post-init."""
        super().__post_init__()
        self._job_type = "TranscribeJob"
        self.available_status = SUMMARIZE_AVAILABLE_STATUS

__post_init__()

Post-init.

Source code in src/wordcab/core_objects/job.py

def __post_init__(self) -> None:
    """Post-init."""
    super().__post_init__()
    self._job_type = "TranscribeJob"
    self.available_status = SUMMARIZE_AVAILABLE_STATUS

---

Last update: 2023-09-25
Created: 2023-09-25