AIP-81 Make list operations to return all results #50132

Prab-27 · 2025-05-02T18:14:55Z

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

Prab-27 · 2025-05-06T13:57:33Z

@bugraoz93 whenever you have some free time, could you review this PR?

bugraoz93

Thanks for the PR @Prab-27 ! We don't need this change with .all(). Paginated Select method already handles that for us, we need to use those parameters and handle this in airflowctl operations.
Let's take the first change as an example,

airflow/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py

Lines 116 to 136 in 3b7f392

    
           @assets_router.get( 
        
               "/assets", 
        
               responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]), 
        
               dependencies=[ 
        
                   Depends(requires_access_asset(method="GET")), 
        
                   Depends(requires_access_asset_alias(method="GET")), 
        
               ], 
        
           ) 
        
           def get_assets( 
        
               limit: QueryLimit, 
        
               offset: QueryOffset, 
        
               name_pattern: QueryAssetNamePatternSearch, 
        
               uri_pattern: QueryUriPatternSearch, 
        
               dag_ids: QueryAssetDagIdPatternSearch, 
        
               only_active: Annotated[OnlyActiveFilter, Depends(OnlyActiveFilter.depends)], 
        
               order_by: Annotated[ 
        
                   SortParam, 
        
                   Depends(SortParam(["id", "name", "uri", "created_at", "updated_at"], AssetModel).dynamic_depends()), 
        
               ], 
        
               session: SessionDep, 
        
           ) -> AssetCollectionResponse:

That is already covered by this method. See below, this is doing a paginated select using those parameters. We need to use offset and limit parameters for paginated select in airflowctl operations.

airflow/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py

Lines 170 to 177 in 3b7f392

    
           assets_select, total_entries = paginated_select( 
        
               statement=assets_select_statement, 
        
               filters=[only_active, name_pattern, uri_pattern, dag_ids], 
        
               order_by=order_by, 
        
               offset=offset, 
        
               limit=limit, 
        
               session=session, 
        
           )

Prab-27 · 2025-05-28T11:24:09Z

cc: @bugraoz93

bugraoz93

Hey @Prab-27, thanks for the changes!

As we discussed in Slack in the below message and in my previous comment. We shouldn't update the API and return everything. We should already leverage the paginated selects from the endpoint and tweak the offset and limit with a loop. This will make the system more resilient

https://apache-airflow.slack.com/archives/C06K9Q5G2UA/p1746635271408909?thread_ts=1746635271.408909&cid=C06K9Q5G2UA

#50132 (review)

Please also see the Slack thread where this again brought up, and we have not gone with direction in any solution.
https://apache-airflow.slack.com/archives/C07813CNKA8/p1747494203440959?thread_ts=1747494203.440959&cid=C07813CNKA8

Prab-27 · 2025-05-28T17:11:21Z

Yeah, I’m sorry. I will update it based on the discussion. Somehow, I had a misconception

bugraoz93 · 2025-05-29T21:44:47Z

Yeah, I’m sorry. I will update it based on the discussion. Somehow, I had a misconception

All good. Thanks for your contributions!

pierrejeambrun

We shouldn't update the API and return everything. We should already leverage the paginated selects from the endpoint and tweak the offset and limit with a loop. This will make the system more resilient

+1 for that. Also I don't know what the client code looks like, but you can leverage httpx or any other async http lib to be able to fetch multiple pages "at a time". For instance 10 pages, each page with a limit of 100. That's 1000 items for each loop basically.

Prab-27 · 2025-06-10T17:04:05Z

Got it !!

Only a few changes are left , I'll update them tomorrow as soon as possible

Prab-27 · 2025-06-11T09:44:36Z

@bugraoz93 is this okay ? If so, I'll fix the static checks and add tests for it

airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.py

pierrejeambrun

I think we need a couple of tests for return_all_entries to be sure that the logic is correct, fetching all the pages as expected depending on different total_entries offset and limit value.

airflow-ctl/src/airflowctl/api/operations.py

bugraoz93

Most parts already covered by Pierre, thanks a lot! I have a single additional comment to optimise the process for all list operations

bugraoz93 · 2025-06-14T09:23:10Z

airflow-ctl/src/airflowctl/api/operations.py

        """List all pools."""
        try:
            self.response = self.client.get("pools")
-            return PoolCollectionResponse.model_validate_json(self.response.content)
+            total_entries = PoolCollectionResponse.model_validate_json(self.response.content).total_entries


We are doing an additional call here. It would be great to include the first set of results, 50 as default, in the results. In return_all_entries, we can start from offset 50 and merge the result. Even just extending what was returned from the response, because the first 50 is already returned as a list. This can reduce at least one call per list operation. Because this is not only returning total entries, but the first 50 records from the database. We can even pass the first batch to the method

Prab-27 · 2025-06-18T19:05:51Z

Some tests need updating . I’ll do my best to make the changes by tomorrow.

resolve conflict

…n.py and restore endpoints - pools.py, variables.py, job.py

… tests

uranusjr · 2025-06-20T03:32:53Z

airflow-ctl/src/airflowctl/api/operations.py

+        if params is None:
+            params = {}
+        params.update({"limit": limit}, **kwargs)


Probably not consequential, but updating params in-place is somewhat surprising. I’d do something like

shared_params = {"limit": limit, **(params or {}), **kwargs} while ...: loop_params = {**shared_params, "offset": offset}

Also, the kwargs feature seems somewhat unnecessary. I would just get rid of it and require everything passed inside params.

uranusjr · 2025-06-20T03:33:59Z

airflow-ctl/src/airflowctl/api/operations.py

+        except ServerResponseError as e:
+            raise e


This is useless. Do you plan to handle the exception in the future?

We need to handle the exceptions closer to the cli layer to prevent custom logging on the API end. So we are generally throwing here, and whoever calls in ctl/cli catch it with proper errors. Do you think we should handle them here?

Do I need to remove the try-except block from the return_all_list method, since all list methods like pools list - already covered this exception?

I think we should move it to a generic method and not add them into individual methods and abstracting entire calls into that method should easily eliminate duplicate calls of the same error. Both works for now, you can also do it elsewhere. We need to refactor a bit how we throw the exceptions in the future, where there are lots of duplicate calls that can be generalised

boring-cyborg bot added the area:API Airflow's REST/HTTP API label May 2, 2025

Prab-27 force-pushed the aip-81-make-list-operations-to-return-all-results branch from 42d0710 to 282fbe1 Compare May 6, 2025 13:44

bugraoz93 reviewed May 6, 2025

View reviewed changes

bugraoz93 added the area:airflow-ctl label May 13, 2025

Prab-27 force-pushed the aip-81-make-list-operations-to-return-all-results branch from 282fbe1 to 73fde34 Compare May 19, 2025 12:46

Prab-27 force-pushed the aip-81-make-list-operations-to-return-all-results branch from 73fde34 to 369cbd0 Compare May 28, 2025 11:20

Prab-27 marked this pull request as ready for review May 28, 2025 11:23

Prab-27 requested review from ephraimbuddy, pierrejeambrun, rawwar, jason810496 and shubhamraj-git as code owners May 28, 2025 11:23

bugraoz93 requested changes May 28, 2025

View reviewed changes

pierrejeambrun reviewed Jun 2, 2025

View reviewed changes

Prab-27 force-pushed the aip-81-make-list-operations-to-return-all-results branch from 369cbd0 to e958d7e Compare June 11, 2025 09:37

Prab-27 requested review from kaxil and potiuk as code owners June 11, 2025 09:37

pierrejeambrun reviewed Jun 12, 2025

View reviewed changes

airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.py Outdated Show resolved Hide resolved

Prab-27 force-pushed the aip-81-make-list-operations-to-return-all-results branch from e958d7e to c57eab2 Compare June 12, 2025 17:39

pierrejeambrun reviewed Jun 13, 2025

View reviewed changes

bugraoz93 reviewed Jun 14, 2025

