Description #

TOS pre-signed URL generation processor

Key features #

• Signature generation mechanism: Generates time-limited pre-signed URLs based on the Volcano Engine TOS SDK
• URL schema handling:
  • Native support for signature conversion of tos/s3 protocol paths
  • Automatically skips paths containing the http/https protocol
• Security control:
  • Configurable signature validity period (default: 3600 seconds)

Input and output specifications #

• Input format:
  • TOS path: string type, supports the following formats:
    • tos://{bucket}/{object}
    • s3://{bucket}/{object}
• Output format:
  • Valid signed URL: https protocol string
  • Exception: Returns None

Configuration parameter description #

• expires: Controls the validity period of the signature (unit: seconds). If too short, it may cause business interruption; if too long, it poses security risks.

Operator parameters #

Input #

Output #

Generated signed URL

Parameters #

If a parameter does not have a default value, it is required.

Examples #

The following code demonstrates how to use daft to run the operator and generate ordinary pre-signed URLs for TOS.

from __future__ import annotations

import os

import daft
from daft import col
from daft.las.functions.text.pre_sign_url_for_tos import PreSignUrlForTos
from daft.las.functions.udf import las_udf

if os.getenv("DAFT_RUNNER", "native") == "ray":
    import logging

    import ray

    def configure_logging():
        logging.basicConfig(
            level=logging.INFO,
            format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S.%s".format(),
        )
        logging.getLogger("tracing.span").setLevel(logging.WARNING)
        logging.getLogger("daft_io.stats").setLevel(logging.WARNING)
        logging.getLogger("DaftStatisticsManager").setLevel(logging.WARNING)
        logging.getLogger("DaftFlotillaScheduler").setLevel(logging.WARNING)
        logging.getLogger("DaftFlotillaDispatcher").setLevel(logging.WARNING)

    ray.init(dashboard_host="0.0.0.0", runtime_env={"worker_process_setup_hook": configure_logging})
    daft.set_runner_ray()

daft.set_execution_config(actor_udf_ready_timeout=600)
daft.set_execution_config(min_cpu_per_task=0)

if __name__ == "__main__":
    # Sample data
    TOS_TEST_DIR = os.getenv("TOS_TEST_DIR", "tos_bucket")
    samples = {"url": [f"tos://{TOS_TEST_DIR}/sample.mp4"]}

    # Use the provided operator to generate signed URLs
    ds = daft.from_pydict(samples)
    ds = ds.with_column(
        "signed_url",
        las_udf(
            PreSignUrlForTos,
            construct_args={
                "expires": 3600,
            },
        )(col("url")),
    )

    ds.show()
    # Output content may vary depending on actual conditions
    # ╭─────────────────────────────┬────────────────────────────────╮
    # │ url                         ┆ signed_url                     │
    # │ ---                         ┆ ---                            │
    # │ Utf8                        ┆ Utf8                           │
    # ╞═════════════════════════════╪════════════════════════════════╡
    # │ tos://tos_bucket/sample.mp4 ┆ https://tos_bucket.tos-cn-bei… │
    # ╰─────────────────────────────┴────────────────────────────────╯