client.rules.lists.items.list - Pagination not working

[IDemixI]

Confirm this is a Python library issue and not an underlying Cloudflare API issue.

• [x] This is an issue with the Python library

Describe the bug

Hi there,

It appears as though pagination is not working for client.rules.lists.items.list(). I am only getting the first page back. I've taken a look at the source code and it appears it could be related to the removal of SyncCursorPagination? I'm rusty as a dev so I wouldn't put money on it.

Please check out this PR though (here) which looks to have made a change. For now the list I have is around 480 items so I'm getting around this with client.rules.lists.items.list(account_id=account_id, list_id=rule_id, per_page=500) but this is a stop-gap solution at best.

See "To Reproduce" for a very basic reproduction.

Any help would be greatly appreciated!

To Reproduce

import os
from cloudflare import Cloudflare

client = Cloudflare(
    api_token=os.environ.get("CLOUDFLARE_API_TOKEN"), # This is the default and can be omitted
)

ip_list = []
rule_id = "xxxxx"
account_id = "xxxxx"

for item in client.rules.lists.items.list(list_id=rule_id, account_id=account_id):
    ip_list.append(account)
print(ip_list)

Code snippets

OS

macOS

Python version

3.11.2

Library version

4.3.1

[erikaskucinskis]

Faced same problem. Had to use RAW API call as a workaround.

[amrutadotorg]

Hi everyone,

I can confirm this bug is still present. The library's automatic pagination for client.rules.lists.items.list() appears broken, only returning the first page of results.

For anyone who needs a reliable, immediate solution in Python, you can bypass the cloudflare library for this specific API call and use the requests library to implement the pagination loop manually. This approach gives you full control and works perfectly.

Here is a complete, runnable example script.

get_all_list_items.py

import os
import requests
import logging

# --- Configuration ---
# It's recommended to set these as environment variables
ACCOUNT_ID = os.environ.get("CLOUDFLARE_ACCOUNT_ID")
LIST_ID = os.environ.get("CLOUDFLARE_LIST_ID")
API_TOKEN = os.environ.get("CLOUDFLARE_API_TOKEN")

# --- Setup Logging ---
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")


def get_all_cloudflare_list_items(account_id: str, list_id: str, api_token: str) -> list:
    """
    Fetches all items from a Cloudflare IP List using manual, robust pagination.

    This function bypasses a bug in the cloudflare-python library's automatic
    iterator by making direct HTTP requests and handling the pagination cursor manually.

    Args:
        account_id: Your Cloudflare Account ID.
        list_id: The ID of the list you want to fetch items from.
        api_token: Your Cloudflare API token.

    Returns:
        A list containing all item dictionaries from the specified list.
    """
    if not all([account_id, list_id, api_token]):
        raise ValueError("Account ID, List ID, and API Token must be provided.")

    all_items = []
    page_count = 0
    # The initial URL does not contain a cursor.
    api_url = f"https://api.cloudflare.com/client/v4/accounts/{account_id}/rules/lists/{list_id}/items"
    
    headers = {
        "Authorization": f"Bearer {api_token}",
        "Content-Type": "application/json",
    }
    
    # Use a session for connection pooling and efficiency
    with requests.Session() as session:
        session.headers.update(headers)
        
        while api_url:
            page_count += 1
            logging.info(f"Fetching page {page_count} from URL: {api_url}")
            
            try:
                response = session.get(api_url)
                # Raise an exception for bad status codes (4xx or 5xx)
                response.raise_for_status()
                
                data = response.json()
                
                # Process the items from the current page
                page_items = data.get("result", [])
                if not page_items:
                    logging.info(f"Received an empty result on page {page_count}. End of list.")
                    break
                
                all_items.extend(page_items)
                logging.info(f"Fetched {len(page_items)} items. Total so far: {len(all_items)}")

                # --- Manual Pagination Logic ---
                # Check for the 'after' cursor to get the next page URL.
                cursors = data.get("result_info", {}).get("cursors", {})
                next_cursor = cursors.get("after")

                if next_cursor:
                    # Construct the URL for the next page
                    api_url = f"https://api.cloudflare.com/client/v4/accounts/{account_id}/rules/lists/{list_id}/items?cursor={next_cursor}"
                else:
                    # No more pages, exit the loop
                    logging.info("No 'after' cursor found. Pagination complete.")
                    api_url = None

            except requests.exceptions.RequestException as e:
                logging.error(f"An HTTP error occurred: {e}")
                # Optionally, you can check the response content for more details
                if e.response is not None:
                    logging.error(f"Error details: {e.response.text}")
                break # Exit loop on error
                
    return all_items


if __name__ == "__main__":
    logging.info("Starting Cloudflare list item fetcher...")
    
    # Example usage:
    # Ensure you have set the following environment variables:
    # CLOUDFLARE_ACCOUNT_ID, CLOUDFLARE_LIST_ID, CLOUDFLARE_API_TOKEN
    
    if not all([ACCOUNT_ID, LIST_ID, API_TOKEN]):
        logging.error("Please set the required environment variables: CLOUDFLARE_ACCOUNT_ID, CLOUDFLARE_LIST_ID, CLOUDFLARE_API_TOKEN")
    else:
        try:
            total_items = get_all_cloudflare_list_items(ACCOUNT_ID, LIST_ID, API_TOKEN)
            logging.info(f"\n--- SCRIPT COMPLETE ---")
            logging.info(f"Successfully fetched a total of {len(total_items)} items from the list.")
            
            # You can now work with the `total_items` list.
            # For example, print the first 5 items:
            if total_items:
                logging.info("\nFirst 5 items:")
                for item in total_items[:5]:
                    print(item)

        except ValueError as e:
            logging.error(e)
        except Exception as e:
            logging.error(f"An unexpected error occurred: {e}")

How to Use This Script

1. Save the code as a Python file (e.g., get_all_list_items.py).
2. Install the requests library if you haven't already:

   pip install requests
   

3. Set the required environment variables in your terminal:

   export CLOUDFLARE_ACCOUNT_ID="YOUR_ACCOUNT_ID"
   export CLOUDFLARE_LIST_ID="YOUR_LIST_ID"
   export CLOUDFLARE_API_TOKEN="YOUR_API_TOKEN"
   

4. Run the script:

   python get_all_list_items.py
   

This provides a clear, working, and self-contained example that directly addresses the problem and can be easily adapted by anyone affected by this bug.

[fholzer]

Here an alternative that uses the client's built-in with_raw_response mechanism to get the cursor value for manual paging.

def fetchListItems(client, accountId, listId, per_page = 100):
    result = []
    next = None
    while True:
        response = client.rules.lists.items.with_raw_response.list(account_id=accountId, list_id=listId, cursor=next, per_page=per_page)
        result += response.parse()
        raw = response.json()
        next = raw["result_info"]["cursors"]["after"] if raw and "result_info" in raw and "cursors" in raw["result_info"] and "after" in raw["result_info"]["cursors"] else None
        if next is None:
            return result

res = fetchListItems(client, accountId, listId)