D401 Support - Utils

airflow/utils/cli.py

@@ -62,7 +62,7 @@ def _check_cli_args(args):
 def action_cli(func=None, check_db=True):
     def action_logging(f: T) -> T:
         """
-        Decorates function to execute function at the same time submitting action_logging but in CLI context.
+        Decorate function to execute function at the same time submitting action_logging but in CLI context.
 
         It will call action logger callbacks twice, one for
         pre-execution and the other one for post-execution.
@@ -86,7 +86,7 @@ def action_logging(f: T) -> T:
         @functools.wraps(f)
         def wrapper(*args, **kwargs):
             """
-            A wrapper for cli functions; assumes Namespace instance as first positional argument.
+            Wrap cli functions; assume Namespace instance as first positional argument.
 
             :param args: Positional argument. It assumes to have Namespace instance
                 at 1st positional argument
@@ -127,7 +127,7 @@ def wrapper(*args, **kwargs):
 
 def _build_metrics(func_name, namespace):
     """
-    Builds metrics dict from function args.
+    Build metrics dict from function args.
 
     It assumes that function arguments is from airflow.bin.cli module's function
     and has Namespace instance where it optionally contains "dag_id", "task_id",
@@ -173,7 +173,7 @@ def _build_metrics(func_name, namespace):
 
 
 def process_subdir(subdir: str | None):
-    """Expands path to absolute by replacing 'DAGS_FOLDER', '~', '.', etc."""
+    """Expand path to absolute by replacing 'DAGS_FOLDER', '~', '.', etc."""
     if subdir:
         if not settings.DAGS_FOLDER:
             raise ValueError("DAGS_FOLDER variable in settings should be filled.")
@@ -183,7 +183,7 @@ def process_subdir(subdir: str | None):
 
 
 def get_dag_by_file_location(dag_id: str):
-    """Returns DAG of a given dag_id by looking up file location."""
+    """Return DAG of a given dag_id by looking up file location."""
     from airflow.models import DagBag, DagModel
 
     # Benefit is that logging from other dags in dagbag will not appear
@@ -219,7 +219,7 @@ def _search_for_dag_file(val: str | None) -> str | None:
 
 def get_dag(subdir: str | None, dag_id: str, from_db: bool = False) -> DAG:
     """
