# This file is auto-generated by oagen. Do not edit.

from __future__ import annotations

from typing import TYPE_CHECKING, Any, Dict, Optional, Union

if TYPE_CHECKING:
    from .._client import AsyncWorkOSClient, WorkOSClient

from .._types import RequestOptions, enum_value, NOT_GIVEN, NotGiven
from .models import (
    CreateDataKeyResponse,
    DecryptResponse,
    ObjectMetadata,
    ObjectSummary,
    ObjectWithoutValue,
    VaultObject,
    VersionListResponse,
)
from .models import VaultOrder
from .._pagination import AsyncPage, SyncPage

# @oagen-ignore-start — client-side AES-GCM imports (hand-maintained)
import base64
import os
from dataclasses import dataclass
from typing import Tuple

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
# @oagen-ignore-end

# @oagen-ignore-start — client-side AES-GCM helpers (hand-maintained)


@dataclass(slots=True)
class DecodedKeys:
    iv: bytes
    tag: bytes
    keys: str
    ciphertext: bytes


def _aes_gcm_encrypt(
    plaintext: bytes, key: bytes, iv: bytes, aad: Optional[bytes]
) -> Dict[str, bytes]:
    encryptor = Cipher(
        algorithms.AES(key), modes.GCM(iv), backend=default_backend()
    ).encryptor()
    if aad:
        encryptor.authenticate_additional_data(aad)
    ciphertext = encryptor.update(plaintext) + encryptor.finalize()
    return {"ciphertext": ciphertext, "iv": iv, "tag": encryptor.tag}


def _aes_gcm_decrypt(
    ciphertext: bytes,
    key: bytes,
    iv: bytes,
    tag: bytes,
    aad: Optional[bytes] = None,
) -> bytes:
    decryptor = Cipher(
        algorithms.AES(key), modes.GCM(iv, tag), backend=default_backend()
    ).decryptor()
    if aad:
        decryptor.authenticate_additional_data(aad)
    return decryptor.update(ciphertext) + decryptor.finalize()


def _encode_u32_leb128(value: int) -> bytes:
    if value < 0 or value > 0xFFFFFFFF:
        raise ValueError("Value must be a 32-bit unsigned integer")
    encoded = bytearray()
    while True:
        byte = value & 0x7F
        value >>= 7
        if value != 0:
            byte |= 0x80
        encoded.append(byte)
        if value == 0:
            break
    return bytes(encoded)


def _decode_u32_leb128(buf: bytes) -> Tuple[int, int]:
    res = 0
    bit = 0
    for i, b in enumerate(buf):
        if i >= 4 and (b & 0x80) != 0:
            raise ValueError("LEB128 integer overflow (was more than 4 bytes)")
        res |= (b & 0x7F) << (7 * bit)
        if (b & 0x80) == 0:
            if res > 0xFFFFFFFF:
                raise ValueError("LEB128 integer overflow (exceeds 32 bits)")
            return res, i + 1
        bit += 1
    raise ValueError("LEB128 integer not found")


def _decode_encrypted_payload(encrypted_data_b64: str) -> DecodedKeys:
    try:
        payload = base64.b64decode(encrypted_data_b64)
    except Exception as e:
        raise ValueError("Base64 decoding failed") from e

    iv = payload[0:12]
    tag = payload[12:28]
    key_len, leb_len = _decode_u32_leb128(payload[28:])
    keys_index = 28 + leb_len
    keys_end = keys_index + key_len
    keys_slice = payload[keys_index:keys_end]
    keys = base64.b64encode(keys_slice).decode("utf-8")
    ciphertext = payload[keys_end:]
    return DecodedKeys(iv=iv, tag=tag, keys=keys, ciphertext=ciphertext)


