Skip to content

SMS API

jukto.interfaces.sms.BaseSMSProvider

Bases: TransportLifecycle, ABC

Source code in src/jukto/interfaces/sms.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class BaseSMSProvider(TransportLifecycle, ABC):
    provider_name: str
    """The Contract: All SMS providers MUST implement these methods."""

    def __init__(
        self,
        api_key: str,
        sender_id: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: float | httpx.Timeout = 30.0,
        *,
        allow_insecure_http: bool = False,
        http_client: Optional[httpx.Client] = None,
        limits: Optional[httpx.Limits] = None,
    ) -> None:
        if base_url is not None:
            validate_endpoint(base_url, allow_insecure_http)
        self.allow_insecure_http = allow_insecure_http
        self.api_key = api_key
        self.sender_id = sender_id
        self.base_url = base_url
        self.timeout = timeout
        self._transport = SyncTransport(timeout=timeout, limits=limits, http_client=http_client,
                                        allow_insecure_http=allow_insecure_http)

    @abstractmethod
    def send_sms(self, phone_numbers: str | list[str], message: str) -> dict[str, Any] | list[Any] | str:
        """Sends an SMS to one or multiple numbers."""
        pass

    @property
    def capabilities(self) -> ProviderCapabilities:
        return ProviderCapabilities(self.provider_name, sms_submission=True,
                                    per_recipient_submission=self.provider_name == 'GreenWeb')

    def submit_sms(self, phone_numbers: str | list[str], message: str) -> SMSSubmissionResult:
        return sms_submission(self.provider_name, self.send_sms(phone_numbers, message))

provider_name instance-attribute

The Contract: All SMS providers MUST implement these methods.

send_sms(phone_numbers, message) abstractmethod

Sends an SMS to one or multiple numbers.

Source code in src/jukto/interfaces/sms.py 
36
37
38
39
@abstractmethod
def send_sms(self, phone_numbers: str | list[str], message: str) -> dict[str, Any] | list[Any] | str:
    """Sends an SMS to one or multiple numbers."""
    pass

jukto.results.Outcome

Bases: str, Enum

Source code in src/jukto/results.py 
36
37
38
39
40
41
class Outcome(str, Enum):
    ACCEPTED = 'accepted'
    DELIVERED = 'delivered'
    FAILED = 'failed'
    PARTIAL = 'partial'
    UNKNOWN = 'unknown'

jukto.results.SMSSubmissionResult dataclass

Source code in src/jukto/results.py 
76
77
78
79
80
81
82
83
84
85
86
87
@dataclass(frozen=True)
class SMSSubmissionResult:
    provider: str
    status: Outcome
    request_id: Optional[str] = field(default=None, repr=False)
    recipients: tuple[SMSRecipientResult, ...] = ()
    provider_status: Any = field(default=None, repr=False)
    raw: Any = field(default=None, repr=False)

    @property
    def reconciliation_required(self) -> bool:
        return self.status == Outcome.UNKNOWN or any(r.status == Outcome.UNKNOWN for r in self.recipients)

jukto.results.SMSRecipientResult dataclass

Source code in src/jukto/results.py 
66
67
68
69
70
71
72
73
@dataclass(frozen=True)
class SMSRecipientResult:
    response_index: int
    status: Outcome
    recipient: Optional[str] = field(default=None, repr=False)
    provider_id: Optional[str] = field(default=None, repr=False)
    provider_status: Any = field(default=None, repr=False)
    raw: Any = field(default=None, repr=False)

jukto.results.SMSBatchResult

Bases: list[Any]

Preserve ordered raw results, including failed and unrecognized entries.

SENT describes provider submission output, not independently verified delivery. Indices refer to response entries, not necessarily original request positions.

