kv_wrapper.py

import httpx
from dataclasses import dataclass
import json
from typing import Union

error_codes: dict = {
    "10001": "service temporarily unavailable",
    "10002": "missing CF-Ray header",
    "10003": "missing account public ID",
    "10004": "missing account tag",
    "10005": "URL parameter account tag does not match JWT account tag",
    "10006": "malformed account tag",
    "10007": "malformed page argument",
    "10008": "malformed per_page argument",
    "10009": "key not found",
    "10010": "malformed namespace",
    "10011": "malformed namespace ID",
    "10012": "malformed value",
    "10013": "namespace not found",
    "10014": "namespace already exists",
    "10015": "missing account internal ID",
    "10016": "malformed account internal ID",
    "10018": "too many namespaces in this account",
    "10019": "missing title",
    "10021": "this namespace does not support the list-keys endpoint",
    "10022": "too many requests",
    "10024": "payload too large",
    "10025": "endpoint does not exist",
    "10026": "not entitled",
    "10028": "invalid limit argument",
    "10029": "invalid request",
    "10030": "key too long",
    "10033": "invalid expiration",
    "10034": "invalid expiration ttl",
    "10035": "this namespace does not support the bulk endpoint",
    "10037": "the user lacks the permissions to perform this operation",
    "10038": "this namespace does not support the list-keys prefix parameter",
    "10041": "invalid \"list keys\" cursor",
    "10042": "illegal key name",
    "10043": "invalid order",
    "10044": "invalid direction",
    "10045": "deprecated endpoint",
    "10046": "too many bulk requests",
    "10047": "invalid metadata",
    "10048": "free limit reached"
}

@dataclass
class Account:
    mail: str
    auth_key: str
    account_id: str

@dataclass
class CfData:
    account: Account
    headers: dict
    namespace_id: Union[str, None]
    api: str

    def __init__(self, namespace_id: str=None, *, account: Account):
        self.account.mail = account.mail
        self.account.auth_key = account.auth_key
        self.headers = {
            'X-Auth-Email': self.mail,
            'X-Auth-Key'  : self.auth_key
        }
        self.namespace_id = namespace_id
        self.account.account_id = account.account_id
        self.api = f"https://api.cloudflare.com/client/v4/accounts/{self.account.account_id}/storage/kv/namespaces/{self.namespace_id}"

class KVNamespaceError(Exception):
    pass

class Namespace:
    account: Account
    headers: dict
    namespace_id: Union[str, None]
    api: str

    def __init__(self, namespace_id: str=None, *, account: Account):
        self.account.mail = account.mail
        self.account.auth_key = account.auth_key
        self.headers = {
            'X-Auth-Email': self.mail,
            'X-Auth-Key'  : self.auth_key,
            'Content-Type': 'application/json'
        }
        self.namespace_id = namespace_id
        self.account.account_id = account.account_id
        self.api = f"https://api.cloudflare.com/client/v4/accounts/{self.account_id}"

    async def create_namespace(self, title: str) -> CfData:
        data = json.dumps({
            'title': title 
        }) # disable json.dumps() to force a 10010 error
        resp = httpx.post(f"{self.api}", headers=self.headers, data=data)
        if resp.status_code == 400:
            resp_json = resp.json()
            raise KVNamespaceError(error_codes[int(resp_json["errors"]["code"])])

        return CfData(self.account.mail, self.account.auth_key, str(resp_json["result"]["id"]), self.account.account_id)

    async def get_namespace(self, namespace_id: str=None, title: str=None) -> CfData:
        # if id given, then send the object of that, else with title, call list_namespaces
        if namespace_id is None and title:
            resp = httpx.get(f"{self.api}", headers=self.headers)
            resp_json = resp.json()
            namespace_id, *_ = [i['id'] for i in resp_json["result"] if i['title'] == title]
            return CfData(self.account.mail, self.account.auth_key, str(namespace_id), self.account.account_id)

        else:
            return CfData(self.account.mail, self.account.auth_key, str(namespace_id), self.account.account_id)

    async def list_namespaces(self) -> list:
        resp = httpx.get(f"{self.api}", headers=self.headers)
        resp_json = resp.json()
        for namespace in resp_json["result"]:
            print(namespace["title"] + ":" + namespace["id"])

    def delete_namespace(self) -> None:
        httpx.delete(f"{self.api}", headers=self.headers)



async def getter(nuke: str, cfobj: CfData):
    resp = httpx.get(f"{cfobj.api}values/{nuke}", headers=cfobj.headers)
    print(resp.text)

async def fetter(nuke: str, data: str, cfobj: CfData):    
    resp = httpx.put(f"{cfobj.api}/values/{nuke}", headers=cfobj.headers, data=data)
    print(resp.text)

def main():
    accobj = Account("xyz@gmail.com", "ksudvcehbcj", "ejfcnkenclemc")
    cfobj = CfData("kcjekjcnecmlk", accobj)
    getter("", cfobj)
    fetter("", "", cfobj)

if __name__ == '__main__':
    main()