skit_pipelines.utils package

Submodules

skit_pipelines.utils.cookies module

construct_cookie_dict(response: List[Dict[str, Any]]) → Dict[str, str]

fetch_latest_cookies(cookie_save_path: str = '/tmp/kf_cookies.json') → Dict[str, str]

load_cookies(COOKIES_PATH: str = '/tmp/kf_cookies.json') → Dict[str, str]

save_cookies(cookies: Dict[str, str], COOKIES_PATH: str = '/tmp/kf_cookies.json') → None

simulate_selenium_connection(username, password) → List[Dict[str, Any]]

skit_pipelines.utils.helpers module

convert_audiourl_to_filename(audiourl: str) → str

converts s3 http turn audio url into the actual filename with .wav extension.

get_unix_epoch_timestamp_from_s3_presigned_url(s3_http_turn_audio_url: str) → int

returns unix epoch timestamp present in URL which is assumed to be in seconds

re_presign_audio_url_if_required(s3_http_turn_audio_url: str) → str

check if the s3 http url for turn audio is valid or not - for downloading purposes only if not re-presign URLs using s3 client, for the next 7 days.

this is highly specific to US, since the bucket is private there.

skit_pipelines.utils.k8s module

get_pipeline_config_kfp(pipeline_name)

skit_pipelines.utils.login module

kubeflow_login(force: bool = False) → kfp._client.Client

skit_pipelines.utils.normalize module

comma_sep_str(string: str, fn=None) → list

non_blank(c: str) → bool

strip(c: str) → str

to_camel_case(name: str) → str

to_snake_case(name: str) → str

skit_pipelines.utils.storage module

create_storage_path(storage_options: StorageOptions, path: str)

skit_pipelines.utils.webhook module

send_webhook_request(url: str, data: Dict[str, Any])

Send finished runs to the webhook url

validate_request_success(resp: Union[requests.models.Response, str])

Validate the response from the webhook request

Module contents

class SlackBlockFactory(content)

Bases: object

build() → Dict

code_block(content)

ping(cc)

text()

text_block()

create_dir_name(org_id: str, dir_type: str) → str

create_file_name(reference: str, file_type: str, ext='.csv') → str

filter_schema(schema: Dict[str, Any], filter_list: list) → Dict[str, Any]

generate_generic_dir_name(name: str) → str