stapel-core 0.3.1

Core Django utilities for the Stapel framework

stapel_core

Shared Python library for Stapel services. Provides JWT authentication, captcha verification, event bus, notifications, and Django utilities used across all backend services.

Part of the Stapel framework.

Quick start for a new Django service

pip install -e ../iron-common-python

Add to INSTALLED_APPS:

INSTALLED_APPS = [
    ...
    'stapel_core.django',
    'stapel_core.django.users',   # if using the shared User model
]

Modules

stapel_core.captcha — Pluggable captcha verification

Backend-agnostic captcha interface. Supports Cloudflare Turnstile, Google reCAPTCHA v2, hCaptcha, and custom backends.

Settings (per service, in settings/base.py):

CAPTCHA_BACKEND = env.str('CAPTCHA_BACKEND', 'turnstile')
CAPTCHA_SECRET  = env.str('CAPTCHA_SECRET', None)  # absent → disabled

Auto-disable: if CAPTCHA_SECRET is None or empty, build_verifier returns NoopVerifier regardless of backend. No separate toggle needed.

DRF integration (add mixin to any serializer):

from stapel_core.django.captcha import CaptchaMixin

class MySerializer(CaptchaMixin, serializers.Serializer):
    captcha_token = serializers.CharField(required=False, allow_blank=True)

    def validate(self, attrs):
        self._require_captcha_if_configured(attrs)
        return attrs

Custom backend — subclass CaptchaVerifier and point to it via a dotted import path:

from stapel_core.captcha import CaptchaVerifier

class MyCaptchaVerifier(CaptchaVerifier):
    def verify(self, token: str, ip: str | None = None) -> bool:
        return my_service.check(token, self.secret)

# settings.py
CAPTCHA_BACKEND = 'myapp.captcha.MyCaptchaVerifier'
CAPTCHA_SECRET  = 'my-secret'

---

stapel_core.django.jwt — JWT authentication

Unified JWT provider (singleton). Supports HS256 and RS256.

from stapel_core.django.jwt.provider import jwt_provider

access, refresh = jwt_provider.create_tokens(user)
payload = jwt_provider.validate_token(access_token)

Settings:

JWT_ALGORITHM    = 'HS256'           # or 'RS256'
JWT_SECRET_KEY   = 'your-secret'     # HS256
JWT_PRIVATE_KEY  = '...'             # RS256
JWT_PUBLIC_KEY   = '...'             # RS256
JWT_ISSUER       = 'https://yourapp.com'
JWT_AUDIENCE     = None
JWT_ACCESS_TOKEN_LIFETIME  = 900     # seconds
JWT_REFRESH_TOKEN_LIFETIME = 604800  # seconds

---

stapel_core.django.authentication — JWT cookie auth

JWTCookieAuthentication reads JWT from access_token cookie or Authorization: Bearer <token> header.

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'stapel_core.django.authentication.JWTCookieAuthentication',
    ],
}

---

stapel_core.django.api — DRF utilities

Symbol	Purpose
StapelDataclassSerializer	Serializer that maps @dataclass fields
StapelResponse(serializer)	Wraps .data automatically
StapelErrorResponse(status, ERR_KEY)	Structured error response
StapelValidationError(ERR_KEY)	Raises DRF validation error with error key
register_service_errors(dict)	Registers error messages for a service
AnchorPagination / CreatedAtAnchorPagination	Cursor-style paginators

---

stapel_core.bus — Event bus

Kafka-backed event bus with an in-memory backend for tests.

Publish (sync, fire-and-forget):

from stapel_core.bus import publish, Event

publish('user.created', Event(
    event_type='user.created',
    service='auth',
    payload={'user_id': '...'},
))

Consume by subclassing the management-command base:

from stapel_core.bus import BaseBusConsumerCommand, Event

class ConsumeUsers(BaseBusConsumerCommand):
    topics = ['user.created']
    consumer_group = 'notifications'

    def handle_event(self, event: Event) -> None:
        ...

Backend is selected via the STAPEL_BUS_BACKEND Django setting (stapel_core.bus.backends.kafka.KafkaBus in production, stapel_core.bus.backends.memory.MemoryBus in tests).

---

stapel_core.notifications — Push notifications

from stapel_core.notifications import request_notification

request_notification(
    notification_type='welcome',
    user_id=str(user.id),
    email=user.email,
    variables={'name': user.username},
    source_service='auth',
)

---

stapel_core.oauth — OAuth provider registry

Provider classes (GoogleProvider, GitHubProvider, etc.) and registry for OAuth consumer flows (when your service accepts OAuth logins from external providers).

---

stapel_core.gdpr — GDPR utilities

Account closure requests, data export, re-registration hashes.

---

Running tests

cd iron-common-python
pip install -e '.[dev]'
pytest stapel_core/tests/ -v