Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions airflow/providers/cncf/kubernetes/hooks/kubernetes.py
Original file line number Diff line number Diff line change
Expand Up @@ -191,7 +191,7 @@ def get_conn(self) -> client.ApiClient:
cluster_context = self._coalesce_param(self.cluster_context, self._get_field("cluster_context"))
kubeconfig_path = self._coalesce_param(self.config_file, self._get_field("kube_config_path"))
kubeconfig = self._get_field("kube_config")
num_selected_configuration = len([o for o in [in_cluster, kubeconfig, kubeconfig_path] if o])
num_selected_configuration = sum(1 for o in [in_cluster, kubeconfig, kubeconfig_path] if o)

if num_selected_configuration > 1:
raise AirflowException(
Expand Down Expand Up @@ -474,7 +474,7 @@ async def _load_config(self):
kubeconfig_path = self._coalesce_param(self.config_file, await self._get_field("kube_config_path"))
kubeconfig = await self._get_field("kube_config")

num_selected_configuration = len([o for o in [in_cluster, kubeconfig, kubeconfig_path] if o])
num_selected_configuration = sum(1 for o in [in_cluster, kubeconfig, kubeconfig_path] if o)

if num_selected_configuration > 1:
raise AirflowException(
Expand Down
6 changes: 3 additions & 3 deletions airflow/providers/cncf/kubernetes/utils/pod_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -443,9 +443,7 @@ def fetch_requested_container_logs(
"""
pod_logging_statuses = []
all_containers = self.get_container_names(pod)
if len(all_containers) == 0:
self.log.error("Could not retrieve containers for the pod: %s", pod.metadata.name)
else:
if all_containers:
if isinstance(container_logs, str):
# fetch logs only for requested container if only one container is provided
if container_logs in all_containers:
Expand Down Expand Up @@ -490,6 +488,8 @@ def fetch_requested_container_logs(
self.log.error(
"Invalid type %s specified for container names input parameter", type(container_logs)
)
else:
self.log.error("Could not retrieve containers for the pod: %s", pod.metadata.name)

return pod_logging_statuses

Expand Down
2 changes: 1 addition & 1 deletion airflow/providers/databricks/hooks/databricks_sql.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ def _get_sql_endpoint_by_name(self, endpoint_name) -> dict[str, Any]:
if "endpoints" not in result:
raise AirflowException("Can't list Databricks SQL endpoints")
lst = [endpoint for endpoint in result["endpoints"] if endpoint["name"] == endpoint_name]
if len(lst) == 0:
if not lst:
raise AirflowException(f"Can't f Databricks SQL endpoint with name '{endpoint_name}'")
return lst[0]

Expand Down
2 changes: 1 addition & 1 deletion airflow/providers/databricks/operators/databricks_sql.py
Original file line number Diff line number Diff line change
Expand Up @@ -297,7 +297,7 @@ def _generate_options(
escape_key: bool = True,
) -> str:
formatted_opts = ""
if opts is not None and len(opts) > 0:
if opts:
pairs = [
f"{escaper.escape_item(k) if escape_key else k} = {escaper.escape_item(v)}"
for k, v in opts.items()
Expand Down
4 changes: 2 additions & 2 deletions airflow/providers/databricks/sensors/databricks_partition.py
Original file line number Diff line number Diff line change
Expand Up @@ -184,7 +184,7 @@ def _generate_partition_query(
if len(partition_columns) < 1:
raise AirflowException(f"Table {table_name} does not have partitions")
formatted_opts = ""
if opts is not None and len(opts) > 0:
if opts:
output_list = []
for partition_col, partition_value in opts.items():
if escape_key:
Expand Down Expand Up @@ -217,7 +217,7 @@ def poke(self, context: Context) -> bool:
"""Checks the table partitions and returns the results."""
partition_result = self._check_table_partitions()
self.log.debug("Partition sensor result: %s", partition_result)
if len(partition_result) >= 1:
if partition_result:
return True
else:
raise AirflowException(f"Specified partition(s): {self.partitions} were not found.")
2 changes: 1 addition & 1 deletion airflow/providers/datadog/sensors/datadog.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,4 +96,4 @@ def poke(self, context: Context) -> bool:
return self.response_check(response)

# If no check was inserted, assume any event that matched yields true.
return len(response) > 0
return bool(response)
2 changes: 1 addition & 1 deletion airflow/providers/docker/operators/docker.py
Original file line number Diff line number Diff line change
Expand Up @@ -413,7 +413,7 @@ def _run_image_with_mounts(self, target_mounts, add_tmp_variable: bool) -> list[
if self.retrieve_output:
return self._attempt_to_retrieve_result()
elif self.do_xcom_push:
if len(log_lines) == 0:
if not log_lines:
return None
try:
if self.xcom_all:
Expand Down
2 changes: 1 addition & 1 deletion airflow/providers/imap/hooks/imap.py
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,7 @@ def has_mail_attachment(
mail_attachments = self._retrieve_mails_attachments_by_name(
name, check_regex, True, mail_folder, mail_filter
)
return len(mail_attachments) > 0
return bool(mail_attachments)

def retrieve_mail_attachments(
self,
Expand Down
2 changes: 1 addition & 1 deletion airflow/providers/openlineage/utils/sql.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ def to_dataset(self, namespace: str, database: str | None = None, schema: str |
return Dataset(
namespace=namespace,
name=name,
facets={"schema": SchemaDatasetFacet(fields=self.fields)} if len(self.fields) > 0 else {},
facets={"schema": SchemaDatasetFacet(fields=self.fields)} if self.fields else {},
)


Expand Down
6 changes: 2 additions & 4 deletions airflow/providers/oracle/transfers/oracle_to_oracle.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,13 +72,11 @@ def _execute(self, src_hook, dest_hook, context) -> None:
target_fields = list(map(lambda field: field[0], cursor.description))

rows_total = 0
rows = cursor.fetchmany(self.rows_chunk)
while rows:
rows_total += len(rows)
for rows in iter(lambda: cursor.fetchmany(self.rows_chunk), []):
dest_hook.bulk_insert_rows(
self.destination_table, rows, target_fields=target_fields, commit_every=self.rows_chunk
)
rows = cursor.fetchmany(self.rows_chunk)
rows_total += len(rows)
self.log.info("Total inserted: %s rows", rows_total)

self.log.info("Finished data transfer.")
Expand Down
2 changes: 1 addition & 1 deletion airflow/providers/plexus/operators/job.py
Original file line number Diff line number Diff line change
Expand Up @@ -144,7 +144,7 @@ def construct_job_params(self, hook: Any) -> dict[Any, Any | None]:
:param hook: plexus hook object
"""
missing_params = self.required_params - set(self.job_params)
if len(missing_params) > 0:
if missing_params:
raise AirflowException(f"Missing the following required job_params: {', '.join(missing_params)}")
params = {}
for prm in self.job_params:
Expand Down
2 changes: 1 addition & 1 deletion airflow/providers/ssh/hooks/ssh.py
Original file line number Diff line number Diff line change
Expand Up @@ -512,7 +512,7 @@ def exec_ssh_client_command(
while not channel.closed or channel.recv_ready() or channel.recv_stderr_ready():
readq, _, _ = select([channel], [], [], cmd_timeout)
if cmd_timeout is not None:
timedout = len(readq) == 0
timedout = not readq
for recv in readq:
if recv.recv_ready():
output = stdout.channel.recv(len(recv.in_buffer))
Expand Down