Source code in src/jukto/results.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class SMSBatchResult(list[Any]):
    """Preserve ordered raw results, including failed and unrecognized entries.

    SENT describes provider submission output, not independently verified delivery.
    Indices refer to response entries, not necessarily original request positions.
    """

    @property
    def sent_indices(self) -> tuple[int, ...]:
        return tuple(i for i, item in enumerate(self) if isinstance(item, dict) and item.get('status') == 'SENT')

    @property
    def failed_indices(self) -> tuple[int, ...]:
        return tuple(i for i, item in enumerate(self) if isinstance(item, dict) and item.get('status') == 'FAILED')

    @property
    def unknown_indices(self) -> tuple[int, ...]:
        known = set(self.sent_indices + self.failed_indices)
        return tuple(i for i in range(len(self)) if i not in known)

    @property
    def normalized(self) -> SMSSubmissionResult:
        return sms_submission('GreenWeb', self)

jukto.providers.sms.alphasms.AlphaSMSClient

Bases: AlphaSMSContract, BaseSMSProvider

Adapter for AlphaSMS (sms.net.bd) API.

Source code in src/jukto/providers/sms/alphasms.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class AlphaSMSClient(AlphaSMSContract, BaseSMSProvider):
    """Adapter for AlphaSMS (sms.net.bd) API."""

    provider_name = "AlphaSMS"

    DEFAULT_URL = "https://api.sms.net.bd/sendsms"

    def __init__(
        self,
        api_key: str,
        sender_id: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: float | httpx.Timeout = 30.0,
        *,
        allow_insecure_http: bool = False,
        http_client: Optional[httpx.Client] = None,
        limits: Optional[httpx.Limits] = None,
    ) -> None:
        super().__init__(
            api_key=api_key,
            sender_id=sender_id,
            base_url=base_url or self.DEFAULT_URL,
            timeout=timeout,
            allow_insecure_http=allow_insecure_http,
            http_client=http_client,
            limits=limits,
        )


    @operation
    def send_sms(self, phone_numbers: str | list[str], message: str) -> dict[str, Any]:
        payload = self._prepare_sms(phone_numbers, message)
        assert self.base_url is not None

        logger.debug('provider=AlphaSMS operation=send_sms event=request')

        try:
            response = self._transport.request("POST", self.base_url, data=payload)
            logger.debug('provider=AlphaSMS operation=send_sms event=response status=%d', response.status_code)
            response.raise_for_status()
            data = decode_json(response, "AlphaSMS")
            return self._inspect_response_data(data, response)
        except httpx.TimeoutException as exc:
            raise ProviderTimeoutError(
                'AlphaSMS request timed out: [details withheld]',
                provider="AlphaSMS",
            ) from None
        except httpx.HTTPStatusError as exc:
            self._handle_http_status_error(exc)
            raise
        except httpx.RequestError as exc:
            raise ProviderConnectionError(
                'Failed to connect to AlphaSMS: [details withheld]',
                provider="AlphaSMS",
            ) from None

jukto.providers.sms.greenweb.GreenWebClient

Bases: GreenWebContract, BaseSMSProvider

Adapter for GreenWeb SMS API.

Source code in src/jukto/providers/sms/greenweb.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class GreenWebClient(GreenWebContract, BaseSMSProvider):
    """Adapter for GreenWeb SMS API."""

    provider_name = "GreenWeb"

    DEFAULT_URL = "https://api.greenweb.com.bd/api.php"

    def __init__(
        self,
        api_key: str,
        sender_id: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: float | httpx.Timeout = 30.0,
        *,
        allow_insecure_http: bool = False,
        http_client: Optional[httpx.Client] = None,
        limits: Optional[httpx.Limits] = None,
    ) -> None:
        super().__init__(
            api_key=api_key,
            sender_id=sender_id,
            base_url=base_url or self.DEFAULT_URL,
            timeout=timeout,
            allow_insecure_http=allow_insecure_http,
            http_client=http_client,
            limits=limits,
        )


    @operation
    def send_sms(self, phone_numbers: str | list[str], message: str) -> dict[str, Any] | list[Any] | str:
        payload = self._prepare_sms(phone_numbers, message)
        assert self.base_url is not None

        logger.debug('provider=GreenWeb operation=send_sms event=request')

        try:
            response = self._transport.request("POST", self.base_url, data=payload)
            logger.debug('provider=GreenWeb operation=send_sms event=response status=%d', response.status_code)
            response.raise_for_status()
            return self._inspect_response_content(response)
        except httpx.TimeoutException as exc:
            logger.error('provider=GreenWeb operation=send_sms event=timeout')
            raise ProviderTimeoutError(
                'Request to GreenWeb timed out: [details withheld]',
                provider="GreenWeb",
            ) from None
        except httpx.HTTPStatusError as exc:
            self._handle_http_status_error(exc)
        except httpx.RequestError as exc:
            logger.error('provider=GreenWeb operation=send_sms event=error')
            raise ProviderConnectionError(
                'Failed to connect to GreenWeb: [details withheld]',
                provider="GreenWeb",
            ) from None

