Completed D400 for airflow/callbacks/* airflow/cli/*

airflow/callbacks/callback_requests.py

@@ -61,6 +61,8 @@ def from_json(cls, json_str: str):
 
 class TaskCallbackRequest(CallbackRequest):
     """
+    Task callback status information.
+
     A Class with information about the success/failure TI callback to be executed. Currently, only failure
     callbacks (when tasks are externally killed) and Zombies are run via DagFileProcessorProcess.
 

airflow/cli/cli_parser.py

@@ -16,7 +16,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-"""Command-line interface"""
+"""Command-line interface."""
 from __future__ import annotations
 
 import argparse
@@ -44,7 +44,7 @@
 
 
 def lazy_load_command(import_path: str) -> Callable:
-    """Create a lazy loader for command"""
+    """Create a lazy loader for command."""
     _, _, name = import_path.rpartition(".")
 
     def command(*args, **kwargs):
@@ -57,10 +57,10 @@ def command(*args, **kwargs):
 
 
 class DefaultHelpParser(argparse.ArgumentParser):
-    """CustomParser to display help message"""
+    """CustomParser to display help message."""
 
     def _check_value(self, action, value):
-        """Override _check_value and check conditionally added command"""
+        """Override _check_value and check conditionally added command."""
         if action.dest == "subcommand" and value == "celery":
             executor = conf.get("core", "EXECUTOR")
             if executor not in (CELERY_EXECUTOR, CELERY_KUBERNETES_EXECUTOR):
@@ -105,7 +105,7 @@ def _check_value(self, action, value):
         super()._check_value(action, value)
 
     def error(self, message):
-        """Override error and use print_instead of print_usage"""
+        """Override error and use print_instead of print_usage."""
         self.print_help()
         self.exit(2, f"\n{self.prog} command error: {message}, see help above.\n")
 
@@ -115,7 +115,7 @@ def error(self, message):
 
 
 class Arg:
-    """Class to keep information about command line argument"""
+    """Class to keep information about command line argument."""
 
     def __init__(
         self,
@@ -141,7 +141,7 @@ def __init__(
             self.kwargs[k] = v
 
     def add_to_parser(self, parser: argparse.ArgumentParser):
-        """Add this argument to an ArgumentParser"""
+        """Add this argument to an ArgumentParser."""
         parser.add_argument(*self.flags, **self.kwargs)
 
 
@@ -163,12 +163,12 @@ def _check(value):
 
 
 def string_list_type(val):
-    """Parses comma-separated list and returns list of string (strips whitespace)"""
+    """Parses comma-separated list and returns list of string (strips whitespace)."""
     return [x.strip() for x in val.split(",")]
 
 
 def string_lower_type(val):
-    """Lowers arg"""
+    """Lowers arg."""
     if not val:
         return
     return val.strip().lower()
@@ -982,7 +982,7 @@ def string_lower_type(val):
 
 
 class ActionCommand(NamedTuple):
-    """Single CLI command"""
+    """Single CLI command."""
 
     name: str
     help: str
@@ -993,7 +993,7 @@ class ActionCommand(NamedTuple):
 
 
 class GroupCommand(NamedTuple):
-    """ClI command with subcommands"""
+    """ClI command with subcommands."""
 
     name: str
     help: str
@@ -2148,7 +2148,7 @@ def _format_action(self, action: Action):
 
 @lru_cache(maxsize=None)
 def get_parser(dag_parser: bool = False) -> argparse.ArgumentParser:
-    """Creates and returns command line argument parser"""
+    """Creates and returns command line argument parser."""
     parser = DefaultHelpParser(prog="airflow", formatter_class=AirflowHelpFormatter)
     subparsers = parser.add_subparsers(dest="subcommand", metavar="GROUP_OR_COMMAND")
     subparsers.required = True
@@ -2163,10 +2163,10 @@ def get_parser(dag_parser: bool = False) -> argparse.ArgumentParser:
 
 
 def _sort_args(args: Iterable[Arg]) -> Iterable[Arg]:
-    """Sort subcommand optional args, keep positional args"""
+    """Sort subcommand optional args, keep positional args."""
 
     def get_long_option(arg: Arg):
-        """Get long option from Arg.flags"""
+        """Get long option from Arg.flags."""
         return arg.flags[0] if len(arg.flags) == 1 else arg.flags[1]
 
     positional, optional = partition(lambda x: x.flags[0].startswith("-"), args)

airflow/cli/commands/celery_command.py

@@ -15,7 +15,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-"""Celery command"""
+"""Celery command."""
 from __future__ import annotations
 
 from multiprocessing import Process
@@ -39,7 +39,7 @@
 
 @cli_utils.action_cli
 def flower(args):
-    """Starts Flower, Celery monitoring tool"""
+    """Starts Flower, Celery monitoring tool."""
     options = [
         "flower",
         conf.get("celery", "BROKER_URL"),
@@ -84,7 +84,7 @@ def flower(args):
 
 
 def _serve_logs(skip_serve_logs: bool = False) -> Process | None:
-    """Starts serve_logs sub-process"""
+    """Starts serve_logs sub-process."""
     if skip_serve_logs is False:
         sub_proc = Process(target=serve_logs)
         sub_proc.start()
@@ -103,7 +103,7 @@ def _run_worker(options, skip_serve_logs):
 
 @cli_utils.action_cli
 def worker(args):
-    """Starts Airflow Celery worker"""
+    """Starts Airflow Celery worker."""
     # Disable connection pool so that celery worker does not hold an unnecessary db connection
     settings.reconfigure_orm(disable_connection_pool=True)
     if not settings.validate_session():
@@ -205,7 +205,7 @@ def worker(args):
 
 @cli_utils.action_cli
 def stop_worker(args):
-    """Sends SIGTERM to Celery worker"""
+    """Sends SIGTERM to Celery worker."""
     # Read PID from file
     if args.pid:
         pid_file_path = args.pid

airflow/cli/commands/config_command.py

@@ -14,7 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-"""Config sub-commands"""
+"""Config sub-commands."""
 from __future__ import annotations
 
 import io
@@ -28,7 +28,7 @@
 
 
 def show_config(args):
-    """Show current application configuration"""
+    """Show current application configuration."""
     with io.StringIO() as output:
         conf.write(output)
         code = output.getvalue()
@@ -38,7 +38,7 @@ def show_config(args):
 
 
 def get_value(args):
-    """Get one value from configuration"""
+    """Get one value from configuration."""
     if not conf.has_section(args.section):
         raise SystemExit(f"The section [{args.section}] is not found in config.")
 

airflow/cli/commands/connection_command.py

@@ -14,7 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-"""Connection sub-commands"""
+"""Connection sub-commands."""
 from __future__ import annotations
 
 import io
@@ -74,7 +74,7 @@ def connections_get(args):
 
 @suppress_logs_and_warning
 def connections_list(args):
-    """Lists all connections at the command line"""
+    """Lists all connections at the command line."""
     with create_session() as session:
         query = session.query(Connection)
         if args.conn_id:
@@ -132,7 +132,7 @@ def _is_stdout(fileio: io.TextIOWrapper) -> bool:
 
 
 def _valid_uri(uri: str) -> bool:
-    """Check if a URI is valid, by checking if both scheme and netloc are available"""
+    """Check if a URI is valid, by checking if both scheme and netloc are available."""
     uri_parts = urlsplit(uri)
     return uri_parts.scheme != "" and uri_parts.netloc != ""
 
@@ -149,7 +149,7 @@ def _get_connection_types() -> list[str]:
 
 
 def connections_export(args):
-    """Exports all connections to a file"""
+    """Exports all connections to a file."""
     file_formats = [".yaml", ".json", ".env"]
     if args.format:
         warnings.warn("Option `--format` is deprecated.  Use `--file-format` instead.", DeprecationWarning)
@@ -199,7 +199,7 @@ def connections_export(args):
 
 @cli_utils.action_cli
 def connections_add(args):
-    """Adds new connection"""
+    """Adds new connection."""
     has_uri = bool(args.conn_uri)
     has_json = bool(args.conn_json)
     has_type = bool(args.conn_type)
@@ -284,7 +284,7 @@ def connections_add(args):
 
 @cli_utils.action_cli
 def connections_delete(args):
-    """Deletes connection from DB"""
+    """Deletes connection from DB."""
     with create_session() as session:
         try:
             to_delete = session.query(Connection).filter(Connection.conn_id == args.conn_id).one()
@@ -299,7 +299,7 @@ def connections_delete(args):
 
 @cli_utils.action_cli(check_db=False)
 def connections_import(args):
-    """Imports connections from a file"""
+    """Imports connections from a file."""
     if os.path.exists(args.file):
         _import_helper(args.file)
     else:

airflow/cli/commands/dag_command.py

@@ -14,7 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-"""Dag sub-commands"""
+"""Dag sub-commands."""
 from __future__ import annotations
 
 import ast
@@ -49,7 +49,7 @@
 
 @cli_utils.action_cli
 def dag_backfill(args, dag=None):
-    """Creates backfill job or dry run for a DAG or list of DAGs using regex"""
+    """Creates backfill job or dry run for a DAG or list of DAGs using regex."""
     logging.basicConfig(level=settings.LOGGING_LEVEL, format=settings.SIMPLE_LOG_FORMAT)
 
     signal.signal(signal.SIGTERM, sigint_handler)
@@ -139,7 +139,7 @@ def dag_backfill(args, dag=None):
 
 @cli_utils.action_cli
 def dag_trigger(args):
-    """Creates a dag run for the specified dag"""
+    """Creates a dag run for the specified dag."""
     api_client = get_current_api_client()
     try:
         message = api_client.trigger_dag(
@@ -152,7 +152,7 @@ def dag_trigger(args):
 
 @cli_utils.action_cli
 def dag_delete(args):
-    """Deletes all DB records related to the specified dag"""
+    """Deletes all DB records related to the specified dag."""
     api_client = get_current_api_client()
     if (
         args.yes
@@ -170,18 +170,18 @@ def dag_delete(args):
 
 @cli_utils.action_cli
 def dag_pause(args):
-    """Pauses a DAG"""
+    """Pauses a DAG."""
     set_is_paused(True, args)
 
 
 @cli_utils.action_cli
 def dag_unpause(args):
-    """Unpauses a DAG"""
+    """Unpauses a DAG."""
     set_is_paused(False, args)
 
 
 def set_is_paused(is_paused, args):
-    """Sets is_paused for DAG by a given dag_id"""
+    """Sets is_paused for DAG by a given dag_id."""
     dag = DagModel.get_dagmodel(args.dag_id)
 
     if not dag:
@@ -193,7 +193,7 @@ def set_is_paused(is_paused, args):
 
 
 def dag_dependencies_show(args):
-    """Displays DAG dependencies, save to file or show as imgcat image"""
+    """Displays DAG dependencies, save to file or show as imgcat image."""
     dot = render_dag_dependencies(SerializedDagModel.get_dag_dependencies())
     filename = args.save
     imgcat = args.imgcat
@@ -212,7 +212,7 @@ def dag_dependencies_show(args):
 
 
 def dag_show(args):
-    """Displays DAG or saves it's graphic representation to the file"""
+    """Displays DAG or saves it's graphic representation to the file."""
     dag = get_dag(args.subdir, args.dag_id)
     dot = render_dag(dag)
     filename = args.save
@@ -323,7 +323,7 @@ def dag_next_execution(args):
 @cli_utils.action_cli
 @suppress_logs_and_warning
 def dag_list_dags(args):
-    """Displays dags with or without stats at the command line"""
+    """Displays dags with or without stats at the command line."""
     dagbag = DagBag(process_subdir(args.subdir))
     if dagbag.import_errors:
         from rich import print as rich_print
@@ -348,7 +348,7 @@ def dag_list_dags(args):
 @cli_utils.action_cli
 @suppress_logs_and_warning
 def dag_list_import_errors(args):
-    """Displays dags with import errors on the command line"""
+    """Displays dags with import errors on the command line."""
     dagbag = DagBag(process_subdir(args.subdir))
     data = []
     for filename, errors in dagbag.import_errors.items():
@@ -362,7 +362,7 @@ def dag_list_import_errors(args):
 @cli_utils.action_cli
 @suppress_logs_and_warning
 def dag_report(args):
-    """Displays dagbag stats at the command line"""
+    """Displays dagbag stats at the command line."""
     dagbag = DagBag(process_subdir(args.subdir))
     AirflowConsole().print_as(
         data=dagbag.dagbag_stats,
@@ -381,7 +381,7 @@ def dag_report(args):
 @suppress_logs_and_warning
 @provide_session
 def dag_list_jobs(args, dag=None, session=NEW_SESSION):
-    """Lists latest n jobs"""
+    """Lists latest n jobs."""
     queries = []
     if dag:
         args.dag_id = dag.dag_id
@@ -411,7 +411,7 @@ def dag_list_jobs(args, dag=None, session=NEW_SESSION):
 @suppress_logs_and_warning
 @provide_session
 def dag_list_dag_runs(args, dag=None, session=NEW_SESSION):
-    """Lists dag runs for a given DAG"""
+    """Lists dag runs for a given DAG."""
     if dag:
         args.dag_id = dag.dag_id
     else:
@@ -484,6 +484,7 @@ def dag_test(args, dag=None, session=None):
 @provide_session
 @cli_utils.action_cli
 def dag_reserialize(args, session: Session = NEW_SESSION):
+    """Serialize a DAG instance."""
     session.query(SerializedDagModel).delete(synchronize_session=False)
 
     if not args.clear_only: