feat: Add purge subcommand

.vscode/settings.json

@@ -27,5 +27,4 @@
     "python.testing.unittestEnabled": false,
     "python.testing.pytestEnabled": true,
     "python-envs.defaultEnvManager": "ms-python.python:system",
-    "python-envs.pythonProjects": [],
 }

akcli/api.py

@@ -9,7 +9,7 @@
 from functools import wraps
 from pathlib import Path
 from time import sleep
-from typing import Any, Optional
+from typing import Any, List, Optional
 
 import requests
 from akamai.edgegrid import EdgeGridAuth, EdgeRc
@@ -29,7 +29,7 @@
     ResourceNotFound,
     TooManyRequests,
 )
-from .models import DigResponse, TranslateResponse
+from .models import DigResponse, PurgeResponse, TranslateResponse
 from .typing import Certificate, GenericFunction, Headers, JSONResponse
 from .utils import highlight
 
@@ -193,29 +193,33 @@ def _request(self, method: str, endpoint: str, *args: Any, **kwargs: Any) -> JSO
         except Exception as e:
             raise RequestError(f"An error occurred while making the request: {e}") from e
 
-    def _get(self, endpoint: str, headers: Optional[Headers] = None) -> JSONResponse:
+    def _get(self, endpoint: str, headers: Optional[Headers] = None, use_cache: bool = True) -> JSONResponse:
         """
         Make a GET request.
         """
-        return self._request(method="GET", endpoint=endpoint, headers=headers)
+        return self._request(method="GET", endpoint=endpoint, headers=headers, use_cache=use_cache)
 
-    def _post(self, endpoint: str, json: dict, headers: Optional[Headers] = None) -> JSONResponse:
+    def _post(
+        self, endpoint: str, json: dict, headers: Optional[Headers] = None, use_cache: bool = True
+    ) -> JSONResponse:
         """
         Make a POST request.
         """
-        return self._request(method="POST", endpoint=endpoint, json=json, headers=headers)
+        return self._request(method="POST", endpoint=endpoint, json=json, headers=headers, use_cache=use_cache)
 
-    def _patch(self, endpoint: str, json: dict, headers: Optional[Headers] = None) -> JSONResponse:
+    def _patch(
+        self, endpoint: str, json: dict, headers: Optional[Headers] = None, use_cache: bool = True
+    ) -> JSONResponse:
         """
         Make a PATCH request.
         """
-        return self._request(method="PATCH", endpoint=endpoint, json=json, headers=headers)
+        return self._request(method="PATCH", endpoint=endpoint, json=json, headers=headers, use_cache=use_cache)
 
-    def _delete(self, endpoint: str, headers: Optional[Headers] = None) -> JSONResponse:
+    def _delete(self, endpoint: str, headers: Optional[Headers] = None, use_cache: bool = True) -> JSONResponse:
         """
         Make a DELETE request.
         """
-        return self._request(method="DELETE", endpoint=endpoint, headers=headers)
+        return self._request(method="DELETE", endpoint=endpoint, headers=headers, use_cache=use_cache)
 
     def dig(self, hostname: str, query_type: str) -> DigResponse:
         """
@@ -244,3 +248,16 @@ def translate(self, id: str, trace: bool) -> TranslateResponse:
         data = self._post(endpoint=endpoint, json=payload)
 
         return TranslateResponse.parse_model(data)
+
+    def purge(self, method: str, network: str, purge_type: str, objects: List) -> PurgeResponse:
+        """
+        Purge content given CP code, tag or URL/ARL.
+        Reference: https://techdocs.akamai.com/purge-cache/reference/
+        """
+        endpoint = f"/ccu/v3/{method}/{purge_type}/{network}"
+        payload = {"objects": objects}
+
+        # Purge is a write operation — caching would silently skip the actual request
+        data = self._post(endpoint=endpoint, json=payload, use_cache=False)
+
+        return PurgeResponse.parse_model(data)

akcli/cache.py

@@ -157,9 +157,11 @@ def wrapper(self, method: str, endpoint: str, *args: Any, **kwargs: Any) -> Seri
         payload = kwargs.get("json")
         cache: Cache = self._cache
         key = cache.generate_key(method, endpoint, payload)
-
         cached = cache.get(key)
-        if cached is not None and cache.use_cache:
+
+        # Handle cases where wrapped function is passing use_cache
+        use_cache = kwargs.pop("use_cache", True) and cache.use_cache
+        if cached is not None and use_cache:
             return cached.data
 
         # If not cached, make request to API and store result

akcli/commands/__init__.py

@@ -7,9 +7,11 @@
 import typer
 
 from .dig import app as dig_app
+from .purge import app as purge_app
 from .translate import app as translate_app
 
 app = typer.Typer()
 
 app.add_typer(dig_app)
 app.add_typer(translate_app)
+app.add_typer(purge_app)

akcli/commands/_common.py

@@ -4,8 +4,6 @@
 Common command arguments and options.
 """
 