jukto.providers.sms.bulksmsbd.BulkSMSBDClient

Bases: BulkSMSBDContract, BaseSMSProvider

Adapter for BulkSMSBD (bulksmsbd.net) API.

Source code in src/jukto/providers/sms/bulksmsbd.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class BulkSMSBDClient(BulkSMSBDContract, BaseSMSProvider):
    """Adapter for BulkSMSBD (bulksmsbd.net) API."""

    provider_name = "BulkSMSBD"

    DEFAULT_URL = "http://bulksmsbd.net/api/smsapi"

    def __init__(
        self,
        api_key: str,
        sender_id: str,
        base_url: Optional[str] = None,
        timeout: float | httpx.Timeout = 30.0,
        *,
        allow_insecure_http: bool = False,
        http_client: Optional[httpx.Client] = None,
        limits: Optional[httpx.Limits] = None,
    ) -> None:
        super().__init__(
            api_key=api_key,
            sender_id=sender_id,
            base_url=base_url or self.DEFAULT_URL,
            timeout=timeout,
            allow_insecure_http=allow_insecure_http,
            http_client=http_client,
            limits=limits,
        )


    @operation
    def send_sms(self, phone_numbers: str | list[str], message: str) -> dict[str, Any]:
        payload = self._prepare_sms(phone_numbers, message)
        assert self.base_url is not None

        logger.debug('provider=BulkSMSBD operation=send_sms event=request')

        try:
            response = self._transport.request("POST", self.base_url, data=payload)
            logger.debug('provider=BulkSMSBD operation=send_sms event=response status=%d', response.status_code)
            response.raise_for_status()
            data = decode_json(response, "BulkSMSBD")
            return self._inspect_response_data(data, response)
        except httpx.TimeoutException as exc:
            raise ProviderTimeoutError(
                'BulkSMSBD request timed out: [details withheld]',
                provider="BulkSMSBD",
            ) from None
        except httpx.HTTPStatusError as exc:
            self._handle_http_status_error(exc)
            raise
        except httpx.RequestError as exc:
            raise ProviderConnectionError(
                'Failed to connect to BulkSMSBD: [details withheld]',
                provider="BulkSMSBD",
            ) from None

Native async (0.1.0a1 alpha)

jukto.interfaces.asynchronous.AsyncBaseSMSProvider

Bases: AsyncTransportLifecycle, ABC

