ts-task-script-utils 1.7.0

Python utility for Tetra Task Scripts

ts-task-script-utils

Version

v1.7.0

Table of Contents

• Summary
• Installation
• Usage
  • Parsing Numbers
  • Parsing Datetimes
    • parse Usage
  • Generating Random UUIDs for Task Scripts
  • Creating datacube Parquet files
  • Logging warning messages in Task Scripts
• Changelog
  • v1.7.0
  • v1.6.0
  • v1.5.0
  • v1.4.0
  • v1.3.1
  • v1.3.0
  • v1.2.0
  • v1.1.1
  • v1.1.0

Summary

Utility functions for Tetra Task Scripts

Installation

pip install ts-task-script-utils

Usage

Parsing Numbers

from task_script_utils.parse import to_int

string_value = '1.0'
int_value = to_int(string_value)

# `int_value` now has the parsed value of the string
assert isinstance(int_value, int)
assert int_value == 1

# it returns `None` if the value is unparseable
string_value = 'not an int'
int_value = to_int(string_value)

assert int_value is None

Parsing Datetimes

DEPRECATION WARNING!

• Do not use the old datetime parser: convert_datetime_to_ts_format (from task_script_utils.convert_datetime_to_ts_format)
• Instead, use the newer parse from task_script_utils.datetime_parser

parse Usage

from task_script_utils.datetime_parser import parse

parse("2004-12-23T12:30 AM +05:30")
parse("2004-12-23T12:30 AM +05:30", <format_list>)
parse("2004-12-23T12:30 AM +05:30", <format_list>, <datetime_config>)

parse() returns a TSDatetime Object. You can use TSDatetime.tsformat() and TSDatetime.isoformat() to get datetime string. You can also use TSDatetime.datetime() to access python datetime object.

You can read more about the datetime parser here.

Generating Random UUIDs for Task Scripts

To generate standard and random UUIDs, Python's uuid module can be used (uuid1 for standard and uuid4 for random). However, to get UUIDs that are reproducible for a given task script and input file, a custom UUID generator is provided: task_script_utils.random.TaskScriptUUIDGenerator.

from pathlib import Path
from task_script_utils.random import TaskScriptUUIDGenerator

input_file = Path(...)
file_content = input_file.read_bytes()
rand = TaskScriptUUIDGenerator("common/my-task-script:v1.0.0", file_content)

# Get 3 random bytes:
random_bytes = rand.randbytes(3)

# Get a random UUID:
uuid = rand.uuid()

It's also possible to use a class method and provide the task script identifiers separately:

from pathlib import Path
from task_script_utils.random import TaskScriptUUIDGenerator

input_file = Path(...)
file_content = input_file.read_bytes()
rand = TaskScriptUUIDGenerator.from_task_script_identifier_parts("common", "my-task-script", "v1.0.0", file_content)

This is a singleton class, meaning creating multiple instances of the class with the same arguments results in getting the identical object back, e.g.:

from pathlib import Path
from task_script_utils.random import TaskScriptUUIDGenerator

input_file = Path(...)
file_content = input_file.read_bytes()
rand1 = TaskScriptUUIDGenerator("common/my-task-script:v1.0.0", file_content)
rand2 = TaskScriptUUIDGenerator("common/my-task-script:v1.0.0", file_content)

assert rand1 is rand2

It's also possible to get the most-recently-created instance through the get_last_created method:

from pathlib import Path
from task_script_utils.random import TaskScriptUUIDGenerator

input_file = Path(...)
file_content = input_file.read_bytes()
rand1 = TaskScriptUUIDGenerator("common/my-task-script:v1.0.0", file_content)

rand2 = TaskScriptUUIDGenerator.get_last_created()

assert rand1 is rand2

Creating datacube Parquet files

See the docs.

Using Python's logging module in Task Scripts

Task Scripts can write workflow logs which are visible to users on TDP, but only if the logs are written via the logger provided by the context object. The context logger is documented here: context.get_logger.

This utility is a wrapper for the context logger which allows Task Scripts to use Python's logging module for creating TDP workflow logs, instead of directly using the context logger object. This means the context logger object doesn't need to be passed around to each function which needs to do logging, and Task Script code can benefit from other features of the Python logging module such as integration with pytest.

To log warning messages on the platform from a task script do the following:

• Setup the log handler in main.py:

from task_script_utils.logging import (
    setup_ts_log_handler,
)

• Then within the function called by the protocol:

setup_ts_log_handler(context.get_logger(), "main")

• In a module where you wish to log a warning:

import logging
logger = logging.getLogger("main." + __name__)

• Log a warning message with:

logger.warning("This is a warning message")

Changelog

v1.7.0

• Add task_script_utils.logging for logging warning messages in task scripts

v1.6.0

• Add task_script_utils.datacubes.parquet for creating Parquet file representations of datacubes

v1.5.0

• Add TaskScriptUUIDGenerator class for generating random UUIDs and random bytes.

v1.4.0

• Add extract-to-decorate functions

v1.3.1

• Update datetime parser usage in README.md

v1.3.0

• Added string parsing functions

v1.2.0

• Add boolean config parameter require_unambiguous_formats to DatetimeConfig
• Add logic to parser._parse_with_formats to be used when DatetimeConfig.require_unambiguous_formats is set to True
  • AmbiguousDatetimeFormatsError is raised if mutually ambiguous formats are detected and differing datetimes are parsed
• Add parameter typing throughout repository
• Refactor datetime_parser package
• Add base class DateTimeInfo
• Segregate parsing logic into ShortDateTimeInfo and LongDateTimeInfo

v1.1.1

• Remove convert_to_ts_iso8601() method

v1.1.0

• Add datetime_parser package