Cloudflare Python Library

repository·main·Indexed 19 days ago

https://github.com/cloudflare/cloudflare-python

The official Python library for the Cloudflare REST API, providing type-safe access for Python 3.9+ applications. It supports both synchronous (Cloudflare) and asynchronous (AsyncCloudflare) workflows, with optional aiohttp support for improved concurrency. The library includes built-in handling for paginated results, automatic retries with exponential backoff, and comprehensive error mapping for API status codes.

Tokens
179.8K
Snippets
689
Records
838
Agent score
66%

What's inside cloudflare-python

  1. Manage Brand Protection core resources

    main

    The Brand Protection API allows you to manage brand security through several sub-resources: submit, queries, matches, logos, and logo_matches.

    • Submit: Use client.brand_protection.submit(account_id=...) to submit brand protection requests and client.brand_protection.url_info(account_id=...) to retrieve URL information.
    • Queries: Manage brand protection queries using client.brand_protection.queries.create(...), delete(...), or bulk(...).
    • Matches: Retrieve or download detected matches via client.brand_protection.matches.get(...) or client.brand_protection.matches.download(...).
    • Logos: Manage brand logos with client.brand_protection.logos.create(...) and client.brand_protection.logos.delete(logo_id, account_id=...).
    • LogoMatches: Access logo-specific matches using client.brand_protection.logo_matches.get(...) or client.brand_protection.logo_matches.download(...).
  2. Update D1 database configuration using update() or edit()

    main

    v5 separates full configuration updates from partial updates to follow RESTful conventions:

    1. update() (Full Update/PUT): Use this to replace the entire configuration. It specifically handles read replication configuration and requires database_id as the first positional parameter.
    2. edit() (Partial Update/PATCH): Use this to modify specific fields without replacing the whole object.

    Both methods currently support read_replication configuration.

    # Full update (PUT) - specifically for read replication
    result = client.d1.database.update(
        database_id="def456",
        account_id="abc123",
        read_replication={
            "enabled": True,
            "regions": ["weur", "enam"]
        }
    )
    
    # Partial update (PATCH)
    result = client.d1.database.edit(
        database_id="def456",
        account_id="abc123",
        read_replication={
            "enabled": True
        }
    )
  3. Differentiate between `null` and missing fields in API responses

    main

    In API responses, both an explicit null value and a missing key are represented as None in the library. To distinguish between them, check the .model_fields_set attribute on the response object. If the field name is not in .model_fields_set, the key was missing from the JSON; otherwise, it was explicitly null.

    if response.my_field is None:
      if 'my_field' not in response.model_fields_set:
        print('Got json like {}, without a "my_field" key present at all.'.format(response))
      else:
        print('Got json like {"my_field": null}.')
  4. Handle paginated list results

    main

    List methods in the Cloudflare API are paginated. The library provides several ways to handle this:

    1. Auto-paginating iterators: Use a for loop (sync) or async for loop (async) to automatically fetch all pages.
    2. Granular control: Use .has_next_page(), .next_page_info(), or .get_next_page() to manually manage page transitions.
    3. Direct access: Access the .result attribute of the returned page object to iterate over the current page's items.
    # Auto-paginating (Sync)
    for account in client.accounts.list():
        print(account)
    
    # Auto-paginating (Async)
    async for account in client.accounts.list():
        print(account)
    
    # Granular control (Async)
    first_page = await client.accounts.list()
    if first_page.has_next_page():
        next_page = await first_page.get_next_page()
  5. Migrate Cloudforce One Threat Events from v4 to v5

    main

    The cloudforceone.threatevents resource has undergone significant changes in v5. Key changes include the removal of insights and crons sub-resources, the deprecation of the get() method, and updates to create(), list(), edit(), and bulk_create() method signatures.

    Summary of Changes

    Featurev4.3.1 Behaviorv5 Behavior
    InsightsSeparate insights sub-resourceConsolidated into main threat events via insight parameter
    CronsSeparate crons sub-resourceRemoved entirely
    create()attacker, attacker_country, indicator_type requiredAll now optional; supports indicators array for multiple indicators
    list()dataset_id required; offset paginationdataset_id optional; supports cursor-based pagination and stix2 format
    get()Standard methodDeprecated. Use datasets.events.get() instead
    edit()Limited parametersAdded created_at, dataset_id, insight, and raw
    bulk_create()Basic bulk creationAdded include_created_events to track results

    Required Actions

    1. Replace insights calls: Move insight data into the insight parameter of the main threat events API.
    2. Remove crons references: Use external scheduling mechanisms.
    3. Update get() calls: Migrate to the dataset-specific endpoint: client.cloudforce_one.threat_events.datasets.events.get().
    4. Update Type Hints: Change dataset_id type hints from List[str] to SequenceNotStr[str] in list() calls.
  6. Import a D1 database in three stages

    main

    The import_() method uses overloaded signatures to manage a multi-step import process. You must provide an MD5 etag of your SQL file for all stages.

    1. action="init": Initializes the import and returns an upload_url.
    2. action="ingest": Triggers the ingestion after you have uploaded the file to the provided URL. Returns a bookmark.
    3. action="poll": Uses the bookmark to check the status until it is complete.
    import hashlib
    import requests
    import time
    
    # Stage 1: Initialize
    with open("database.sql", "rb") as f:
        etag = hashlib.md5(f.read()).hexdigest()
    
    result = client.d1.database.import_(
        database_id="def456",
        account_id="abc123",
        action="init",
        etag=etag
    )
    upload_url = result.upload_url
    
    # Stage 2: Upload and Ingest
    with open("database.sql", "rb") as f:
        requests.put(upload_url, data=f)
    
    result = client.d1.database.import_(
        database_id="def456",
        account_id="abc123",
        action="ingest",
        etag=etag,
        filename="database.sql"
    )
    bookmark = result.bookmark
    
    # Stage 3: Poll for Completion
    while True:
        status = client.d1.database.import_(
            database_id="def456",
            account_id="abc123",
            action="poll",
            current_bookmark=bookmark
        )
        if status.status == "complete":
            break
        time.sleep(5)
  7. Migrate D1 Database resource from v4 to v5

    main

    The D1 Database resource (d1.database) has been completely rewritten in v5. While core functionality remains, the API surface, method signatures, and path structures have changed significantly. It is not a drop-in replacement. Key changes include:

    • Path Parameters: Most methods now require database_id as the first positional parameter.
    • Update Semantics: update() is now for full configuration replacement (PUT), while edit() is for partial updates (PATCH).
    • Pagination: list() now returns a paginated response instead of a simple list.
    • Querying: query() and raw() now support single-query and batch-query modes via overloads.
    • Import/Export: These processes are now multi-stage and require explicit polling logic.
  8. Migrate optional parameter handling from NotGiven to Omit

    main

    In v5, the SDK has transitioned from using NotGiven to using Omit for handling optional parameters in API calls. This provides better type safety and clearer semantics.

    Actions Required:

    1. Replace Imports: Change from cloudflare._types import NOT_GIVEN, NotGiven to from cloudflare._types import omit, Omit.
    2. Replace Sentinel Values: Use omit instead of NOT_GIVEN when explicitly passing a parameter you wish to exclude.
    3. Update Type Hints: Use Omit instead of NotGiven in function signatures.

    Best Practice: Instead of explicitly passing omit, simply do not pass the optional parameter at all.

    # Before (v4.3.1)
    from cloudflare._types import NOT_GIVEN, NotGiven
    result = client.queues.consumers.create(
        queue_id="abc123",
        account_id="def456",
        script_name="my-worker",
        dead_letter_queue=NOT_GIVEN
    )
    
    # After (v5)
    from cloudflare._types import omit, Omit
    result = client.queues.consumers.create(
        queue_id="abc123",
        account_id="def456",
        script_name="my-worker",
        dead_letter_queue=omit
    )
    
    # Preferred approach (v5)
    result = client.queues.consumers.create(
        queue_id="abc123",
        account_id="def456",
        script_name="my-worker"
    )
    # After (v5)
    from cloudflare._types import omit, Omit
    
    result = client.queues.consumers.create(
        queue_id="abc123",
        account_id="def456",
        script_name="my-worker",
        dead_letter_queue=omit,
        settings=omit
    )
  9. Migrate Abuse Reports from v4.3.1 to v5

    main

    The create() endpoint for the abusereports resource has changed its URL structure.

    Old endpoint: post /accounts/{account_id}/abuse-reports/{report_type} New endpoint: post /accounts/{account_id}/abuse-reports/{report_param}

    Review your code to ensure the parameter passed to the endpoint matches the new requirement.

  10. Use Omit and omit for optional parameters in v5

    main

    In v5, the SDK replaced NotGiven and NOT_GIVEN with Omit and omit to handle optional parameters. This provides better type safety when distinguishing between an omitted parameter and an explicit None (null) value.

    Actions:

    1. Update Imports: Change from cloudflare._types import NOT_GIVEN, NotGiven to from cloudflare._types import omit, Omit.
    2. Update Sentinel Values: Replace NOT_GIVEN with omit in method calls.
    3. Update Type Hints: Use Omit in function signatures.

    Best Practice: Instead of explicitly passing omit, simply omit the optional parameter from the function call entirely.

    # Preferred approach: just omit the parameter
    result = client.queues.consumers.create(
        queue_id="abc123",
        account_id="def456",
        script_name="my-worker"
    )
    
    # Explicit approach using omit
    from cloudflare._types import omit
    result = client.queues.consumers.create(
        queue_id="abc123",
        account_id="def456",
        script_name="my-worker",
        dead_letter_queue=omit
    )
  11. Use the synchronous Cloudflare client

    main

    Import Cloudflare to perform synchronous API calls. The client automatically looks for the CLOUDFLARE_API_TOKEN environment variable. You can also provide api_email as a keyword argument, though using a .env file with CLOUDFLARE_EMAIL is recommended for security.

    import os
    from cloudflare import Cloudflare
    
    client = Cloudflare(
        api_token=os.environ.get("CLOUDFLARE_API_TOKEN"),  # This is the default and can be omitted
    )
    
    zone = client.zones.create(
        account={"id": "023e105f4ecef8ad9ca31a8372d0c353"},
        name="example.com",
        type="full",
    )
    print(zone.id)