Source code in src/jukto/interfaces/asynchronous.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class AsyncBaseSMSProvider(AsyncTransportLifecycle, ABC):
    provider_name: str
    """The Contract: All SMS providers MUST implement these methods."""

    def __init__(
        self,
        api_key: str,
        sender_id: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: float | httpx.Timeout = 30.0,
        *,
        allow_insecure_http: bool = False,
        http_client: Optional[httpx.AsyncClient] = None,
        limits: Optional[httpx.Limits] = None,
    ) -> None:
        if base_url is not None:
            validate_endpoint(base_url, allow_insecure_http)
        self.allow_insecure_http = allow_insecure_http
        self.api_key = api_key
        self.sender_id = sender_id
        self.base_url = base_url
        self.timeout = timeout
        self._transport = AsyncTransport(timeout=timeout, limits=limits, http_client=http_client,
                                        allow_insecure_http=allow_insecure_http)

    @abstractmethod
    async def send_sms(self, phone_numbers: str | list[str], message: str) -> dict[str, Any] | list[Any] | str:
        """Sends an SMS to one or multiple numbers."""
        pass

    @property
    def capabilities(self) -> ProviderCapabilities:
        return ProviderCapabilities(self.provider_name, sms_submission=True,
                                    per_recipient_submission=self.provider_name == 'GreenWeb')

    async def submit_sms(self, phone_numbers: str | list[str], message: str) -> SMSSubmissionResult:
        return sms_submission(self.provider_name, await self.send_sms(phone_numbers, message))

provider_name instance-attribute

The Contract: All SMS providers MUST implement these methods.

send_sms(phone_numbers, message) abstractmethod async

Sends an SMS to one or multiple numbers.

Source code in src/jukto/interfaces/asynchronous.py 
36
37
38
39
@abstractmethod
async def send_sms(self, phone_numbers: str | list[str], message: str) -> dict[str, Any] | list[Any] | str:
    """Sends an SMS to one or multiple numbers."""
    pass

jukto.providers.sms.async_alphasms.AsyncAlphaSMSClient

Bases: AlphaSMSContract, AsyncBaseSMSProvider

Adapter for AlphaSMS (sms.net.bd) API.

Source code in src/jukto/providers/sms/async_alphasms.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class AsyncAlphaSMSClient(AlphaSMSContract, AsyncBaseSMSProvider):
    """Adapter for AlphaSMS (sms.net.bd) API."""

    provider_name = "AlphaSMS"

    DEFAULT_URL = "https://api.sms.net.bd/sendsms"

    def __init__(
        self,
        api_key: str,
        sender_id: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: float | httpx.Timeout = 30.0,
        *,
        allow_insecure_http: bool = False,
        http_client: Optional[httpx.AsyncClient] = None,
        limits: Optional[httpx.Limits] = None,
    ) -> None:
        super().__init__(
            api_key=api_key,
            sender_id=sender_id,
            base_url=base_url or self.DEFAULT_URL,
            timeout=timeout,
            allow_insecure_http=allow_insecure_http,
            http_client=http_client,
            limits=limits,
        )


    @operation
    async def send_sms(self, phone_numbers: str | list[str], message: str) -> dict[str, Any]:
        payload = self._prepare_sms(phone_numbers, message)
        assert self.base_url is not None

        logger.debug('provider=AlphaSMS operation=send_sms event=request')

        try:
            response = await self._transport.request("POST", self.base_url, data=payload)
            logger.debug('provider=AlphaSMS operation=send_sms event=response status=%d', response.status_code)
            response.raise_for_status()
            data = decode_json(response, "AlphaSMS")
            return self._inspect_response_data(data, response)
        except httpx.TimeoutException as exc:
            raise ProviderTimeoutError(
                'AlphaSMS request timed out: [details withheld]',
                provider="AlphaSMS",
            ) from None
        except httpx.HTTPStatusError as exc:
            self._handle_http_status_error(exc)
            raise
        except httpx.RequestError as exc:
            raise ProviderConnectionError(
                'Failed to connect to AlphaSMS: [details withheld]',
                provider="AlphaSMS",
            ) from None

jukto.providers.sms.async_greenweb.AsyncGreenWebClient

Bases: GreenWebContract, AsyncBaseSMSProvider

Adapter for GreenWeb SMS API.