-    Returns DAG of a given dag_id.
+    Return DAG of a given dag_id.
 
     First we'll try to use the given subdir.  If that doesn't work, we'll try to
     find the correct path (assuming it's a file) and failing that, use the configured
@@ -248,7 +248,7 @@ def get_dag(subdir: str | None, dag_id: str, from_db: bool = False) -> DAG:
 
 
 def get_dags(subdir: str | None, dag_id: str, use_regex: bool = False):
-    """Returns DAG(s) matching a given regex or dag_id."""
+    """Return DAG(s) matching a given regex or dag_id."""
     from airflow.models import DagBag
 
     if not use_regex:
@@ -276,7 +276,7 @@ def get_dag_by_pickle(pickle_id: int, session: Session = NEW_SESSION) -> DAG:
 
 
 def setup_locations(process, pid=None, stdout=None, stderr=None, log=None):
-    """Creates logging paths."""
+    """Create logging paths."""
     if not stderr:
         stderr = os.path.join(settings.AIRFLOW_HOME, f"airflow-{process}.err")
     if not stdout:
@@ -293,7 +293,7 @@ def setup_locations(process, pid=None, stdout=None, stderr=None, log=None):
 
 
 def setup_logging(filename):
-    """Creates log file handler for daemon process."""
+    """Create log file handler for daemon process."""
     root = logging.getLogger()
     handler = NonCachingFileHandler(filename)
     formatter = logging.Formatter(settings.SIMPLE_LOG_FORMAT)
@@ -306,7 +306,7 @@ def setup_logging(filename):
 
 def sigint_handler(sig, frame):
     """
-    Returns without error on SIGINT or SIGTERM signals in interactive command mode.
+    Return without error on SIGINT or SIGTERM signals in interactive command mode.
 
     e.g. CTRL+C or kill <PID>
     """
@@ -315,7 +315,7 @@ def sigint_handler(sig, frame):
 
 def sigquit_handler(sig, frame):
     """
-    Helps debug deadlocks by printing stacktraces when this gets a SIGQUIT.
+    Help debug deadlocks by printing stacktraces when this gets a SIGQUIT.
 
     e.g. kill -s QUIT <PID> or CTRL+
     """
@@ -340,7 +340,7 @@ class ColorMode:
 
 
 def should_use_colors(args) -> bool:
-    """Processes arguments and decides whether to enable color in output."""
+    """Process arguments and decide whether to enable color in output."""
     if args.color == ColorMode.ON:
         return True
     if args.color == ColorMode.OFF:
@@ -361,7 +361,7 @@ def should_ignore_depends_on_past(args) -> bool:
 
 
 def suppress_logs_and_warning(f: T) -> T:
-    """Decorator to suppress logging and warning messages in cli functions."""
+    """Suppress logging and warning messages in cli functions."""
 
     @functools.wraps(f)
     def _wrapper(*args, **kwargs):

airflow/utils/cli_action_loggers.py

@@ -29,7 +29,7 @@
 
 
 def register_pre_exec_callback(action_logger):
-    """Registers more action_logger function callback for pre-execution.
+    """Register more action_logger function callback for pre-execution.
 
     This function callback is expected to be called with keyword args.
     For more about the arguments that is being passed to the callback,
@@ -43,7 +43,7 @@ def register_pre_exec_callback(action_logger):
 
 
 def register_post_exec_callback(action_logger):
-    """Registers more action_logger function callback for post-execution.
+    """Register more action_logger function callback for post-execution.
 
     This function callback is expected to be called with keyword args.
     For more about the arguments that is being passed to the callback,
@@ -57,7 +57,7 @@ def register_post_exec_callback(action_logger):
 
 
 def on_pre_execution(**kwargs):
-    """Calls callbacks before execution.
+    """Call callbacks before execution.
 
     Note that any exception from callback will be logged but won't be propagated.
 
@@ -73,7 +73,7 @@ def on_pre_execution(**kwargs):
 
 
 def on_post_execution(**kwargs):
-    """Calls callbacks after execution.
+    """Call callbacks after execution.
 
     As it's being called after execution, it can capture status of execution,
     duration, etc. Note that any exception from callback will be logged but
@@ -91,7 +91,8 @@ def on_post_execution(**kwargs):
 
 
 def default_action_log(sub_command, user, task_id, dag_id, execution_date, host_name, full_command, **_):
-    """Default action logger callback that behaves similar to ``action_logging``.
+    """
+    Behave similar to ``action_logging``; default action logger callback.
 
     The difference is this function uses the global ORM session, and pushes a
     ``Log`` row into the database instead of actually logging.

airflow/utils/cli_app_builder.py

@@ -31,7 +31,7 @@
 
 @lru_cache(maxsize=None)
 def _return_appbuilder(app: Flask) -> AirflowAppBuilder:
-    """Returns an appbuilder instance for the given app."""
+    """Return an appbuilder instance for the given app."""
     init_appbuilder(app)
     init_plugins(app)
     return app.appbuilder  # type: ignore[attr-defined]

airflow/utils/code_utils.py

@@ -23,7 +23,7 @@
 
 
 def get_python_source(x: Any) -> str | None:
-    """Helper function to get Python source (or not), preventing exceptions."""
+    """Get Python source (or not), preventing exceptions."""
     if isinstance(x, str):
         return x
 
@@ -80,7 +80,7 @@ def prepare_code_snippet(file_path: str, line_no: int, context_lines_count: int
 
 
 def get_terminal_formatter(**opts):
-    """Returns the best formatter available in the current terminal."""
+    """Return the best formatter available in the current terminal."""
     if "256" in os.environ.get("TERM", ""):
         from pygments.formatters.terminal256 import Terminal256Formatter
 

airflow/utils/configuration.py

@@ -27,7 +27,7 @@
 
 def tmp_configuration_copy(chmod=0o600, include_env=True, include_cmds=True):
     """
-    Returns a path for a temporary file including a full copy of the configuration settings.
+    Return a path for a temporary file including a full copy of the configuration settings.
 
     :param include_env: Should the value of configuration from ``AIRFLOW__``
         environment variables be included or not

airflow/utils/dag_cycle_tester.py

@@ -58,7 +58,7 @@ def check_cycle(dag: DAG) -> None:
     task_dict = dag.task_dict
 
     def _check_adjacent_tasks(task_id, current_task):
-        """Returns first untraversed child task, else None if all tasks traversed."""
+        """Return first untraversed child task, else None if all tasks traversed."""
         for adjacent_task in current_task.get_direct_relative_ids():
             if visited[adjacent_task] == CYCLE_IN_PROGRESS:
                 msg = f"Cycle detected in DAG: {dag.dag_id}. Faulty task: {task_id}"

airflow/utils/dates.py

@@ -140,7 +140,7 @@ def round_time(
     delta: str | timedelta | relativedelta,
     start_date: datetime = timezone.make_aware(datetime.min),
 ):
-    """Returns ``start_date + i * delta`` for given ``i`` where the result is closest to ``dt``.
+    """Return ``start_date + i * delta`` for given ``i`` where the result is closest to ``dt``.
 
     .. code-block:: pycon
 

airflow/utils/db.py

@@ -790,7 +790,7 @@ def _get_current_revision(session):
 
 def check_migrations(timeout):
     """
-    Function to wait for all airflow migrations to complete.
+    Wait for all airflow migrations to complete.
 
     :param timeout: Timeout for the migration in seconds
     :return: None
@@ -1380,11 +1380,12 @@ def _move_duplicate_data_to_new_table(
 
 def check_bad_references(session: Session) -> Iterable[str]:
     """
-    Starting in Airflow 2.2, we began a process of replacing `execution_date` with `run_id` in many tables.
+    Go through each table and look for records that can't be mapped to a dag run.
 
-    Here we go through each table and look for records that can't be mapped to a dag run.
     When we find such "dangling" rows we back them up in a special table and delete them
     from the main table.
+
+    Starting in Airflow 2.2, we began a process of replacing `execution_date` with `run_id` in many tables.
     """
     from airflow.models.dagrun import DagRun
     from airflow.models.renderedtifields import RenderedTaskInstanceFields
@@ -1534,7 +1535,7 @@ def _revision_greater(config, this_rev, base_rev):
 
 def _revisions_above_min_for_offline(config, revisions) -> None:
     """
-    Checks that all supplied revision ids are above the minimum revision for the dialect.
+    Check that all supplied revision ids are above the minimum revision for the dialect.
 
     :param config: Alembic config
     :param revisions: list of Alembic revision ids
@@ -1725,7 +1726,7 @@ def downgrade(*, to_revision, from_revision=None, show_sql_only=False, session:
 
 def drop_airflow_models(connection):
     """
-    Drops all airflow models.
+    Drop all airflow models.
 
     :param connection: SQLAlchemy Connection
     :return: None
@@ -1760,7 +1761,7 @@ def drop_airflow_moved_tables(connection):
 @provide_session
 def check(session: Session = NEW_SESSION):
     """
-    Checks if the database works.
+    Check if the database works.
 
     :param session: session of the sqlalchemy
     """

airflow/utils/decorators.py

@@ -30,7 +30,7 @@
 
 def apply_defaults(func: T) -> T:
     """
-    This decorator is deprecated.
+    Use apply_default decorator for the `default_args` feature to work properly; deprecated.
 
     In previous versions, all subclasses of BaseOperator must use apply_default decorator for the"
     `default_args` feature to work properly.
@@ -58,7 +58,7 @@ def wrapper(*args, **kwargs):
 
 def remove_task_decorator(python_source: str, task_decorator_name: str) -> str:
     """
-    Removes @task or similar decorators as well as @setup and @teardown.
+    Remove @task or similar decorators as well as @setup and @teardown.
 
     :param python_source: The python source code
     :param task_decorator_name: the decorator name

airflow/utils/dot_renderer.py

@@ -37,7 +37,7 @@
 
 def _refine_color(color: str):
     """
-    Converts color in #RGB (12 bits) format to #RRGGBB (32 bits), if it possible.
+    Convert color in #RGB (12 bits) format to #RRGGBB (32 bits), if it possible.
 
     Otherwise, it returns the original value. Graphviz does not support colors in #RGB format.
 
@@ -144,7 +144,7 @@ def _draw_nodes(
 
 def render_dag_dependencies(deps: dict[str, list[DagDependency]]) -> graphviz.Digraph:
     """
-    Renders the DAG dependency to the DOT object.
+    Render the DAG dependency to the DOT object.
 
     :param deps: List of DAG dependencies
     :return: Graphviz object
@@ -169,7 +169,7 @@ def render_dag_dependencies(deps: dict[str, list[DagDependency]]) -> graphviz.Di
 
 def render_dag(dag: DAG, tis: list[TaskInstance] | None = None) -> graphviz.Digraph:
     """
-    Renders the DAG object to the DOT object.
+    Render the DAG object to the DOT object.
 
     If an task instance list is passed, the nodes will be painted according to task statuses.
 

airflow/utils/edgemodifier.py

@@ -154,15 +154,15 @@ def set_downstream(
     def update_relative(
         self, other: DependencyMixin, upstream: bool = True, edge_modifier: EdgeModifier | None = None
     ) -> None:
-        """Called if we're not the "main" side of a relationship; we still run the same logic, though."""
+        """Update relative if we're not the "main" side of a relationship; still run the same logic."""
         if upstream:
             self.set_upstream(other)
         else:
             self.set_downstream(other)
 
     def add_edge_info(self, dag, upstream_id: str, downstream_id: str):
         """
-        Adds or updates task info on the DAG for this specific pair of tasks.
+        Add or update task info on the DAG for this specific pair of tasks.
 
         Called either from our relationship trigger methods above, or directly
         by set_upstream/set_downstream in operators.
@@ -172,5 +172,5 @@ def add_edge_info(self, dag, upstream_id: str, downstream_id: str):
 
 # Factory functions
 def Label(label: str):
-    """Creates an EdgeModifier that sets a human-readable label on the edge."""
+    """Create an EdgeModifier that sets a human-readable label on the edge."""
     return EdgeModifier(label=label)

airflow/utils/email.py

@@ -287,7 +287,7 @@ def send_mime_email(
 
 def get_email_address_list(addresses: str | Iterable[str]) -> list[str]:
     """
-    Returns a list of email addresses from the provided input.
+    Return a list of email addresses from the provided input.
 
     :param addresses: A string or iterable of strings containing email addresses.
     :return: A list of email addresses.
@@ -305,7 +305,7 @@ def get_email_address_list(addresses: str | Iterable[str]) -> list[str]:
 
 def _get_smtp_connection(host: str, port: int, timeout: int, with_ssl: bool) -> smtplib.SMTP:
     """
-    Returns an SMTP connection to the specified host and port, with optional SSL encryption.
+    Return an SMTP connection to the specified host and port, with optional SSL encryption.
 
     :param host: The hostname or IP address of the SMTP server.
     :param port: The port number to connect to on the SMTP server.

airflow/utils/empty_set.py

@@ -20,7 +20,7 @@
 
 def _get_empty_set_for_configuration() -> set[tuple[str, str]]:
     """
-    Retrieves an empty_set_for_configuration.
+    Retrieve an empty_set_for_configuration.
 
     This method is only needed because configuration module has a deprecated method called set, and it
     confuses mypy. This method will be removed when we remove the deprecated method.

airflow/utils/event_scheduler.py

@@ -31,7 +31,7 @@ def call_regular_interval(
         arguments=(),
         kwargs={},
     ):
-        """Helper to call a function at (roughly) a given interval."""
+        """Call a function at (roughly) a given interval."""
 
         def repeat(*args, **kwargs):
             action(*args, **kwargs)