MT-23076: support api token expiration in create and reset

examples/general/api_tokens.py

@@ -20,10 +20,14 @@ def get_api_token(account_id: int, api_token_id: int) -> ApiToken:
 
 def create_api_token(account_id: int) -> ApiTokenWithToken:
     # The full token value is only returned once on the response — store it securely.
+    # Omit expires_at for the server default expiration, pass an ISO 8601
+    # date-time for an explicit expiry, or pass expires_at=None for a token
+    # that never expires.
     return api_tokens_api.create(
         account_id=account_id,
         token_params=mt.CreateApiTokenParams(
             name="My API Token",
+            expires_at="2027-06-01T00:00:00Z",
             resources=[
                 mt.ApiTokenResource(
                     resource_type="account",
@@ -37,9 +41,22 @@ def create_api_token(account_id: int) -> ApiTokenWithToken:
 
 def reset_api_token(account_id: int, api_token_id: int) -> ApiTokenWithToken:
     # The reset response includes the new full token value once — store it securely.
+    # Omit token_params for the server default expiration of the new token.
     return api_tokens_api.reset(account_id=account_id, api_token_id=api_token_id)
 
 
+def reset_api_token_with_expiration(
+    account_id: int, api_token_id: int
+) -> ApiTokenWithToken:
+    # Pass an ISO 8601 date-time for an explicit expiry of the new token,
+    # or expires_at=None for a token that never expires.
+    return api_tokens_api.reset(
+        account_id=account_id,
+        api_token_id=api_token_id,
+        token_params=mt.ResetApiTokenParams(expires_at="2027-06-01T00:00:00Z"),
+    )
+
+
 def delete_api_token(account_id: int, api_token_id: int) -> DeletedObject:
     return api_tokens_api.delete(account_id=account_id, api_token_id=api_token_id)
 
@@ -57,5 +74,8 @@ def delete_api_token(account_id: int, api_token_id: int) -> DeletedObject:
     reset = reset_api_token(ACCOUNT_ID, created.id)
     print(reset)
 
-    deleted = delete_api_token(ACCOUNT_ID, reset.id)
+    reset_with_expiration = reset_api_token_with_expiration(ACCOUNT_ID, reset.id)
+    print(reset_with_expiration)
+
+    deleted = delete_api_token(ACCOUNT_ID, reset_with_expiration.id)
     print(deleted)

mailtrap/__init__.py

@@ -8,6 +8,8 @@
 from .models.accounts import AccountAccessFilterParams
 from .models.api_tokens import ApiTokenResource
 from .models.api_tokens import CreateApiTokenParams
+from .models.api_tokens import ResetApiTokenParams
+from .models.common import UNSET
 from .models.contacts import ContactEventParams
 from .models.contacts import ContactExportFilter
 from .models.contacts import ContactListParams

mailtrap/api/resources/api_tokens.py

@@ -4,6 +4,7 @@
 from mailtrap.models.api_tokens import ApiToken
 from mailtrap.models.api_tokens import ApiTokenWithToken
 from mailtrap.models.api_tokens import CreateApiTokenParams
+from mailtrap.models.api_tokens import ResetApiTokenParams
 from mailtrap.models.common import DeletedObject
 
 
@@ -31,6 +32,11 @@ def create(
         """
         Create a new API token. The full token value is only returned once
         in the response — store it securely.
+
+        expires_at is an optional token expiration as an ISO 8601 date-time.
+        Omit it for the server default (a 1-year default is being rolled out).
+        Pass an explicit None for a token that never expires. Past or
+        more-than-5-years-ahead values are rejected with a 422 error.
         """
         response = self._client.post(
             self._api_path(account_id), json=token_params.api_data
@@ -44,13 +50,27 @@ def delete(self, account_id: int, api_token_id: int) -> DeletedObject:
         self._client.delete(self._api_path(account_id, api_token_id))
         return DeletedObject(id=api_token_id)
 
-    def reset(self, account_id: int, api_token_id: int) -> ApiTokenWithToken:
+    def reset(
+        self,
+        account_id: int,
+        api_token_id: int,
+        token_params: Optional[ResetApiTokenParams] = None,
+    ) -> ApiTokenWithToken:
         """
         Expire the requested token and create a new token with the same
         permissions. The full new token value is returned once — store it
         securely. Only tokens that have not already been reset can be reset.
+
+        expires_at is an optional expiration of the new token as an ISO 8601
+        date-time. Omit token_params or expires_at for the server default
+        (a 1-year default is being rolled out). Pass an explicit None for a
+        token that never expires. Past or more-than-5-years-ahead values are
+        rejected with a 422 error.
         """
-        response = self._client.post(f"{self._api_path(account_id, api_token_id)}/reset")
+        response = self._client.post(
+            f"{self._api_path(account_id, api_token_id)}/reset",
+            json=token_params.api_data if token_params is not None else None,
+        )
         return ApiTokenWithToken(**response)
 
     @staticmethod

mailtrap/models/api_tokens.py

@@ -1,10 +1,13 @@
+from typing import Any
 from typing import Optional
 from typing import Union
 
 from pydantic import Field
 from pydantic.dataclasses import dataclass
 
+from mailtrap.models.common import UNSET
 from mailtrap.models.common import RequestParams
+from mailtrap.models.common import UnsetType
 
 
 @dataclass
@@ -32,4 +35,28 @@ class ApiTokenWithToken(ApiToken):
 @dataclass
 class CreateApiTokenParams(RequestParams):
     name: str
+    expires_at: Union[str, None, UnsetType] = UNSET
     resources: list[ApiTokenResource] = Field(default_factory=list)
+
+    @property
+    def api_data(self) -> dict[str, Any]:
+        data = super().api_data
+        # exclude_none strips an explicit None, but here it must be sent
+        # as "expires_at": null (a token that never expires).
+        if self.expires_at is None:
+            data["expires_at"] = None
+        return data
+
+
+@dataclass
+class ResetApiTokenParams(RequestParams):
+    expires_at: Union[str, None, UnsetType] = UNSET
+
+    @property
+    def api_data(self) -> dict[str, Any]:
+        data = super().api_data
+        # exclude_none strips an explicit None, but here it must be sent
+        # as "expires_at": null (a token that never expires).
+        if self.expires_at is None:
+            data["expires_at"] = None
+        return data

mailtrap/models/common.py

@@ -1,22 +1,65 @@
 from typing import Any
+from typing import Optional
 from typing import TypeVar
 from typing import Union
 from typing import cast
 
+from pydantic import GetCoreSchemaHandler
 from pydantic import TypeAdapter
 from pydantic.dataclasses import dataclass
+from pydantic_core import core_schema
 
 T = TypeVar("T", bound="RequestParams")
 
 
+class UnsetType:
+    """
+    Sentinel type for request fields that should be omitted from the payload.
+
+    api_data drops fields whose value is UNSET, keeping an omitted field
+    distinct from an explicit None value.
+    """
+
+    _instance: Optional["UnsetType"] = None
+
+    def __new__(cls) -> "UnsetType":
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+
+    def __repr__(self) -> str:
+        return "UNSET"
+
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls, source_type: Any, handler: GetCoreSchemaHandler
+    ) -> core_schema.CoreSchema:
+        return core_schema.is_instance_schema(
+            cls,
+            serialization=core_schema.plain_serializer_function_ser_schema(
+                cls._serialize
+            ),
+        )
+
+    @staticmethod
+    def _serialize(value: "UnsetType") -> "UnsetType":
+        return value
+
+
+UNSET = UnsetType()
+
+
 @dataclass
 class RequestParams:
     @property
     def api_data(self: T) -> dict[str, Any]:
-        return cast(
+        data = cast(
             dict[str, Any],
             TypeAdapter(type(self)).dump_python(self, by_alias=True, exclude_none=True),
         )
+        return {
+            key: value for key, value in data.items() if not isinstance(value, UnsetType)
+        }
 
     @property
     def api_query_params(self: T) -> dict[str, Any]:

tests/unit/api/general/test_api_tokens.py

@@ -1,3 +1,4 @@
+import json
 from typing import Any
 
 import pytest
@@ -11,6 +12,7 @@
 from mailtrap.models.api_tokens import ApiTokenResource
 from mailtrap.models.api_tokens import ApiTokenWithToken
 from mailtrap.models.api_tokens import CreateApiTokenParams
+from mailtrap.models.api_tokens import ResetApiTokenParams
 from mailtrap.models.common import DeletedObject
 from tests import conftest
 
@@ -176,6 +178,11 @@ def test_get_by_id_should_return_api_token(
                 conftest.FORBIDDEN_RESPONSE,
                 conftest.FORBIDDEN_ERROR_MESSAGE,
             ),
+            (
+                conftest.VALIDATION_ERRORS_STATUS_CODE,
+                {"errors": {"expires_at": ["must be in the future"]}},
+                "expires_at: must be in the future",
+            ),
         ],
     )
     @responses.activate
@@ -228,6 +235,65 @@ def test_create_should_return_api_token_with_full_token_value(
             b'[{"resource_type": "account", "resource_id": 3229, "access_level": 100}]}'
         )
 
+    @responses.activate
+    def test_create_should_omit_expires_at_from_body_by_default(
+        self, client: ApiTokensApi, sample_api_token_dict: dict
+    ) -> None:
+        responses.post(
+            BASE_API_TOKENS_URL,
+            json={**sample_api_token_dict, "token": "a1b2c3d4e5f6"},
+            status=200,
+        )
+
+        client.create(ACCOUNT_ID, CreateApiTokenParams(name="My API Token"))
+
+        body = json.loads(responses.calls[0].request.body)
+        assert body == {"name": "My API Token", "resources": []}
+
+    @responses.activate
+    def test_create_should_send_null_expires_at_for_never_expiring_token(
+        self, client: ApiTokensApi, sample_api_token_dict: dict
+    ) -> None:
+        responses.post(
+            BASE_API_TOKENS_URL,
+            json={**sample_api_token_dict, "token": "a1b2c3d4e5f6"},
+            status=200,
+        )
+
+        client.create(
+            ACCOUNT_ID, CreateApiTokenParams(name="My API Token", expires_at=None)
+        )
+
+        body = json.loads(responses.calls[0].request.body)
+        assert body == {"name": "My API Token", "resources": [], "expires_at": None}
+
+    @responses.activate
+    def test_create_should_send_expires_at_value(
+        self, client: ApiTokensApi, sample_api_token_dict: dict
+    ) -> None:
+        responses.post(
+            BASE_API_TOKENS_URL,
+            json={
+                **sample_api_token_dict,
+                "expires_at": "2027-06-01T00:00:00Z",
+                "token": "a1b2c3d4e5f6",
+            },
+            status=200,
+        )
+
+        token = client.create(
+            ACCOUNT_ID,
+            CreateApiTokenParams(name="My API Token", expires_at="2027-06-01T00:00:00Z"),
+        )
+
+        body = json.loads(responses.calls[0].request.body)
+        assert body == {
+            "name": "My API Token",
+            "expires_at": "2027-06-01T00:00:00Z",
+            "resources": [],
+        }
+        assert token.expires_at == "2027-06-01T00:00:00Z"
+
     @pytest.mark.parametrize(
         "status_code,response_json,expected_error_message",
         [
@@ -292,6 +358,11 @@ def test_delete_should_return_deleted_object(self, client: ApiTokensApi) -> None
                 conftest.NOT_FOUND_RESPONSE,
                 conftest.NOT_FOUND_ERROR_MESSAGE,
             ),
+            (
+                conftest.VALIDATION_ERRORS_STATUS_CODE,
+                {"errors": {"expires_at": ["must be in the future"]}},
+                "expires_at: must be in the future",
+            ),
         ],
     )
     @responses.activate
@@ -328,3 +399,47 @@ def test_reset_should_return_api_token_with_full_token_value(
         assert isinstance(token, ApiTokenWithToken)
         assert token.id == API_TOKEN_ID
         assert token.token == "new-token-value"
+
+        assert len(responses.calls) == 1
+        assert responses.calls[0].request.body is None
+
+    @responses.activate
+    def test_reset_should_send_null_expires_at_for_never_expiring_token(
+        self, client: ApiTokensApi, sample_api_token_dict: dict
+    ) -> None:
+        responses.post(
+            f"{BASE_API_TOKENS_URL}/{API_TOKEN_ID}/reset",
+            json={**sample_api_token_dict, "token": "new-token-value"},
+            status=200,
+        )
+
+        client.reset(
+            ACCOUNT_ID, API_TOKEN_ID, token_params=ResetApiTokenParams(expires_at=None)
+        )
+
+        body = json.loads(responses.calls[0].request.body)
+        assert body == {"expires_at": None}
+
+    @responses.activate
+    def test_reset_should_send_expires_at_value(
+        self, client: ApiTokensApi, sample_api_token_dict: dict
+    ) -> None:
+        responses.post(
+            f"{BASE_API_TOKENS_URL}/{API_TOKEN_ID}/reset",
+            json={
+                **sample_api_token_dict,
+                "expires_at": "2027-06-01T00:00:00Z",
+                "token": "new-token-value",
+            },
+            status=200,
+        )
+
+        token = client.reset(
+            ACCOUNT_ID,
+            API_TOKEN_ID,
+            token_params=ResetApiTokenParams(expires_at="2027-06-01T00:00:00Z"),
+        )
+
+        body = json.loads(responses.calls[0].request.body)
+        assert body == {"expires_at": "2027-06-01T00:00:00Z"}
+        assert token.expires_at == "2027-06-01T00:00:00Z"