diff --git a/examples/domain_forge.py b/examples/domain_forge.py new file mode 100644 index 0000000..a2441ea --- /dev/null +++ b/examples/domain_forge.py @@ -0,0 +1,36 @@ +import os + +import resend + +if not os.environ["RESEND_API_KEY"]: + raise EnvironmentError("RESEND_API_KEY is missing") + +domain_id = "d91cd9bd-1176-453e-8fc1-35364d380206" + +# Enable: activate Forge on an existing domain. Returns the full domain. +# return_path is optional: when omitted, Resend picks a free DNS label. +# Enabling an already-migrated domain is idempotent and returns its +# current state. +enable_params: resend.Domains.Forge.EnableParams = { + "id": domain_id, + "return_path": "send", +} +domain: resend.Domain = resend.Domains.Forge.enable(enable_params) +print(f"Forge enabled on domain: {domain['id']}") +print(f"Domain status: {domain['status']}") + +# Move the Forge return path with the classic domain update. This does not +# start verification: fetch the new record with Domains.get, add it to your +# DNS, then verify with scope=["forge"]. +update_params: resend.Domains.UpdateParams = { + "id": domain_id, + "forge_return_path": "outbound", +} +updated = resend.Domains.update(update_params) +print(f"Forge return path moved on domain: {updated['id']}") + +# Verify only the Forge record with the classic domain verify, scoped to +# Forge. Verification runs asynchronously; poll Domains.get for the state. +verify_params: resend.Domains.VerifyParams = {"scope": ["forge"]} +verified = resend.Domains.verify(domain_id=domain_id, params=verify_params) +print(f"Forge verification triggered for domain: {verified['id']}") diff --git a/resend/__init__.py b/resend/__init__.py index 6e5aa88..9653ee8 100644 --- a/resend/__init__.py +++ b/resend/__init__.py @@ -36,6 +36,7 @@ from .domains._domains import Domains from .domains.claims._domain_claim import DomainClaim, DomainClaimRecord from .domains.claims._domain_claims import DomainClaims +from .domains.forge._domain_forge import DomainForge from .emails._attachment import Attachment, RemoteAttachment from .emails._attachments import Attachments as EmailAttachments from .emails._batch import Batch, BatchValidationError @@ -167,6 +168,7 @@ "ApiKeys", "Domains", "DomainClaims", + "DomainForge", "Batch", "Audiences", "Automations", diff --git a/resend/domains/_domains.py b/resend/domains/_domains.py index aa5f2ca..b3c34b0 100644 --- a/resend/domains/_domains.py +++ b/resend/domains/_domains.py @@ -7,6 +7,7 @@ from resend.domains._domain import Domain from resend.domains._record import Record from resend.domains.claims._domain_claims import DomainClaims +from resend.domains.forge._domain_forge import DomainForge from resend.pagination_helper import PaginationHelper # Async imports (optional - only available with pip install resend[async]) @@ -17,12 +18,17 @@ TlsOptions = Literal["enforced", "opportunistic"] +VerifyDomainScope = Literal["forge"] + class Domains: class Claims(DomainClaims): pass + class Forge(DomainForge): + pass + class ListParams(TypedDict): limit: NotRequired[int] """ @@ -145,6 +151,21 @@ class UpdateParams(TypedDict): """ The custom subdomain used for click and open tracking links (e.g., "links"). """ + forge_return_path: NotRequired[str] + """ + Move the domain's Forge return path to another host. Only possible + while the Forge record is still verifying. Moving the path does not + start verification: fetch the new record with `Domains.get`, add it to + your DNS, then call `Domains.verify` with `scope=["forge"]`. + """ + + class VerifyParams(TypedDict): + scope: NotRequired[List[VerifyDomainScope]] + """ + Limit the check to specific records. Pass `["forge"]` to verify only + the Forge record and leave the other records untouched. Requires Forge + to be enabled on the domain. When omitted, all records are checked. + """ class CreateParams(TypedDict): name: str @@ -272,20 +293,24 @@ def remove(cls, domain_id: str) -> Domain: return resp @classmethod - def verify(cls, domain_id: str) -> Domain: + def verify(cls, domain_id: str, params: Optional[VerifyParams] = None) -> Domain: """ Verify an existing domain. see more: https://resend.com/docs/api-reference/domains/verify-domain Args: domain_id (str): The domain ID + params (Optional[VerifyParams]): Optional verification parameters + - scope: Pass ["forge"] to re-check only the Forge record Returns: Domain: The verified domain object """ path = f"/domains/{domain_id}/verify" resp = request.Request[Domain]( - path=path, params={}, verb="post" + path=path, + params=cast(Dict[Any, Any], params) if params else {}, + verb="post", ).perform_with_content() return resp @@ -382,19 +407,25 @@ async def remove_async(cls, domain_id: str) -> Domain: return resp @classmethod - async def verify_async(cls, domain_id: str) -> Domain: + async def verify_async( + cls, domain_id: str, params: Optional[VerifyParams] = None + ) -> Domain: """ Verify an existing domain (async). see more: https://resend.com/docs/api-reference/domains/verify-domain Args: domain_id (str): The domain ID + params (Optional[VerifyParams]): Optional verification parameters + - scope: Pass ["forge"] to re-check only the Forge record Returns: Domain: The verified domain object """ path = f"/domains/{domain_id}/verify" resp = await AsyncRequest[Domain]( - path=path, params={}, verb="post" + path=path, + params=cast(Dict[Any, Any], params) if params else {}, + verb="post", ).perform_with_content() return resp diff --git a/resend/domains/forge/__init__.py b/resend/domains/forge/__init__.py new file mode 100644 index 0000000..20672a3 --- /dev/null +++ b/resend/domains/forge/__init__.py @@ -0,0 +1,3 @@ +from ._domain_forge import DomainForge + +__all__ = ["DomainForge"] diff --git a/resend/domains/forge/_domain_forge.py b/resend/domains/forge/_domain_forge.py new file mode 100644 index 0000000..33bfe6f --- /dev/null +++ b/resend/domains/forge/_domain_forge.py @@ -0,0 +1,74 @@ +from typing import Any, Dict, Mapping + +from typing_extensions import NotRequired, TypedDict + +from resend import request +from resend.domains._domain import Domain + +# Async imports (optional - only available with pip install resend[async]) +try: + from resend.async_request import AsyncRequest +except ImportError: + pass + + +class DomainForge: + class EnableParams(TypedDict): + id: str + """ + The domain ID. + """ + return_path: NotRequired[str] + """ + The subdomain for the Return-Path address. When omitted, Resend picks + a free DNS label for the domain. + """ + + @classmethod + def enable(cls, params: EnableParams) -> Domain: + """ + Activate Forge on an existing domain. Idempotent on the API side: + enabling an already-migrated domain returns its current state. + + To move the Forge return path afterwards, use `Domains.update` with + `forge_return_path`. To re-check only the Forge record, use + `Domains.verify` with `scope=["forge"]`. + + Args: + params (EnableParams): The Forge enable parameters + + Returns: + Domain: The full domain object + """ + path = f"/domains/{params['id']}/forge" + resp = request.Request[Domain]( + path=path, params=cls._body(params), verb="post" + ).perform_with_content() + return resp + + @classmethod + async def enable_async(cls, params: EnableParams) -> Domain: + """ + Activate Forge on an existing domain (async). Idempotent on the API + side: enabling an already-migrated domain returns its current state. + + To move the Forge return path afterwards, use `Domains.update_async` + with `forge_return_path`. To re-check only the Forge record, use + `Domains.verify_async` with `scope=["forge"]`. + + Args: + params (EnableParams): The Forge enable parameters + + Returns: + Domain: The full domain object + """ + path = f"/domains/{params['id']}/forge" + resp = await AsyncRequest[Domain]( + path=path, params=cls._body(params), verb="post" + ).perform_with_content() + return resp + + @staticmethod + def _body(params: Mapping[str, Any]) -> Dict[str, Any]: + """The id addresses the path, not the request body.""" + return {k: v for k, v in params.items() if k != "id"} diff --git a/resend/version.py b/resend/version.py index 2c8a584..5cd4c2a 100644 --- a/resend/version.py +++ b/resend/version.py @@ -1,4 +1,4 @@ -__version__ = "2.53.2" +__version__ = "2.54.0" def get_version() -> str: diff --git a/tests/domain_forge_async_test.py b/tests/domain_forge_async_test.py new file mode 100644 index 0000000..845b29d --- /dev/null +++ b/tests/domain_forge_async_test.py @@ -0,0 +1,43 @@ +import pytest + +import resend +from resend.exceptions import NoContentError +from tests.conftest import AsyncResendBaseTest +from tests.domain_forge_test import forge_domain_json + +# flake8: noqa + +pytestmark = pytest.mark.asyncio + + +class TestDomainForgeAsync(AsyncResendBaseTest): + async def test_domain_forge_enable_async(self) -> None: + self.set_mock_json(forge_domain_json()) + + params: resend.Domains.Forge.EnableParams = { + "id": "d91cd9bd-1176-453e-8fc1-35364d380206", + "return_path": "send", + } + domain = await resend.Domains.Forge.enable_async(params) + assert domain["id"] == "d91cd9bd-1176-453e-8fc1-35364d380206" + assert domain["name"] == "example.com" + + async def test_domain_forge_enable_async_without_return_path(self) -> None: + self.set_mock_json(forge_domain_json()) + + params: resend.Domains.Forge.EnableParams = { + "id": "d91cd9bd-1176-453e-8fc1-35364d380206", + } + domain = await resend.Domains.Forge.enable_async(params) + assert domain["id"] == "d91cd9bd-1176-453e-8fc1-35364d380206" + assert domain["status"] == "pending" + + async def test_should_enable_forge_async_raise_exception_when_no_content( + self, + ) -> None: + self.set_mock_json(None) + params: resend.Domains.Forge.EnableParams = { + "id": "d91cd9bd-1176-453e-8fc1-35364d380206", + } + with pytest.raises(NoContentError): + _ = await resend.Domains.Forge.enable_async(params) diff --git a/tests/domain_forge_test.py b/tests/domain_forge_test.py new file mode 100644 index 0000000..a8b68b2 --- /dev/null +++ b/tests/domain_forge_test.py @@ -0,0 +1,114 @@ +from typing import Any, Dict +from unittest import TestCase +from unittest.mock import create_autospec + +import resend +from resend.exceptions import NoContentError +from resend.http_client import HTTPClient +from tests.conftest import ResendBaseTest + +# flake8: noqa + + +def forge_domain_json() -> Dict[str, Any]: + return { + "object": "domain", + "id": "d91cd9bd-1176-453e-8fc1-35364d380206", + "name": "example.com", + "status": "pending", + "created_at": "2026-11-11 10:00:00.000000+00", + "region": "us-east-1", + "records": [ + { + "record": "SPF", + "name": "send", + "value": "v=spf1 include:spf.resend.com ~all", + "type": "TXT", + "ttl": "Auto", + "status": "pending", + }, + { + "record": "DKIM", + "name": "resend._domainkey", + "value": "resend._domainkey.example.com.dkim.resend.com.", + "type": "CNAME", + "ttl": "Auto", + "status": "pending", + }, + ], + "open_tracking": False, + "click_tracking": False, + } + + +class TestDomainForge(ResendBaseTest): + def test_domain_forge_enable(self) -> None: + self.set_mock_json(forge_domain_json()) + + params: resend.Domains.Forge.EnableParams = { + "id": "d91cd9bd-1176-453e-8fc1-35364d380206", + "return_path": "send", + } + domain: resend.Domain = resend.Domains.Forge.enable(params) + assert domain["id"] == "d91cd9bd-1176-453e-8fc1-35364d380206" + assert domain["name"] == "example.com" + assert domain["status"] == "pending" + assert ( + self.mock.call_args.kwargs["url"] + == "https://api.resend.com/domains/d91cd9bd-1176-453e-8fc1-35364d380206/forge" + ) + + def test_domain_forge_enable_without_return_path(self) -> None: + self.set_mock_json(forge_domain_json()) + + params: resend.Domains.Forge.EnableParams = { + "id": "d91cd9bd-1176-453e-8fc1-35364d380206", + } + domain = resend.Domains.Forge.enable(params) + assert domain["id"] == "d91cd9bd-1176-453e-8fc1-35364d380206" + assert domain["status"] == "pending" + + def test_should_enable_forge_raise_exception_when_no_content(self) -> None: + self.set_mock_json(None) + params: resend.Domains.Forge.EnableParams = { + "id": "d91cd9bd-1176-453e-8fc1-35364d380206", + } + with self.assertRaises(NoContentError): + _ = resend.Domains.Forge.enable(params) + + +class TestDomainForgeRequestBody(TestCase): + def setUp(self) -> None: + resend.api_key = "re_123" + self.mock_client = create_autospec(HTTPClient, instance=True) + self.mock_client.request.return_value = ( + b'{"id": "d91cd9bd-1176-453e-8fc1-35364d380206"}', + 200, + {"Content-Type": "application/json"}, + ) + self.previous_http_client = resend.default_http_client + resend.default_http_client = self.mock_client + + def tearDown(self) -> None: + resend.default_http_client = self.previous_http_client + + def test_enable_sends_snake_case_return_path_and_no_id(self) -> None: + params: resend.Domains.Forge.EnableParams = { + "id": "d91cd9bd-1176-453e-8fc1-35364d380206", + "return_path": "send", + } + resend.Domains.Forge.enable(params) + + _, kwargs = self.mock_client.request.call_args + assert kwargs["method"] == "post" + assert kwargs["json"] == {"return_path": "send"} + + def test_enable_sends_empty_body_without_return_path(self) -> None: + params: resend.Domains.Forge.EnableParams = { + "id": "d91cd9bd-1176-453e-8fc1-35364d380206", + } + resend.Domains.Forge.enable(params) + + _, kwargs = self.mock_client.request.call_args + assert kwargs["method"] == "post" + assert kwargs["json"] == {} diff --git a/tests/domains_async_test.py b/tests/domains_async_test.py index c9f900b..1949cbc 100644 --- a/tests/domains_async_test.py +++ b/tests/domains_async_test.py @@ -370,3 +370,15 @@ async def test_should_verify_domains_async_raise_exception_when_no_content( _ = await resend.Domains.verify_async( domain_id="d91cd9bd-1176-453e-8fc1-35364d380206" ) + + async def test_domains_verify_async_with_forge_scope(self) -> None: + self.set_mock_json( + {"object": "domain", "id": "d91cd9bd-1176-453e-8fc1-35364d380206"} + ) + + params: resend.Domains.VerifyParams = {"scope": ["forge"]} + domain = await resend.Domains.verify_async( + domain_id="d91cd9bd-1176-453e-8fc1-35364d380206", + params=params, + ) + assert domain["id"] == "d91cd9bd-1176-453e-8fc1-35364d380206" diff --git a/tests/domains_test.py b/tests/domains_test.py index 53ae913..f8d0ae8 100644 --- a/tests/domains_test.py +++ b/tests/domains_test.py @@ -1,5 +1,9 @@ +from unittest import TestCase +from unittest.mock import create_autospec + import resend from resend.exceptions import NoContentError +from resend.http_client import HTTPClient from tests.conftest import ResendBaseTest # flake8: noqa @@ -180,6 +184,22 @@ def test_should_verify_domains_raise_exception_when_no_content(self) -> None: domain_id="d91cd9bd-1176-453e-8fc1-35364d380206", ) + def test_domains_verify_with_forge_scope(self) -> None: + self.set_mock_json( + {"object": "domain", "id": "d91cd9bd-1176-453e-8fc1-35364d380206"} + ) + + params: resend.Domains.VerifyParams = {"scope": ["forge"]} + domain = resend.Domains.verify( + domain_id="d91cd9bd-1176-453e-8fc1-35364d380206", + params=params, + ) + assert domain["id"] == "d91cd9bd-1176-453e-8fc1-35364d380206" + assert ( + self.mock.call_args.kwargs["url"] + == "https://api.resend.com/domains/d91cd9bd-1176-453e-8fc1-35364d380206/verify" + ) + def test_domains_create_with_tracking_subdomain(self) -> None: self.set_mock_json( { @@ -393,3 +413,47 @@ def test_domains_list_with_before_param(self) -> None: assert domains["has_more"] is False assert len(domains["data"]) == 1 assert domains["data"][0]["id"] == "domain-3" + + +class TestDomainsRequestBody(TestCase): + def setUp(self) -> None: + resend.api_key = "re_123" + self.mock_client = create_autospec(HTTPClient, instance=True) + self.mock_client.request.return_value = ( + b'{"id": "d91cd9bd-1176-453e-8fc1-35364d380206"}', + 200, + {"Content-Type": "application/json"}, + ) + self.previous_http_client = resend.default_http_client + resend.default_http_client = self.mock_client + + def tearDown(self) -> None: + resend.default_http_client = self.previous_http_client + + def test_verify_sends_forge_scope(self) -> None: + params: resend.Domains.VerifyParams = {"scope": ["forge"]} + resend.Domains.verify( + domain_id="d91cd9bd-1176-453e-8fc1-35364d380206", params=params + ) + + _, kwargs = self.mock_client.request.call_args + assert kwargs["method"] == "post" + assert kwargs["json"] == {"scope": ["forge"]} + + def test_verify_sends_empty_body_without_scope(self) -> None: + resend.Domains.verify(domain_id="d91cd9bd-1176-453e-8fc1-35364d380206") + + _, kwargs = self.mock_client.request.call_args + assert kwargs["method"] == "post" + assert kwargs["json"] == {} + + def test_update_sends_forge_return_path(self) -> None: + params: resend.Domains.UpdateParams = { + "id": "d91cd9bd-1176-453e-8fc1-35364d380206", + "forge_return_path": "outbound", + } + resend.Domains.update(params) + + _, kwargs = self.mock_client.request.call_args + assert kwargs["method"] == "patch" + assert kwargs["json"]["forge_return_path"] == "outbound"