libera_utils.aws.manual_processing#

Module for manual processing in the SDC

This module provides functionality to manually run processing jobs in the SDC by submitting custom processing graph configurations. This supports testing of individual processing steps via single-node DAGs (the step-function-trigger CLI), as well as fully custom DAGs defined by the user (the manual-processing CLI). The step-function-trigger CLI is intended to be used for quick testing of individual processing steps, while the manual-processing CLI is intended to be used for more complex processing runs that may involve multiple steps and custom input parameters. Manual processing runs without explicit custom DAGs are also supported via the manual-processing CLI for purposes such as reprocessing.

Both CLIs work by emitting a single ManualProcessing event to the SDC central EventBridge bus (the LiberaSDCEventBus). The SDC Job Creator picks up the event and builds the processing job(s) from it, then the rest of the SDC’s event-driven orchestration runs the node(s) end-to-end. Input products are assumed to already be ingested into the SDC (File Metadata + Data Availability rows exist) before the event is sent.

Functions

manual_processing_cli_handler(parsed_args)

CLI handler function for the manual-processing subcommand.

start_manual_processing(applicable_dates, *, ...)

Start a manual processing job by emitting a single ManualProcessing event to the SDC event bus.

step_function_trigger(algorithm_name, ...[, ...])

Manually trigger a single processing step for a single applicable date.

step_function_trigger_cli_handler(parsed_args)

CLI handler function for the step-function-trigger subcommand.
libera_utils.aws.manual_processing._to_date(value: str | date | datetime) → date#

Normalize a date-like value (ISO string, datetime, or date) to a datetime.date.

libera_utils.aws.manual_processing._validate_dag_config(dag: dict) → None#

Do basic validation of a custom DAG configuration before sending it to the SDC.

This is a best-effort early check to catch obvious errors (invalid identifiers, snake_case keys, dangling upstream references) before the event is emitted. The authoritative validation (including acyclicity) happens in the SDC Job Creator via the ProcessingSystemDagConfig Pydantic model, which lives in the libera_cdk repo.

Parameters:

dag (dict) – The custom DAG configuration, shaped like {"nodes": {<node-id>: <node>}} with kebab-case node keys.

Raises:

ValueError – If the structure, node ids, product ids, node keys, or upstream references are invalid.

libera_utils.aws.manual_processing._verify_jobs_created(job_ids: list[ULID], *, boto_session: Session, wait_time: float) → None#

Poll the SDC Coordination Table to confirm a job metadata record exists for each job id.

This is a basic check that the event was well-formed enough for the Job Creator to create the job(s). It does not guarantee the processing will succeed, since the job can still fail after creation. Jobs that do not appear within wait_time are logged as warnings rather than raising, since job creation is asynchronous. Used by both the manual-processing and step-function-trigger CLIs.

Parameters:

  • job_ids (list[ULID]) – The job ids to look for (PK of the metadata record), one per applicable date.

  • boto_session (boto3.Session) – Boto3 session used to discover and read the Coordination Table.

  • wait_time (float) – Maximum number of seconds to wait for all jobs to appear.

libera_utils.aws.manual_processing._verify_start_nodes_running(job_ids: list[ULID], start_processing_step_ids: list[ProcessingStepIdentifier], *, boto_session: Session, wait_time: float) → None#

Poll the SDC Coordination Table to confirm each job’s start node(s) move from PENDING to RUNNING.

For each job, this waits for every start node record (PK=job_id, SK=<processing-step-id>) to move from PENDING to RUNNING; SUCCEEDED is also accepted in case a node finishes very quickly. A node that reaches a final non-running state (NOTRUN or FAILED) or never leaves PENDING within wait_time is logged as a warning rather than raising, since node scheduling is asynchronous. A common cause of NOTRUN is that the node’s input data products are not yet available in the SDC.

This is used only by the step-function-trigger CLI, which runs a single node for a single applicable date. General manual-processing runs only confirm job creation (see _verify_jobs_created).

Parameters:

  • job_ids (list[ULID]) – The job ids to check, one per applicable date.

  • start_processing_step_ids (list[ProcessingStepIdentifier]) – The start nodes to check the status of within each job graph.

  • boto_session (boto3.Session) – Boto3 session used to discover and read the Coordination Table.

  • wait_time (float) – Maximum number of seconds to wait.

