FastCF

import httpx
from collections import Counter

acctok = os.getenv('CLOUDFLARE_ACC_TOK')
usrtok = os.environ['CLOUDFLARE_USR_TOK']
acctid = os.getenv('CLOUDFLARE_ACCT_ID')
acceml = os.getenv('CLOUDFLARE_EML_ADD')

base = 'https://api.cloudflare.com/client/v4'

uhdrs = {'Authorization': f'Bearer {usrtok}'}
r = httpx.get(f'{base}/user/tokens/verify', headers=uhdrs)
r.json()['success']

True

if acctok and acctid:
    ahdrs = {'Authorization': f'Bearer {acctok}'}
    r = httpx.get(f'{base}/accounts/{acctid}/tokens/verify', headers=ahdrs)
    print(r.json()['success'])

True

glbtok = os.getenv('CLOUDFLARE_API_KEY')
if acceml and glbtok:
    ghdrs = {'X-Auth-Email': acceml, 'X-Auth-Key': glbtok, 'Content-Type': 'application/json'}
    r = httpx.get(f'{base}/user', headers=ghdrs)
    print(r.json()['success'])

True

r = httpx.get(f'{base}/zones', headers=uhdrs)
[(z['name'], z['id'][:8]) for z in r.json()['result'][:1]]

[('aaa.as', '6afde8dc')]

if acctok and acctid:
    r = httpx.get(f'{base}/zones', headers=ahdrs, params={'account.id': acctid})
    print([(z['name'], z['id'][:8]) for z in r.json()['result'][:1]])

[('aaa.as', '6afde8dc')]

CloudflareApi

Cloudflare’s operationIds are machine artifacts (zones-get vs zones-0-get — the -0- is a generator’s dedup counter), so rather than cleaning them up we name every op mechanically from its verb and path: the group is the path’s non-parameter segments, nested (zones.dns_records), and the method is the HTTP verb — except that when a group has both a collection GET and a single-item GET they become list and get. Two fixups keep the whole surface collision-free: sibling paths differing only by an interior parameter gain a _by_<param> suffix, and an action segment that would shadow an op folds into its parent as <segment>_<verb>. So GET /zones is zones.list, GET /zones/{zone_id} is zones.get, and GET /user/tokens/verify is user.tokens.verify.get — a user who knows the URL knows the call.

---

source

cf_names

def cf_names(
    spec
):

Map each op’s canonical (verb, snaked-segments) key to (group_segments, op_name), collision-free across the surface

---

source

cf_group_func

def cf_group_func(
    names
):

A fastspec group_func looking up precomputed names by canonical verb+path

The parsed surface ships with the package as cf_spec.py, fastspec’s compact form (SpecParser.save): 2.6MB instead of the 8MB raw schema, loaded in ~30ms instead of a ~0.3s parse, with the naming machinery above running only at regeneration time. build_spec refreshes it from Cloudflare’s published schema; op names can shift when the spec changes, so review that diff before shipping it.

---

source

build_spec

def build_spec(
    nm:str='fastcflare/cf_spec.py',
    url:str='https://raw.githubusercontent.com/cloudflare/api-schemas/main/openapi.json'
):

Regenerate the compact spec module cf_spec.py from Cloudflare’s OpenAPI schema.

cli = OpenAPIClient(_cf_spec(), headers={'Authorization': f'Bearer {usrtok}', 'Content-Type': 'application/json'})
sorted(cli.groups)

['accounts',
 'certificates',
 'internal',
 'ips',
 'live',
 'memberships',
 'organizations',
 'radar',
 'ready',
 'signed_url',
 'system',
 'tenants',
 'user',
 'users',
 'zones']

The naming is only trustworthy if it can never silently collide — OpGroup exposes ops with setattr, so a duplicate name would shadow its sibling (exactly the bug this design replaces). This walk asserts the built tree is clean: within every group, op names are unique (duplicates allowed only for the same endpoint spelled two ways in the spec) and no op shares a name with a subgroup.

