Django, FastAPI and dependency injection

These source-checkout examples make synchronous SMS submissions behind an application boundary. Install the optional dependencies:

python -m pip install -e '.[dev,examples]'
python -m pytest tests/test_documentation_examples.py --no-cov

No credentials are read or HTTP requests made when importing the examples. Tests inject a fake or HTTPX MockTransport. Real factories require your own AlphaSMS key; never replace it with the synthetic test key for a live request. Framework versions are resolved for each CI Python version; older Python environments may resolve older framework releases. Check your framework's own supported-version policy separately from Jukto's Python compatibility.

These endpoints demonstrate integration mechanics, not a complete messaging service. Mount them inside your authenticated/authorized order workflow, enforce consent and rate limits, and persist operation/outcome evidence before retrying. Django's normal CSRF middleware should remain enabled. The examples deliberately avoid logging or returning raw provider data; they do not implement durable reconciliation storage or a delivery receipt processor.

Shared application boundary

The consumer depends on BaseSMSProvider, makes exactly one attempt and returns only outcome metadata. Empty inputs fail before submission. Provider phone/content constraints still require separate validation against a verified specification. Batch exceptions retain per-entry outcomes; ambiguous failures remain explicit. HTTP 202 means the example processed a submission result, not guaranteed delivery. HTTP 502 is not a safe-retry signal.

"""An application boundary shared by Django and FastAPI examples."""
from __future__ import annotations
import os
from typing import Any
from jukto import AlphaSMSClient, BaseSMSProvider, JuktoConfigurationError, JuktoError, SMSBatchError, SMSSubmissionResult


def alpha_client(key: str) -> AlphaSMSClient:
    if not key or not key.strip():
        raise JuktoConfigurationError("Configure JUKTO_ALPHA_API_KEY before serving requests")
    return AlphaSMSClient(api_key=key, timeout=10.0)


def alpha_from_environment() -> AlphaSMSClient:
    return alpha_client(os.environ.get("JUKTO_ALPHA_API_KEY", ""))


def safe_result(result: SMSSubmissionResult) -> dict[str, Any]:
    # Do not expose raw responses, SMS content, credentials or recipient information.
    return {"status": result.status.value,
            "reconciliation_required": result.reconciliation_required,
            "recipients": [{"response_index": item.response_index, "status": item.status.value}
                           for item in result.recipients]}


def notify(provider: BaseSMSProvider, payload: Any) -> tuple[dict[str, Any], int]:
    if not isinstance(payload, dict) or any(not isinstance(payload.get(key), str) or not payload[key].strip()
                                          for key in ("phone", "message")):
        return {"error": "phone and message must be nonempty strings"}, 400
    try:
        result = provider.submit_sms(payload["phone"], payload["message"])
    except SMSBatchError as error:
        if error.normalized_result is not None:
            return safe_result(error.normalized_result), 202
        return {"error": "batch_result_unavailable", "status": "unknown", "reconciliation_required": True}, 502
    except JuktoError as error:
        # One attempt only. The application must persist/reconcile uncertain effects.
        return {"error": "provider_error", "status": error.outcome.value,
                "reconciliation_required": error.reconciliation_required}, 502
    return safe_result(result), 202

Django

Set JUKTO_ALPHA_API_KEY = os.environ["JUKTO_ALPHA_API_KEY"] in Django settings, using your deployment's secret store. Mount sms_view in your URL configuration with path("sms", sms_view). A missing/empty key returns a safe configuration error. This sync view blocks its request worker during SDK calls and closes a request-owned provider on exit, including failure paths. It trades cross-request pooling for simple ownership; use an explicitly managed worker-local lifetime if you need reuse across requests. Do not hide initialization in AppConfig.ready() without accounting for reload/fork and shutdown behavior.

Django's view documentation describes the request/response boundary.

"""Sync Django view with request-owned provider lifetime and a factory injection seam."""
from __future__ import annotations
import json
from typing import Callable
from django.conf import settings
from django.http import HttpRequest, JsonResponse
from django.views.decorators.http import require_POST
from jukto import BaseSMSProvider, JuktoConfigurationError
from examples.application import alpha_client, notify


def configured_provider() -> BaseSMSProvider:
    return alpha_client(getattr(settings, "JUKTO_ALPHA_API_KEY", ""))