# @oagen-ignore-end
class Vault:
    """Vault API resources."""

    def __init__(self, client: "WorkOSClient") -> None:
        self._client = client

    def create_data_key(
        self,
        *,
        context: Dict[str, str],
        request_options: Optional[RequestOptions] = None,
    ) -> CreateDataKeyResponse:
        """Create a data key

        Generate an isolated encryption key for local encryption operations.

        Args:
            context: Map of values used to determine the encryption key.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            CreateDataKeyResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "context": context,
        }
        return self._client.request(
            method="post",
            path=("vault", "v1", "keys", "data-key"),
            body=body,
            model=CreateDataKeyResponse,
            request_options=request_options,
        )

    def create_decrypt(
        self,
        *,
        keys: str,
        request_options: Optional[RequestOptions] = None,
    ) -> DecryptResponse:
        """Decrypt a data key

        Decrypt a previously encrypted data key from WorkOS Vault.

        Args:
            keys: Base64-encoded encrypted data key to decrypt.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DecryptResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "keys": keys,
        }
        return self._client.request(
            method="post",
            path=("vault", "v1", "keys", "decrypt"),
            body=body,
            model=DecryptResponse,
            request_options=request_options,
        )

    def create_rekey(
        self,
        *,
        context: Dict[str, str],
        encrypted_keys: str,
        request_options: Optional[RequestOptions] = None,
    ) -> CreateDataKeyResponse:
        """Re-encrypt a data key

        Decrypt an existing data key and re-encrypt it under a new key context.

        Args:
            context: Map of values used to determine the new encryption key.
            encrypted_keys: Base64-encoded encrypted data key blob to re-encrypt.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            CreateDataKeyResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "context": context,
            "encrypted_keys": encrypted_keys,
        }
        return self._client.request(
            method="post",
            path=("vault", "v1", "keys", "rekey"),
            body=body,
            model=CreateDataKeyResponse,
            request_options=request_options,
        )

    def list_kv(
        self,
        *,
        limit: Optional[int] = None,
        before: Optional[str] = None,
        after: Optional[str] = None,
        order: Optional[Union[VaultOrder, str]] = None,
        search: Optional[str] = None,
        updated_after: Optional[str] = None,
        request_options: Optional[RequestOptions] = None,
    ) -> SyncPage[ObjectSummary]:
        """List objects

        List all encrypted objects with cursor-based pagination.

        Args:
            limit: Upper limit on the number of objects to return. Defaults to `10`.
            before: Cursor for the previous page of results.
            after: Cursor for the next page of results.
            order: Sort direction for results.
            search: Filter results by name or structured search JSON.
            updated_after: ISO 8601 timestamp to filter by last modified time.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            SyncPage[ObjectSummary]

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params = {
            k: v
            for k, v in {
                "limit": limit,
                "before": before,
                "after": after,
                "order": enum_value(order) if order is not None else None,
                "search": search,
                "updatedAfter": updated_after,
            }.items()
            if v is not None
        }
        return self._client.request_page(
            method="get",
            path=("vault", "v1", "kv"),
            model=ObjectSummary,
            params=params,
            request_options=request_options,
        )

    def create_kv(
        self,
        *,
        key_context: Dict[str, str],
        name: str,
        value: str,
        request_options: Optional[RequestOptions] = None,
    ) -> ObjectMetadata:
        """Create an object

        Encrypt and store a new key-value object.

        Args:
            key_context: Map of values used to determine the encryption key.
            name: Unique name for the object.
            value: Plaintext data to encrypt and store.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ObjectMetadata

        Raises:
            BadRequestError: If the request is malformed (400).
            ConflictError: If a conflict occurs (409).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "key_context": key_context,
            "name": name,
            "value": value,
        }
        return self._client.request(
            method="post",
            path=("vault", "v1", "kv"),
            body=body,
            model=ObjectMetadata,
            request_options=request_options,
        )

    def get_name(
        self,
        name: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> VaultObject:
        """Read an object by name

        Fetch and decrypt an object by its unique name.

        Args:
            name: Unique name of the object.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            VaultObject

        Raises:
            BadRequestError: If the request is malformed (400).
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        return self._client.request(
            method="get",
            path=("vault", "v1", "kv", "name", str(name)),
            model=VaultObject,
            request_options=request_options,
        )

    def get_kv(
        self,
        id: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> VaultObject:
        """Read an object by ID

        Fetch and decrypt an object by its unique identifier.

        Args:
            id: Unique identifier of the object.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            VaultObject

        Raises:
            BadRequestError: If the request is malformed (400).
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        return self._client.request(
            method="get",
            path=("vault", "v1", "kv", str(id)),
            model=VaultObject,
            request_options=request_options,
        )

    def update_kv(
        self,
        id: str,
        *,
        value: str,
        version_check: Union[str, None, NotGiven] = NOT_GIVEN,
        request_options: Optional[RequestOptions] = None,
    ) -> ObjectWithoutValue:
        """Update an object

        Update the value of an existing encrypted object.

        Args:
            id: Unique identifier of the object.
            value: New plaintext value.
            version_check: ID of the expected current version for optimistic locking.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ObjectWithoutValue

        Raises:
            BadRequestError: If the request is malformed (400).
            ConflictError: If a conflict occurs (409).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "value": value,
        }
        if not isinstance(version_check, NotGiven):
            body["version_check"] = version_check
        return self._client.request(
            method="put",
            path=("vault", "v1", "kv", str(id)),
            body=body,
            model=ObjectWithoutValue,
            request_options=request_options,
        )

    def delete_kv(
        self,
        id: str,
        *,
        version_check: Optional[str] = None,
        request_options: Optional[RequestOptions] = None,
    ) -> None:
        """Delete an object

        Delete an encrypted object.

        Args:
            id: Unique identifier of the object.
            version_check: Expected current version for optimistic locking.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Raises:
            NotFoundError: If the resource is not found (404).
            ConflictError: If a conflict occurs (409).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params: Dict[str, Any] = {
            k: v
            for k, v in {
                "version_check": version_check,
            }.items()
            if v is not None
        }
        self._client.request(
            method="delete",
            path=("vault", "v1", "kv", str(id)),
            params=params,
            request_options=request_options,
        )

    def list_kv_metadata(
        self,
        id: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> ObjectWithoutValue:
        """Describe an object

        Fetch metadata for an object without decrypting it.

        Args:
            id: Unique identifier of the object.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ObjectWithoutValue

        Raises:
            BadRequestError: If the request is malformed (400).
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        return self._client.request(
            method="get",
            path=("vault", "v1", "kv", str(id), "metadata"),
            model=ObjectWithoutValue,
            request_options=request_options,
        )

    def list_kv_versions(
        self,
        id: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> VersionListResponse:
        """List object versions

        Retrieve all versions for a specific object.

        Args:
            id: Unique identifier of the object.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            VersionListResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        return self._client.request(
            method="get",
            path=("vault", "v1", "kv", str(id), "versions"),
            model=VersionListResponse,
            request_options=request_options,
        )

    # @oagen-ignore-start — client-side AES-GCM encrypt/decrypt (hand-maintained)

    def encrypt(
        self,
        *,
        data: str,
        key_context: Dict[str, str],
        associated_data: Optional[str] = None,
    ) -> str:
        """Encrypt data locally using AES-GCM with a data key derived from the context."""
        key_pair = self.create_data_key(context=key_context)

        key = base64.b64decode(key_pair.data_key)
        key_blob = base64.b64decode(key_pair.encrypted_keys)
        prefix_len_buffer = _encode_u32_leb128(len(key_blob))
        aad_buffer = associated_data.encode("utf-8") if associated_data else None
        iv = os.urandom(12)

        result = _aes_gcm_encrypt(data.encode("utf-8"), key, iv, aad_buffer)

        combined = (
            result["iv"]
            + result["tag"]
            + prefix_len_buffer
            + key_blob
            + result["ciphertext"]
        )
        return base64.b64encode(combined).decode("utf-8")

    def decrypt(
        self, *, encrypted_data: str, associated_data: Optional[str] = None
    ) -> str:
        """Decrypt data that was previously encrypted using the encrypt method."""
        decoded = _decode_encrypted_payload(encrypted_data)
        data_key = self.create_decrypt(keys=decoded.keys)

        key = base64.b64decode(data_key.data_key)
        aad_buffer = associated_data.encode("utf-8") if associated_data else None

        decrypted_bytes = _aes_gcm_decrypt(
            ciphertext=decoded.ciphertext,
            key=key,
            iv=decoded.iv,
            tag=decoded.tag,
            aad=aad_buffer,
        )
        return decrypted_bytes.decode("utf-8")

    # @oagen-ignore-end


class AsyncVault:
    """Vault API resources (async)."""

    def __init__(self, client: "AsyncWorkOSClient") -> None:
        self._client = client

    async def create_data_key(
        self,
        *,
        context: Dict[str, str],
        request_options: Optional[RequestOptions] = None,
    ) -> CreateDataKeyResponse:
        """Create a data key

        Generate an isolated encryption key for local encryption operations.

        Args:
            context: Map of values used to determine the encryption key.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            CreateDataKeyResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "context": context,
        }
        return await self._client.request(
            method="post",
            path=("vault", "v1", "keys", "data-key"),
            body=body,
            model=CreateDataKeyResponse,
            request_options=request_options,
        )

    async def create_decrypt(
        self,
        *,
        keys: str,
        request_options: Optional[RequestOptions] = None,
    ) -> DecryptResponse:
        """Decrypt a data key

        Decrypt a previously encrypted data key from WorkOS Vault.

        Args:
            keys: Base64-encoded encrypted data key to decrypt.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DecryptResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "keys": keys,
        }
        return await self._client.request(
            method="post",
            path=("vault", "v1", "keys", "decrypt"),
            body=body,
            model=DecryptResponse,
            request_options=request_options,
        )

    async def create_rekey(
        self,
        *,
        context: Dict[str, str],
        encrypted_keys: str,
        request_options: Optional[RequestOptions] = None,
    ) -> CreateDataKeyResponse:
        """Re-encrypt a data key

        Decrypt an existing data key and re-encrypt it under a new key context.

        Args:
            context: Map of values used to determine the new encryption key.
            encrypted_keys: Base64-encoded encrypted data key blob to re-encrypt.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            CreateDataKeyResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "context": context,
            "encrypted_keys": encrypted_keys,
        }
        return await self._client.request(
            method="post",
            path=("vault", "v1", "keys", "rekey"),
            body=body,
            model=CreateDataKeyResponse,
            request_options=request_options,
        )

    async def list_kv(
        self,
        *,
        limit: Optional[int] = None,
        before: Optional[str] = None,
        after: Optional[str] = None,
        order: Optional[Union[VaultOrder, str]] = None,
        search: Optional[str] = None,
        updated_after: Optional[str] = None,
        request_options: Optional[RequestOptions] = None,
    ) -> AsyncPage[ObjectSummary]:
        """List objects

        List all encrypted objects with cursor-based pagination.

        Args:
            limit: Upper limit on the number of objects to return. Defaults to `10`.
            before: Cursor for the previous page of results.
            after: Cursor for the next page of results.
            order: Sort direction for results.
            search: Filter results by name or structured search JSON.
            updated_after: ISO 8601 timestamp to filter by last modified time.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            AsyncPage[ObjectSummary]

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params = {
            k: v
            for k, v in {
                "limit": limit,
                "before": before,
                "after": after,
                "order": enum_value(order) if order is not None else None,
                "search": search,
                "updatedAfter": updated_after,
            }.items()
            if v is not None
        }
        return await self._client.request_page(
            method="get",
            path=("vault", "v1", "kv"),
            model=ObjectSummary,
            params=params,
            request_options=request_options,
        )

    async def create_kv(
        self,
        *,
        key_context: Dict[str, str],
        name: str,
        value: str,
        request_options: Optional[RequestOptions] = None,
    ) -> ObjectMetadata:
        """Create an object

        Encrypt and store a new key-value object.

        Args:
            key_context: Map of values used to determine the encryption key.
            name: Unique name for the object.
            value: Plaintext data to encrypt and store.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ObjectMetadata

        Raises:
            BadRequestError: If the request is malformed (400).
            ConflictError: If a conflict occurs (409).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "key_context": key_context,
            "name": name,
            "value": value,
        }
        return await self._client.request(
            method="post",
            path=("vault", "v1", "kv"),
            body=body,
            model=ObjectMetadata,
            request_options=request_options,
        )

    async def get_name(
        self,
        name: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> VaultObject:
        """Read an object by name

        Fetch and decrypt an object by its unique name.

        Args:
            name: Unique name of the object.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            VaultObject

        Raises:
            BadRequestError: If the request is malformed (400).
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        return await self._client.request(
            method="get",
            path=("vault", "v1", "kv", "name", str(name)),
            model=VaultObject,
            request_options=request_options,
        )

    async def get_kv(
        self,
        id: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> VaultObject:
        """Read an object by ID

        Fetch and decrypt an object by its unique identifier.

        Args:
            id: Unique identifier of the object.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            VaultObject

        Raises:
            BadRequestError: If the request is malformed (400).
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        return await self._client.request(
            method="get",
            path=("vault", "v1", "kv", str(id)),
            model=VaultObject,
            request_options=request_options,
        )

    async def update_kv(
        self,
        id: str,
        *,
        value: str,
        version_check: Union[str, None, NotGiven] = NOT_GIVEN,
        request_options: Optional[RequestOptions] = None,
    ) -> ObjectWithoutValue:
        """Update an object

        Update the value of an existing encrypted object.

        Args:
            id: Unique identifier of the object.
            value: New plaintext value.
            version_check: ID of the expected current version for optimistic locking.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ObjectWithoutValue

        Raises:
            BadRequestError: If the request is malformed (400).
            ConflictError: If a conflict occurs (409).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "value": value,
        }
        if not isinstance(version_check, NotGiven):
            body["version_check"] = version_check
        return await self._client.request(
            method="put",
            path=("vault", "v1", "kv", str(id)),
            body=body,
            model=ObjectWithoutValue,
            request_options=request_options,
        )

    async def delete_kv(
        self,
        id: str,
        *,
        version_check: Optional[str] = None,
        request_options: Optional[RequestOptions] = None,
    ) -> None:
        """Delete an object

        Delete an encrypted object.

        Args:
            id: Unique identifier of the object.
            version_check: Expected current version for optimistic locking.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Raises:
            NotFoundError: If the resource is not found (404).
            ConflictError: If a conflict occurs (409).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params: Dict[str, Any] = {
            k: v
            for k, v in {
                "version_check": version_check,
            }.items()
            if v is not None
        }
        await self._client.request(
            method="delete",
            path=("vault", "v1", "kv", str(id)),
            params=params,
            request_options=request_options,
        )

    async def list_kv_metadata(
        self,
        id: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> ObjectWithoutValue:
        """Describe an object

        Fetch metadata for an object without decrypting it.

        Args:
            id: Unique identifier of the object.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ObjectWithoutValue

        Raises:
            BadRequestError: If the request is malformed (400).
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        return await self._client.request(
            method="get",
            path=("vault", "v1", "kv", str(id), "metadata"),
            model=ObjectWithoutValue,
            request_options=request_options,
        )

    async def list_kv_versions(
        self,
        id: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> VersionListResponse:
        """List object versions

        Retrieve all versions for a specific object.

        Args:
            id: Unique identifier of the object.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            VersionListResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        return await self._client.request(
            method="get",
            path=("vault", "v1", "kv", str(id), "versions"),
            model=VersionListResponse,
            request_options=request_options,
        )

    # @oagen-ignore-start — client-side AES-GCM encrypt/decrypt (hand-maintained)

    async def encrypt(
        self,
        *,
        data: str,
        key_context: Dict[str, str],
        associated_data: Optional[str] = None,
    ) -> str:
        """Encrypt data locally using AES-GCM with a data key derived from the context."""
        key_pair = await self.create_data_key(context=key_context)

        key = base64.b64decode(key_pair.data_key)
        key_blob = base64.b64decode(key_pair.encrypted_keys)
        prefix_len_buffer = _encode_u32_leb128(len(key_blob))
        aad_buffer = associated_data.encode("utf-8") if associated_data else None
        iv = os.urandom(12)

        result = _aes_gcm_encrypt(data.encode("utf-8"), key, iv, aad_buffer)
        combined = (
            result["iv"]
            + result["tag"]
            + prefix_len_buffer
            + key_blob
            + result["ciphertext"]
        )
        return base64.b64encode(combined).decode("utf-8")

    async def decrypt(
        self, *, encrypted_data: str, associated_data: Optional[str] = None
    ) -> str:
        """Decrypt data that was previously encrypted using the encrypt method."""
        decoded = _decode_encrypted_payload(encrypted_data)
        data_key = await self.create_decrypt(keys=decoded.keys)

        key = base64.b64decode(data_key.data_key)
        aad_buffer = associated_data.encode("utf-8") if associated_data else None

        decrypted_bytes = _aes_gcm_decrypt(
            ciphertext=decoded.ciphertext,
            key=key,
            iv=decoded.iv,
            tag=decoded.tag,
            aad=aad_buffer,
        )
        return decrypted_bytes.decode("utf-8")

    # @oagen-ignore-end