def _surface_ok(g):
    nms = [o.name for o in g.ops]
    for n,c in Counter(nms).items():
        if c>1: assert len({tuple(snake(s) for s in o.op_spec.path.strip('/').split('/')) for o in g.ops if o.name==n})==1, f'dup op {g.name}.{n}'
    assert not set(nms)&set(g.subgroups), f'op/subgroup shadow in {g.name}: {set(nms)&set(g.subgroups)}'
    for sg in g.subgroups.values(): _surface_ok(sg)

for grp in cli.groups.values(): _surface_ok(grp)
len(cli.groups), len(cli.groups['zones'].subgroups)

(15, 60)

---

source

CloudflareApi

def CloudflareApi(
    token:NoneType=None, # API token, defaulting to `CF_API_TOKEN` when neither `token` nor `api_key` is given
    account_id:NoneType=None, # Account that owns the token, for account-owned tokens
    email:NoneType=None, # Account email for a global API key, defaulting to `CF_API_EMAIL`
    api_key:NoneType=None, # Global API key, used with `email` in place of a token
):

Cloudflare API client supporting both user and account tokens

ucf = CloudflareApi(token=usrtok)
(await ucf.verify()).result.status

'active'

if acctok and acctid:
    acf = CloudflareApi(token=acctok, account_id=acctid)
    print((await acf.verify()).result.status)

active

Listing and fetching zones are now distinct calls: zones.list filters the collection, zones.get takes a zone_id.

ucf.zones.list

List Zones

Parameters: - name (str, optional) - status (str, optional) - account_id (str, optional) - account_name (str, optional) - order (str, optional) - direction (str, optional) - page (float, default: 1) - per_page (float, default: 20) - match (str, default: ‘all’)

(await ucf.zones.list(name='answer.ai')).result[0].name

'answer.ai'

zid = (await ucf.zones.list()).result[0].id
(await ucf.zones.dns_records.list(zone_id=zid)).result[0].name

'*.aaa.as'

if acctok and acctid: print((await acf.zones.dns_records.list(zone_id=zid)).result[0].name)

*.aaa.as

---

source

CloudflareApi.create_token

async def create_token(
    doms, # Domain names the token may act on
    perm_names, # Permission-group names, such as 'Zone Read' or 'DNS Write'. Names that don't match are skipped
    name, # Name for the new token
    grp:str='account.zone.', # Resource prefix after `com.cloudflare.api.`
):

Create a user API token limited to doms. The token’s value is shown only in this response.

pgs = (await ucf.user.tokens.permission_groups.get()).result
[p.name for p in pgs if 'DNS' in p.name or 'Zone' in p.name][:20]

['Account DNS Settings Read',
 'Account DNS Settings Write',
 'Cloudflare Zero Trust Secure DNS Locations Write',
 'DNS Firewall Read',
 'DNS Firewall Write',
 'DNS Read',
 'DNS View Read',
 'DNS View Write',
 'DNS Write',
 'Zone Custom Asset Read',
 'Zone Custom Asset Write',
 'Zone DNS Settings Read',
 'Zone DNS Settings Write',
 'Zone Read',
 'Zone Security Center Insights Read',
 'Zone Security Center Insights Write',
 'Zone Settings Read',
 'Zone Settings Write',
 'Zone Transform Rules Read',
 'Zone Transform Rules Write']

With two domains, the policy must carry two distinct zone resources — under the old colliding surface the name filter was silently dropped, every domain resolved to the same zone, and the dict collapsed them to one:

tok = await ucf.create_token(['answer.ai', 'aaa.as'], ['Zone Read'], 'fastcflare-test-token')
assert len(tok.result.policies[0].resources)==2
tok.success, tok.result.name

(True, 'fastcflare-test-token')

Clean up the token we just created:

res = await ucf.user.tokens.delete(tok.result.id)
assert res.success