libera_utils.aws.manual_processing.manual_processing_cli_handler(parsed_args: Namespace) → None#

CLI handler function for the manual-processing subcommand.

Reads an optional custom DAG configuration from a JSON file and submits a ManualProcessing event for the given applicable date(s).

Parameters:

parsed_args (argparse.Namespace) – The parsed object of CLI arguments.

libera_utils.aws.manual_processing.start_manual_processing(applicable_dates: list[str | date | datetime], *, boto_session: Session, custom_dag_config: dict | None = None, start_processing_step_ids: list[str | ProcessingStepIdentifier] | None = None, process_downstream: bool = True, job_ids: list[ULID] | None = None, verify: bool = False, wait_time: float = 60) → list[ULID]#

Start a manual processing job by emitting a single ManualProcessing event to the SDC event bus.

One job graph is created per applicable date. The SDC Job Creator builds the job(s) from the event and the rest of the SDC orchestration runs them. Input products are assumed to already be ingested into the SDC.

Parameters:

  • applicable_dates (list of str, date, or datetime) – The applicable dates to process data for. A separate job is created for each date.

  • boto_session (boto3.Session) – Boto3 session used for all AWS interactions. Created once by the CLI handler (with the LiberaUtils role assumed) and passed in so the same authenticated session is used throughout.

  • custom_dag_config (dict, optional) – A custom DAG configuration ({"nodes": {...}} with kebab-case node keys). When None (default) the SDC’s static default DAG is used.

  • start_processing_step_ids (list of str or ProcessingStepIdentifier, optional) – Entry nodes for the job graph. When None (default) the Job Creator starts from the DAG’s root nodes.

  • process_downstream (bool, optional) – When True (default) the job graph includes all downstream descendants of the start nodes. When False, only the start nodes themselves are included – the simple way to run a single node from the default DAG.

  • job_ids (list[ULID], optional) – Caller-supplied job ids, paired positionally with applicable_dates (lengths must match). When None and verify is True, fresh ulids are minted so the created jobs can be polled for. When None and verify is False, the Job Creator mints its own ids and none are sent.

  • verify (bool, optional) – When True, poll the Coordination Table to confirm each job was created. This requires job ids to be known ahead of time, so they are minted when not supplied. Defaults to False. Note that the step function console URL(s) are logged regardless of this flag.

  • wait_time (float, optional) – Maximum seconds to wait when verifying job creation, by default 60.

Returns:

The job ids associated with the submitted jobs (empty if none were supplied or minted).

Return type:

list[ULID]

libera_utils.aws.manual_processing.step_function_trigger(algorithm_name: str | ProcessingStepIdentifier, applicable_day: str | date | datetime, *, boto_session: Session, verify: bool = False, wait_time: float = 60) → list[ULID]#

Manually trigger a single processing step for a single applicable date.

This is a thin special case of start_manual_processing(): it submits a ManualProcessing event using the SDC’s default DAG with process_downstream=False and the single step as the start node, which the Job Creator reduces to a one-node job. Input/output products for the node come from the SDC’s static DAG asset, so no custom DAG is needed. Input products are assumed to already be ingested into the SDC.

Parameters:

  • algorithm_name (str or ProcessingStepIdentifier) – The processing step to run.

  • applicable_day (str, date, or datetime) – The day of data to process.

  • boto_session (boto3.Session) – Boto3 session used for all AWS interactions.

  • verify (bool, optional) – When True, poll the Coordination Table to confirm the job was created and that its single start node moves from PENDING to RUNNING. Defaults to False. The step function console URL is logged regardless of this flag.

  • wait_time (float, optional) – Maximum seconds to wait for each verification stage (job creation, then start node running), by default 60.

Returns:

The job ids associated with the submitted job.

Return type:

list[ULID]

libera_utils.aws.manual_processing.step_function_trigger_cli_handler(parsed_args: Namespace) → None#

CLI handler function for the step-function-trigger subcommand.

Parameters:

parsed_args (argparse.Namespace) – The parsed object of CLI arguments.