Skip to content

Commit

Permalink
Apply review suggestions
Browse files Browse the repository at this point in the history
  • Loading branch information
phanikumv committed Jul 3, 2023
1 parent 57c61b8 commit c56bafc
Show file tree
Hide file tree
Showing 2 changed files with 5 additions and 6 deletions.
6 changes: 3 additions & 3 deletions airflow/utils/db.py
Original file line number Diff line number Diff line change
Expand Up @@ -1102,7 +1102,7 @@ def check_conn_type_null(session: Session) -> Iterable[str]:

n_nulls = []
try:
n_nulls = session.execute(select(Connection.conn_id).filter(Connection.conn_type.is_(None))).all()
n_nulls = session.scalars(select(Connection.conn_id).where(Connection.conn_type.is_(None))).all()
except (exc.OperationalError, exc.ProgrammingError, exc.InternalError):
# fallback if tables hasn't been created yet
session.rollback()
Expand Down Expand Up @@ -1144,7 +1144,7 @@ def check_run_id_null(session: Session) -> Iterable[str]:
dagrun_table.c.run_id.is_(None),
dagrun_table.c.execution_date.is_(None),
)
invalid_dagrun_count = session.scalar(select(func.count(dagrun_table.c.id)).filter(invalid_dagrun_filter))
invalid_dagrun_count = session.scalar(select(func.count(dagrun_table.c.id)).where(invalid_dagrun_filter))
if invalid_dagrun_count > 0:
dagrun_dangling_table_name = _format_airflow_moved_table_name(dagrun_table.name, "2.2", "dangling")
if dagrun_dangling_table_name in inspect(session.get_bind()).get_table_names():
Expand Down Expand Up @@ -1335,7 +1335,7 @@ def _move_duplicate_data_to_new_table(
dialect_name = bind.dialect.name

query = (
session.select(*[getattr(source_table.c, x.name).label(str(x.name)) for x in source_table.columns])
select(*[getattr(source_table.c, x.name).label(str(x.name)) for x in source_table.columns])
.select_from(source_table)
.join(subquery, and_(*[getattr(source_table.c, x) == getattr(subquery.c, x) for x in uniqueness]))
)
Expand Down
5 changes: 2 additions & 3 deletions airflow/www/views.py
Original file line number Diff line number Diff line change
Expand Up @@ -941,7 +941,6 @@ def _iter_parsed_moved_data_table_names():
select(Log)
.where(Log.event == "robots")
.where(Log.dttm > (utcnow() - datetime.timedelta(days=7)))
# .count()
)
robots_file_access_count = session.scalar(
select(func.count()).select_from(robots_file_access_count)
Expand Down Expand Up @@ -4734,11 +4733,11 @@ def action_mulduplicate(self, connections, session: Session = NEW_SESSION):

potential_connection_ids = [f"{base_conn_id}_copy{i}" for i in range(1, 11)]

query = session.execute(
query = session.scalars(
select(Connection.conn_id).where(Connection.conn_id.in_(potential_connection_ids))
)

found_conn_id_set = {conn_id for conn_id, in query}
found_conn_id_set = {conn_id for conn_id in query}

possible_conn_id_iter = (
connection_id
Expand Down

0 comments on commit c56bafc

Please sign in to comment.