A command-line interface (CLI) lets a program receive a command and its arguments as text. In a data pipeline, this gives a containerized task a clear entry point: the task runs a named command, passes the inputs for that run, and leaves the processing logic in the program inside the image. section 11.4.2 section 14.5
For a ratings-fetching task, a Python function named `main` can become a CLI command through `click.command`. `click.option` declares inputs such as required start and end dates, an output path, API credentials, and a batch size. Click parses and validates those arguments before `main` uses them to request ratings and write a file. Python’s built-in `argparse` is an alternative; Click is used here for its concise API. section 11.4.2
Listing 11.10 Skeleton for a Python CLI script based on the click library
#!/usr/bin/env python #1
import logging
import click
logging.basicConfig(level=logging.INFO) #2
@click.command() #3
@click.option( #4
"--start_date",
type=click.DateTime(formats=["%Y-%m-%d"]),
required=True,
help="Start date for ratings.",
)
@click.option( #5
...
)
@click.option(
...
)
...
def main(start_date, ...): #6
"""CLI script for fetching ratings from the movielens API."""
...
if __name__ == "__main__": #7
main()
The scaffold shows where the command is declared, how a required date option is typed, and how script execution reaches `main`. section 11.4.2
Check the container command
After adding the script and its dependencies to a Docker image, build the image and run `docker run --rm manning-airflow/movielens-fetch fetch-ratings --help`. The generated help lists accepted options, required inputs, date formats, and descriptions. This checks the command interface before supplying real API credentials or fetching ratings. section 11.4.2
A CLI can also select which operation a shared image performs. A daily recipe pipeline uses a Python package with a Typer command named `preprocess`. Its Airflow `DockerOperator` passes `preprocess` followed by a MinIO path containing the run’s templated `data_interval_start` date. The command loads that day’s raw recipes, creates chunks and UUIDs, and saves a preprocessed DataFrame. Shared DockerOperator settings are kept in a dictionary rather than repeated across tasks. section 14.5
Listing 14.1 Calling the CLI command to preprocess the recipe from our task
import pendulum
from airflow.providers.docker.operators.docker import DockerOperator
from airflow.timetables.trigger import CronTriggerTimetable
from airflow import DAG
ENVIRONMENT = {
# All our environment variables
}
docker_kwargs= { #1
"image":" vectorvault_cli:latest",
"environment":ENVIRONMENT,
"docker_url":"tcp://docker-socket-proxy:2375",
"network_mode":"chapter14_default",
"environment":ENVIRONMENT,
"auto_remove":"success",
"tty":True,
}
with DAG(
dag_id=" 01_ingestion",
schedule=CronTriggerTimetable("@daily", timezone="UTC"),
start_date=pendulum.today("UTC"),
):
...
preprocess_recipes = DockerOperator(
task_id="preprocess_recipes",
command=[
"preprocess", #2
"s3://data/{{data_interval_start | ds}}", #3
],
**docker_kwargs #4
)
The task configuration shows the boundary between scheduling a container and passing a command with a date-specific data path into it. section 14.5
