Module omnipy.config.job

Overview

View Source

from dataclasses import dataclass, field

import os

from pathlib import Path

from omnipy.api.enums import (ConfigOutputStorageProtocolOptions,

                              ConfigPersistOutputsOptions,

                              ConfigRestoreOutputsOptions)

from omnipy.api.protocols.public.config import (IsLocalOutputStorage,

                                                IsOutputStorage,

                                                IsS3OutputStorage)

def _get_persist_data_dir_path() -> str:

    return str(Path.cwd().joinpath(Path('outputs')))

@dataclass

class LocalOutputStorage:

    persist_data_dir_path: str = field(default_factory=_get_persist_data_dir_path)

@dataclass

class S3OutputStorage:

    persist_data_dir_path: str = os.path.join('omnipy', 'outputs')

    endpoint_url: str = ''

    bucket_name: str = ''

    access_key: str = ''

    secret_key: str = ''

@dataclass

class OutputStorage:

    persist_outputs: ConfigPersistOutputsOptions = \

        ConfigPersistOutputsOptions.ENABLE_FLOW_AND_TASK_OUTPUTS

    restore_outputs: ConfigRestoreOutputsOptions = \

        ConfigRestoreOutputsOptions.DISABLED

    protocol: ConfigOutputStorageProtocolOptions = ConfigOutputStorageProtocolOptions.LOCAL

    local: IsLocalOutputStorage = field(default_factory=LocalOutputStorage)

    s3: IsS3OutputStorage = field(default_factory=S3OutputStorage)

@dataclass

class JobConfig:

    output_storage: IsOutputStorage = field(default_factory=OutputStorage)

Classes

JobConfig

class JobConfig(
    output_storage: omnipy.api.protocols.public.config.IsOutputStorage = <factory>
)

JobConfig(output_storage: omnipy.api.protocols.public.config.IsOutputStorage = )

View Source

@dataclass

class JobConfig:

    output_storage: IsOutputStorage = field(default_factory=OutputStorage)

---

Methods

eq

def __eq__(
    self,
    other
)

Return self==value.

Parameters:

Name	Type	Description	Default
other

LocalOutputStorage

class LocalOutputStorage(
    persist_data_dir_path: str = <factory>
)

LocalOutputStorage(persist_data_dir_path: str = )

View Source

@dataclass

class LocalOutputStorage:

    persist_data_dir_path: str = field(default_factory=_get_persist_data_dir_path)

---

Methods

eq

def __eq__(
    self,
    other
)

Return self==value.

Parameters:

Name	Type	Description	Default
other

OutputStorage

class OutputStorage(
    persist_outputs: omnipy.api.enums.ConfigPersistOutputsOptions = <ConfigPersistOutputsOptions.ENABLE_FLOW_AND_TASK_OUTPUTS: 'all'>,
    restore_outputs: omnipy.api.enums.ConfigRestoreOutputsOptions = <ConfigRestoreOutputsOptions.DISABLED: 'disabled'>,
    protocol: omnipy.api.enums.ConfigOutputStorageProtocolOptions = <ConfigOutputStorageProtocolOptions.LOCAL: 'local'>,
    local: omnipy.api.protocols.public.config.IsLocalOutputStorage = <factory>,
    s3: omnipy.api.protocols.public.config.IsS3OutputStorage = <factory>
)

OutputStorage(persist_outputs: omnipy.api.enums.ConfigPersistOutputsOptions = , restore_outputs: omnipy.api.enums.ConfigRestoreOutputsOptions = , protocol: omnipy.api.enums.ConfigOutputStorageProtocolOptions = , local: omnipy.api.protocols.public.config.IsLocalOutputStorage = , s3: omnipy.api.protocols.public.config.IsS3OutputStorage = )

View Source

@dataclass

class OutputStorage:

    persist_outputs: ConfigPersistOutputsOptions = \

        ConfigPersistOutputsOptions.ENABLE_FLOW_AND_TASK_OUTPUTS

    restore_outputs: ConfigRestoreOutputsOptions = \

        ConfigRestoreOutputsOptions.DISABLED

    protocol: ConfigOutputStorageProtocolOptions = ConfigOutputStorageProtocolOptions.LOCAL

    local: IsLocalOutputStorage = field(default_factory=LocalOutputStorage)

    s3: IsS3OutputStorage = field(default_factory=S3OutputStorage)

---

Class variables

persist_outputs

protocol

restore_outputs

Methods

eq

def __eq__(
    self,
    other
)

Return self==value.

Parameters:

Name	Type	Description	Default
other

S3OutputStorage

class S3OutputStorage(
    persist_data_dir_path: str = 'omnipy/outputs',
    endpoint_url: str = '',
    bucket_name: str = '',
    access_key: str = '',
    secret_key: str = ''
)

S3OutputStorage(persist_data_dir_path: str = 'omnipy/outputs', endpoint_url: str = '', bucket_name: str = '', access_key: str = '', secret_key: str = '')

View Source

@dataclass

class S3OutputStorage:

    persist_data_dir_path: str = os.path.join('omnipy', 'outputs')

    endpoint_url: str = ''

    bucket_name: str = ''

    access_key: str = ''

    secret_key: str = ''

---

Class variables

access_key

bucket_name

endpoint_url

persist_data_dir_path

secret_key

Methods

eq

def __eq__(
    self,
    other
)

Return self==value.

Parameters:

Name	Type	Description	Default
other