-from typing import Optional
-
 import typer
 from merge_args import merge_args
 from typing_extensions import Annotated
@@ -24,7 +22,7 @@ def common_args(
     @merge_args(func)
     def wrapper(
         ctx: typer.Context,
-        json: Annotated[Optional[bool], typer.Option("--json", help="Output in JSON format.")] = None,
+        json: Annotated[bool, typer.Option(help="Print output in JSON format.")] = False,
         **kwargs,
     ) -> None:
         return func(ctx=ctx, **kwargs)

akcli/commands/dig.py

@@ -10,11 +10,12 @@
 import typer
 from typing_extensions import Annotated
 
-from akcli.commands._common import common_args
 from akcli.config import Config
 from akcli.exceptions import DigNoAnswerWarning, MutuallyExclusiveArgs, handle_exceptions
 from akcli.utils import create_table, highlight, print_json
 
+from ._common import common_args
+
 _COMMAND_NAME = "dig"
 
 _COLUMNS_HEADERS = [

akcli/commands/purge.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+
+"""
+Tool to purge an object or list of objects on the Akamai Network.
+"""
+
+from enum import Enum
+from pathlib import Path
+from typing import List, Optional
+
+import typer
+from typing_extensions import Annotated
+
+from akcli.config import Config
+from akcli.exceptions import InvalidParams, MutuallyExclusiveArgs, handle_exceptions
+from akcli.utils import print_json, print_success
+
+from ._common import common_args
+
+_COMMAND_NAME = "purge"
+
+app = typer.Typer()
+config = Config().purge
+
+
+class _PurgeMethod(str, Enum):
+    """
+    Supported purge methods.
+    """
+
+    invalidate = "invalidate"
+    delete = "delete"
+
+
+class _PurgeNetwork(str, Enum):
+    """
+    Supported networks.
+    """
+
+    staging = "staging"
+    production = "production"
+
+
+class _PurgeType(str, Enum):
+    """
+    Supported purge types.
+    """
+
+    tag = "tag"
+    url = "url"
+    cpcode = "cpcode"
+
+
+def _validate_urls(urls: List[str]) -> None:
+    """Ensure that URLs start with http:// or https://"""
+    for u in urls:
+        if not u.startswith(("http://", "https://")):
+            raise InvalidParams(
+                f"Invalid URL [b]'{u}[/b]'. URLs/ARLs must include the protocol scheme (http:// or https://)."
+            )
+
+
+def _file_to_list(f: Path) -> List[str]:
+    """Parse the non-empty lines of a file into a list, stripping whitespace."""
+    return [obj.strip() for obj in f.read_text().splitlines() if obj.strip()]
+
+
+@app.command(name=_COMMAND_NAME)
+@handle_exceptions()
+@common_args
+def purge(
+    ctx: typer.Context,
+    objects: Annotated[
+        Optional[List[str]], typer.Argument(help="List with the objects to purge (CP codes, tags or URLs/ARLs).")
+    ] = None,
+    from_file: Annotated[
+        Optional[Path], typer.Option(help="Path to a file with one object per line.", exists=True, file_okay=True)
+    ] = None,
+    method: Annotated[_PurgeMethod, typer.Option(help="Method used to purge.")] = config.method,  # type: ignore
+    network: Annotated[_PurgeNetwork, typer.Option(help="Akamai network.")] = config.network,  # type: ignore
+    purge_type: Annotated[_PurgeType, typer.Option(help="The type of objects to purge.")] = config.type,  # type: ignore
+) -> None:
+    """
+    Tool to purge an object or list of objects on the Akamai Network.
+    """
+    api = ctx.obj.api
+    console = ctx.obj.console
+
+    # Validate args
+    if objects and from_file:
+        raise MutuallyExclusiveArgs("'--from-file' and positional arguments are mutually exclusive.")
+
+    if not objects and not from_file:
+        raise InvalidParams("Provide either objects as positional arguments or use '--from-file'.")
+
+    if from_file is not None:
+        objects = _file_to_list(from_file)
+
+    # Raise error if URLs/ARLs don't start with 'http://' or 'https://'
+    if purge_type == _PurgeType.url:
+        _validate_urls(objects)  # type: ignore
+
+    response = api.purge(method, network, purge_type, objects)
+
+    if ctx.params.get("json"):
+        print_json(console, response.model_dump(by_alias=True))
+        raise typer.Exit()
+
+    msg = (
+        f"Your request will be completed in [b]{response.estimated_seconds}[/b] seconds.\n"
+        f"Purge ID: [u]{response.purge_id}[/u]"
+    )
+
+    print_success(console, msg)

akcli/commands/translate.py

@@ -11,7 +11,6 @@
 from rich.table import Table
 from typing_extensions import Annotated
 
-from akcli.commands._common import common_args
 from akcli.config import Config
 from akcli.exceptions import TranslateNoLogsWarning, handle_exceptions
 from akcli.typing import TableParams
@@ -22,6 +21,8 @@
     snakecase_to_title,
 )
 
+from ._common import common_args
+
 _COMMAND_NAME = "translate"
 
 _COLUMNS_HEADERS = [

akcli/config.py

@@ -45,6 +45,10 @@
 
 _DEFAULT_TRANSLATE_TRACE = False
 
+_DEFAULT_PURGE_METHOD = "invalidate"
+_DEFAULT_PURGE_NETWORK = "staging"
+_DEFAULT_PURGE_TYPE = "url"
+
 MIN_REQUEST_TIMEOUT = 0
 MAX_REQUEST_TIMEOUT = 120
 
@@ -93,6 +97,9 @@ class _MainOptions(_OptionsBase):
     proxy: Optional[str] = _DEFAULT_PROXY
 
 
+""
+
+
 @dataclass
 class _DigOptions(_OptionsBase):
     """
@@ -113,6 +120,17 @@ class _TranslateOptions(_OptionsBase):
     trace: bool = _DEFAULT_TRANSLATE_TRACE
 
 
+@dataclass
+class _PurgeOptions(_OptionsBase):
+    """
+    Dataclass that contains all the options for the `akcli purge` command.
+    """
+
+    method: str = _DEFAULT_PURGE_METHOD
+    network: str = _DEFAULT_PURGE_NETWORK
+    type: str = _DEFAULT_PURGE_TYPE
+
+
 class Config:
     """
     Handle configuration file loading and initialization and exposing all the config options.
@@ -134,6 +152,7 @@ class Config:
     main: _MainOptions
     dig: _DigOptions
     translate: _TranslateOptions
+    purge: _PurgeOptions
 
     def __new__(cls, *args: Any, **kwargs: Any) -> "Config":
         """

akcli/exceptions.py

@@ -158,17 +158,24 @@ class InvalidEdgeRcSection(HandledException):
 class InvalidPanelType(HandledException):
     """Raised when an invalid panel type is used."""
 
-    exit_code = 61
+    exit_code = 71
     default_msg = "Invalid panel type specified."
 
 
 class MutuallyExclusiveArgs(HandledException):
     """Raised if mutually exclusive arguments are used together."""
 
-    exit_code = 62
+    exit_code = 72
     default_msg = "Mutually exclusive arguments."
 
 
+class InvalidParams(HandledException):
+    """Raised when invalid parameters are passed to a command."""
+
+    exit_code = 73
+    default_msg = "Invalid parameters."
+
+
 # ----------------------
 # Warnings
 # ----------------------

akcli/models/__init__.py

@@ -5,9 +5,7 @@
 """
 
 from .dig_response import DigResponse
+from .purge_response import PurgeResponse
 from .translate_response import TranslateResponse
 
-__all__ = [
-    "DigResponse",
-    "TranslateResponse",
-]
+__all__ = ["DigResponse", "PurgeResponse", "TranslateResponse"]

akcli/models/purge_response.py

@@ -0,0 +1,20 @@
+# /usr/bin/env python3
+
+"""
+Generated with `datamodel-code-generator` (https://github.com/koxudaxi/datamodel-code-generator)
+and refined with manual adjustments.
+"""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from ._base import BaseAPIModel
+
+
+class PurgeResponse(BaseAPIModel):
+    detail: Optional[str] = None
+    estimated_seconds: Optional[int] = None
+    http_status: Optional[int] = None
+    purge_id: Optional[str] = None
+    support_id: Optional[str] = None

akcli/typing.py

@@ -35,7 +35,7 @@
 SerializedConfig = Dict[str, SerializedOptions]
 """Type alias representing the entire configuration file with all the options in serialized format."""
 
-PanelType = Literal["info", "result", "warning", "error"]
+PanelType = Literal["info", "error", "warning", "result", "success"]
 """String literals representing valid panel types for rich panels."""
 
 ColumnHeaders = List[_BaseDict]