84 lines
3.3 KiB
Python
84 lines
3.3 KiB
Python
|
|
from python_core.config import settings
|
|
import httpx
|
|
from python_core.utils.logger import setup_logger
|
|
from typing import Dict
|
|
|
|
logger = setup_logger(__name__)
|
|
|
|
class Kv:
|
|
kv: Dict[str, str]
|
|
def __init__(self):
|
|
self.cf_account_id = settings.cloudflare_account_id
|
|
self.cf_kv_api_token = settings.cloudflare_api_key
|
|
self.cf_kv_id = settings.cloudflare_kv_id
|
|
|
|
def gets(self, keys: list[str]):
|
|
try:
|
|
with httpx.Client() as client:
|
|
response = client.post(
|
|
f"https://api.cloudflare.com/client/v4/accounts/{self.cf_account_id}/storage/kv/namespaces/{self.cf_kv_id}/bulk/get",
|
|
headers={"Authorization": f"Bearer {self.cf_kv_api_token}"},
|
|
json={
|
|
"keys": keys,
|
|
"type": "json"
|
|
}
|
|
)
|
|
response.raise_for_status()
|
|
except httpx.RequestError as e:
|
|
logger.error(f"An error occurred while put kv to cloudflare")
|
|
raise e
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(f"HTTP error occurred while get kv from cloudflare {str(e)}")
|
|
raise e
|
|
except Exception as e:
|
|
logger.error(f"An unexpected error occurred: {str(e)}")
|
|
raise e
|
|
|
|
def sets(self, caches: Dict[str, str]):
|
|
try:
|
|
with httpx.Client() as client:
|
|
response = client.put(
|
|
f"https://api.cloudflare.com/client/v4/accounts/{self.cf_account_id}/storage/kv/namespaces/{self.cf_kv_id}/bulk",
|
|
headers={"Authorization": f"Bearer {self.cf_kv_api_token}"},
|
|
json=[
|
|
{
|
|
"based64": False,
|
|
"key": key,
|
|
"value": value,
|
|
}
|
|
for (key, value) in caches.items()
|
|
]
|
|
)
|
|
response.raise_for_status()
|
|
except httpx.RequestError as e:
|
|
logger.error(f"An error occurred while put kv to cloudflare")
|
|
raise e
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(f"HTTP error occurred while get kv from cloudflare {str(e)}")
|
|
raise e
|
|
except Exception as e:
|
|
logger.error(f"An unexpected error occurred: {str(e)}")
|
|
raise e
|
|
|
|
def remove(self, keys: list[str]):
|
|
with httpx.Client() as client:
|
|
try:
|
|
response = client.post(
|
|
f"https://api.cloudflare.com/client/v4/accounts/{self.cf_account_id}/storage/kv/namespaces/{self.cf_kv_id}/bulk/delete",
|
|
headers={"Authorization": f"Bearer {self.cf_kv_api_token}"},
|
|
json=keys
|
|
)
|
|
response.raise_for_status()
|
|
except httpx.RequestError as e:
|
|
logger.error(f"An error occurred while put kv to cloudflare")
|
|
raise e
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(f"HTTP error occurred while get kv from cloudflare {str(e)}")
|
|
raise e
|
|
except Exception as e:
|
|
logger.error(f"An unexpected error occurred: {str(e)}")
|
|
raise e
|
|
|
|
|
|
kv = Kv() |