def make_sms_view(factory: Callable[[], BaseSMSProvider] = configured_provider) -> Callable[[HttpRequest], JsonResponse]:
    @require_POST
    def view(request: HttpRequest) -> JsonResponse:
        try:
            payload = json.loads(request.body)
        except (ValueError, UnicodeDecodeError):
            return JsonResponse({"error": "invalid_json"}, status=400)
        try:
            # SDK calls block this sync request worker. Fresh provider per request,
            # closed even on validation or provider failures; no import-time network.
            with factory() as provider:
                body, status = notify(provider, payload)
        except JuktoConfigurationError:
            return JsonResponse({"error": "provider_configuration"}, status=503)
        return JsonResponse(body, status=status)
    return view


sms_view = make_sms_view()

FastAPI

Set JUKTO_ALPHA_API_KEY in your deployment environment. Run uvicorn examples.fastapi_app:app from the source checkout. The factory runs at lifespan startup and an absent key fails startup before serving requests. The provider is reused across requests in one worker, then closed after serving ends. Do not close it inside an individual request. Each process has a separate provider.

The route is a normal def, so FastAPI runs the blocking SDK operation in a worker thread. Shutdown cleanup is also offloaded. For native awaited requests in the alpha, use the async lifespan example. Follow FastAPI's lifespan and sync/async guidance.

"""Application-owned sync provider; blocking operations run in FastAPI's thread pool."""
from __future__ import annotations
from contextlib import asynccontextmanager
from typing import Any, AsyncIterator, Callable
from fastapi import Depends, FastAPI, Request
from fastapi.responses import JSONResponse
from starlette.concurrency import run_in_threadpool
from jukto import BaseSMSProvider
from examples.application import alpha_from_environment, notify


def get_provider(request: Request) -> BaseSMSProvider:
    return request.app.state.sms_provider


def create_app(factory: Callable[[], BaseSMSProvider] = alpha_from_environment) -> FastAPI:
    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncIterator[None]:
        provider = factory()  # Construction is lazy: it does not perform HTTP.
        app.state.sms_provider = provider
        try:
            yield
        finally:
            await run_in_threadpool(provider.close)

    app = FastAPI(lifespan=lifespan)

    @app.post("/sms")
    def submit(payload: dict[str, Any], provider: BaseSMSProvider = Depends(get_provider)) -> JSONResponse:
        # A normal def route runs in a worker thread; never call this sync SDK
        # directly inside an async def request handler on the event loop.
        body, status = notify(provider, payload)
        return JSONResponse(body, status_code=status)

    return app


app = create_app()  # No credentials are read until lifespan startup.

Fake provider and offline request testing

This deterministic fake implements the same injection contract but its status output is synthetic, not vendor evidence. Application tests can inject accepted, unknown or partial outcomes and exceptions without opening a network connection. Its call ledger records sensitive test inputs; do not log it in production.

"""Deterministic fake for application tests, not evidence of a vendor contract."""
from __future__ import annotations
from typing import Any, Optional
from jukto import BaseSMSProvider, JuktoConfigurationError, JuktoError, Outcome, SMSRecipientResult, SMSSubmissionResult


class FakeSMSProvider(BaseSMSProvider):
    provider_name = "ExampleFake"

    def __init__(self, outcome: Outcome = Outcome.ACCEPTED, error: Optional[JuktoError] = None) -> None:
        super().__init__(api_key="synthetic-fake-key")
        self.outcome = outcome
        self.error = error
        self.calls: list[tuple[str | list[str], str]] = []
        self.closed = False

    def submit_sms(self, phone_numbers: str | list[str], message: str) -> SMSSubmissionResult:
        if self.closed:
            raise JuktoConfigurationError("Fake provider is closed")
        self.calls.append((phone_numbers, message))
        if self.error is not None:
            raise self.error
        recipients = (SMSRecipientResult(0, Outcome.ACCEPTED), SMSRecipientResult(1, Outcome.UNKNOWN)) if self.outcome == Outcome.PARTIAL else ()
        return SMSSubmissionResult(self.provider_name, self.outcome, recipients=recipients, raw={"synthetic": True})

    def send_sms(self, phone_numbers: str | list[str], message: str) -> dict[str, Any]:
        return {"status": self.submit_sms(phone_numbers, message).status.value, "synthetic": True}

    def close(self) -> None:
        self.closed = True
        super().close()

For FastAPI use TestClient(create_app(lambda: fake)) as a context manager so startup/shutdown run. For Django pass make_sms_view(lambda: fake) a RequestFactory POST. The checked-in request tests exercise both paths, configuration failures, partial/unknown results, one-attempt timeouts, redacted output and ownership. Another test injects a real AlphaSMS adapter with MockTransport to verify reuse and that borrowed clients stay open.