Source code in src/jukto/providers/sms/async_greenweb.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class AsyncGreenWebClient(GreenWebContract, AsyncBaseSMSProvider):
    """Adapter for GreenWeb SMS API."""

    provider_name = "GreenWeb"

    DEFAULT_URL = "https://api.greenweb.com.bd/api.php"

    def __init__(
        self,
        api_key: str,
        sender_id: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: float | httpx.Timeout = 30.0,
        *,
        allow_insecure_http: bool = False,
        http_client: Optional[httpx.AsyncClient] = None,
        limits: Optional[httpx.Limits] = None,
    ) -> None:
        super().__init__(
            api_key=api_key,
            sender_id=sender_id,
            base_url=base_url or self.DEFAULT_URL,
            timeout=timeout,
            allow_insecure_http=allow_insecure_http,
            http_client=http_client,
            limits=limits,
        )


    @operation
    async def send_sms(self, phone_numbers: str | list[str], message: str) -> dict[str, Any] | list[Any] | str:
        payload = self._prepare_sms(phone_numbers, message)
        assert self.base_url is not None

        logger.debug('provider=GreenWeb operation=send_sms event=request')

        try:
            response = await self._transport.request("POST", self.base_url, data=payload)
            logger.debug('provider=GreenWeb operation=send_sms event=response status=%d', response.status_code)
            response.raise_for_status()
            return self._inspect_response_content(response)
        except httpx.TimeoutException as exc:
            logger.error('provider=GreenWeb operation=send_sms event=timeout')
            raise ProviderTimeoutError(
                'Request to GreenWeb timed out: [details withheld]',
                provider="GreenWeb",
            ) from None
        except httpx.HTTPStatusError as exc:
            self._handle_http_status_error(exc)
        except httpx.RequestError as exc:
            logger.error('provider=GreenWeb operation=send_sms event=error')
            raise ProviderConnectionError(
                'Failed to connect to GreenWeb: [details withheld]',
                provider="GreenWeb",
            ) from None

jukto.providers.sms.async_bulksmsbd.AsyncBulkSMSBDClient

Bases: BulkSMSBDContract, AsyncBaseSMSProvider

Adapter for BulkSMSBD (bulksmsbd.net) API.

Source code in src/jukto/providers/sms/async_bulksmsbd.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class AsyncBulkSMSBDClient(BulkSMSBDContract, AsyncBaseSMSProvider):
    """Adapter for BulkSMSBD (bulksmsbd.net) API."""

    provider_name = "BulkSMSBD"

    DEFAULT_URL = "http://bulksmsbd.net/api/smsapi"

    def __init__(
        self,
        api_key: str,
        sender_id: str,
        base_url: Optional[str] = None,
        timeout: float | httpx.Timeout = 30.0,
        *,
        allow_insecure_http: bool = False,
        http_client: Optional[httpx.AsyncClient] = None,
        limits: Optional[httpx.Limits] = None,
    ) -> None:
        super().__init__(
            api_key=api_key,
            sender_id=sender_id,
            base_url=base_url or self.DEFAULT_URL,
            timeout=timeout,
            allow_insecure_http=allow_insecure_http,
            http_client=http_client,
            limits=limits,
        )


    @operation
    async def send_sms(self, phone_numbers: str | list[str], message: str) -> dict[str, Any]:
        payload = self._prepare_sms(phone_numbers, message)
        assert self.base_url is not None

        logger.debug('provider=BulkSMSBD operation=send_sms event=request')

        try:
            response = await self._transport.request("POST", self.base_url, data=payload)
            logger.debug('provider=BulkSMSBD operation=send_sms event=response status=%d', response.status_code)
            response.raise_for_status()
            data = decode_json(response, "BulkSMSBD")
            return self._inspect_response_data(data, response)
        except httpx.TimeoutException as exc:
            raise ProviderTimeoutError(
                'BulkSMSBD request timed out: [details withheld]',
                provider="BulkSMSBD",
            ) from None
        except httpx.HTTPStatusError as exc:
            self._handle_http_status_error(exc)
            raise
        except httpx.RequestError as exc:
            raise ProviderConnectionError(
                'Failed to connect to BulkSMSBD: [details withheld]',
                provider="BulkSMSBD",
            ) from None