diff options
Diffstat (limited to 'arthur/apis')
| -rw-r--r-- | arthur/apis/cloudflare/zones.py | 3 | ||||
| -rw-r--r-- | arthur/apis/kubernetes/deployments.py | 6 | ||||
| -rw-r--r-- | arthur/apis/kubernetes/jobs.py | 7 |
3 files changed, 6 insertions, 10 deletions
diff --git a/arthur/apis/cloudflare/zones.py b/arthur/apis/cloudflare/zones.py index 7d407a0..1139d25 100644 --- a/arthur/apis/cloudflare/zones.py +++ b/arthur/apis/cloudflare/zones.py @@ -1,5 +1,4 @@ """APIs for managing Cloudflare zones.""" -from typing import Optional import aiohttp @@ -10,7 +9,7 @@ AUTH_HEADER = {"Authorization": f"Bearer {CONFIG.cloudflare_token}"} async def list_zones( session: aiohttp.ClientSession, - zone_name: Optional[str] = None, + zone_name: str | None = None, ) -> dict[str, str]: """List all Cloudflare zones.""" endpoint = "https://api.cloudflare.com/client/v4/zones" diff --git a/arthur/apis/kubernetes/deployments.py b/arthur/apis/kubernetes/deployments.py index f7b4d5c..971c4a0 100644 --- a/arthur/apis/kubernetes/deployments.py +++ b/arthur/apis/kubernetes/deployments.py @@ -1,5 +1,5 @@ """APIs for working with Kubernetes deployments.""" -from datetime import datetime, timezone +from datetime import UTC, datetime from kubernetes_asyncio import client from kubernetes_asyncio.client.api_client import ApiClient @@ -18,9 +18,7 @@ async def restart_deployment(deployment: str, namespace: str) -> None: "template": { "metadata": { "annotations": { - "king-arthur.pydis.com/restartedAt": datetime.now( - timezone.utc - ).isoformat() + "king-arthur.pydis.com/restartedAt": datetime.now(UTC).isoformat() } } } diff --git a/arthur/apis/kubernetes/jobs.py b/arthur/apis/kubernetes/jobs.py index a5edc53..606177c 100644 --- a/arthur/apis/kubernetes/jobs.py +++ b/arthur/apis/kubernetes/jobs.py @@ -1,19 +1,18 @@ """APIs for interacting with Kubernetes Jobs & Cronjobs.""" -from typing import Any, Optional +from typing import Any from kubernetes_asyncio import client from kubernetes_asyncio.client.api_client import ApiClient from kubernetes_asyncio.client.models import V1CronJob, V1CronJobList, V1Job -async def list_cronjobs(namespace: Optional[str] = None) -> V1CronJobList: +async def list_cronjobs(namespace: str | None = None) -> V1CronJobList: """Query the Kubernetes API for a list of cronjobss in the provided namespace.""" async with ApiClient() as api: api = client.BatchV1Api(api) if namespace: return await api.list_namespaced_cron_job(namespace) - else: - return await api.list_cron_job_for_all_namespaces() + return await api.list_cron_job_for_all_namespaces() async def get_cronjob(namespace: str, cronjob_name: str) -> V1CronJob: |