View reviewed changes

Prab-27 force-pushed the aip-81-make-list-operations-to-return-all-results branch from c57eab2 to 0d397ac Compare June 18, 2025 18:54

ensure total_entries return all entries for corresponding endpoints

df02854

Prab-27 added 15 commits June 19, 2025 04:09

use API endpoint parameters limit and offset to retrieve all entries

b50896e

resolve conflict

use a common fun to get all values

0f3fb7b

Use a common fun for return all results

07fb98a

create a method in BaseOperation and use in Pools list, restore commo…

50a1552

…n.py and restore endpoints - pools.py, variables.py, job.py

improve looping and fix static checks

eade6e9

restore endpoints and fix static checks for job.py

b3c2c39

fix static checks from common.py

a2f71c7

update offset params

0ffa6f8

update default limit = 50 to limit

4da3c57

remove equal condition from while loop and remove unnecessary params

1007e9c

use first pass response and merge it with the return list

33ba6a7

define missing parameters-type in return_all_entries method

30c0d57

add tests for return_all_entries and update list in pool and variable…

fe1e99f

… tests

fix tests

660c8d3

fix static checks

a1a034e

Prab-27 force-pushed the aip-81-make-list-operations-to-return-all-results branch from 0d397ac to a1a034e Compare June 19, 2025 18:31

uranusjr reviewed Jun 20, 2025

View reviewed changes

	@assets_router.get(
	"/assets",
	responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
	dependencies=[
	Depends(requires_access_asset(method="GET")),
	Depends(requires_access_asset_alias(method="GET")),
	],
	)
	def get_assets(
	limit: QueryLimit,
	offset: QueryOffset,
	name_pattern: QueryAssetNamePatternSearch,
	uri_pattern: QueryUriPatternSearch,
	dag_ids: QueryAssetDagIdPatternSearch,
	only_active: Annotated[OnlyActiveFilter, Depends(OnlyActiveFilter.depends)],
	order_by: Annotated[
	SortParam,
	Depends(SortParam(["id", "name", "uri", "created_at", "updated_at"], AssetModel).dynamic_depends()),
	],
	session: SessionDep,
	) -> AssetCollectionResponse:

	assets_select, total_entries = paginated_select(
	statement=assets_select_statement,
	filters=[only_active, name_pattern, uri_pattern, dag_ids],
	order_by=order_by,
	offset=offset,
	limit=limit,
	session=session,
	)

AIP-81 Make list operations to return all results #50132

Are you sure you want to change the base?

AIP-81 Make list operations to return all results #50132

Conversation

Prab-27 commented May 2, 2025

Uh oh!

Prab-27 commented May 6, 2025

Uh oh!

bugraoz93 left a comment

Choose a reason for hiding this comment

Uh oh!

Prab-27 commented May 28, 2025

Uh oh!

bugraoz93 left a comment

Choose a reason for hiding this comment

Uh oh!

Prab-27 commented May 28, 2025

Uh oh!

bugraoz93 commented May 29, 2025

Uh oh!

pierrejeambrun left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Prab-27 commented Jun 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Prab-27 commented Jun 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

pierrejeambrun left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

bugraoz93 left a comment

Choose a reason for hiding this comment

Uh oh!

bugraoz93 Jun 14, 2025

Choose a reason for hiding this comment

Uh oh!

Prab-27 commented Jun 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

uranusjr Jun 20, 2025

Choose a reason for hiding this comment

Uh oh!

uranusjr Jun 20, 2025

Choose a reason for hiding this comment

Uh oh!

bugraoz93 Jun 20, 2025

Choose a reason for hiding this comment

Uh oh!

Prab-27 Jun 21, 2025

Choose a reason for hiding this comment

Uh oh!

bugraoz93 Jun 30, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

pierrejeambrun left a comment •

edited

Loading

Prab-27 commented Jun 10, 2025 •

edited

Loading

Prab-27 commented Jun 11, 2025 •

edited

Loading

Prab-27 commented Jun 18, 2025 •

edited

Loading