django-manager 1.3.3

Standardized service layer and data processing abstractions for Django applications.

django-manager

Standardized service layer and data processing abstractions for Django applications.

django-manager provides a collection of reusable patterns, middleware, and utilities that help you build well-structured Django REST APIs without reinventing the wheel on every project.

---

Features

• Service Layer Abstractions — Decouple business logic from views using a clean service-oriented architecture.
• Pipeline Engine — Build composable data processing pipelines with built-in error handling and branching support.
• Adapter Registry — Plug-and-play adapter pattern with health monitoring, circuit breakers, and async support.
• Schema Validation — Lightweight schema definitions, migration helpers, and backward-compatibility checks.
• Transport Layer — Unified transport abstraction with connection pooling, retries, and metrics collection.
• Event Emitters — In-process event bus with filtering, async dispatch, history tracking, and replay.
• Worker Toolkit — Task queue, worker pools, retry policies, and result backends for background processing.
• Connector Framework — Managed external service connectors with health checks, pooling, and factory pattern.
• Codec Support — Pluggable serialization with content negotiation, streaming, and integrity verification.

Quick Start

Installation

pip install django-manager

Add to your Django project

# settings.py
INSTALLED_APPS = [
    ...
    "django_manager.fabric",
]

Basic Pipeline Example

from django_manager.pipeline import PipelineEngine, Stage

class ValidateInput(Stage):
    name = "validate"

    def process(self, data, context):
        if "email" not in data:
            raise ValueError("Email is required")
        return data

class NormalizeData(Stage):
    name = "normalize"

    def process(self, data, context):
        data["email"] = data["email"].strip().lower()
        return data

engine = PipelineEngine()
engine.add_stage(ValidateInput())
engine.add_stage(NormalizeData())

result = engine.execute({"email": "  [e-mail]  "})
# {"email": "[e-mail]"}

Adapter Pattern

from django_manager.adapters import AdapterRegistry, BaseAdapter

class EmailAdapter(BaseAdapter):
    name = "email"

    def execute(self, **kwargs):
        # send email logic
        return {"status": "sent", "to": kwargs["recipient"]}

registry = AdapterRegistry()
registry.register(EmailAdapter())

response = registry.get("email").execute(recipient="[e-mail]")

Event Bus

from django_manager.emitters import EventBus

bus = EventBus()

@bus.on("user.created")
def send_welcome(event, **kwargs):
    print(f"Welcome {kwargs['username']}!")

bus.emit("user.created", username="alice")

Schema Validation

from django_manager.schema import SchemaBuilder

schema = (
    SchemaBuilder("UserInput")
    .string("username", required=True)
    .string("email", required=True)
    .integer("age")
    .build()
)

Architecture

django_manager/
├── adapters/       # Adapter pattern with registry, monitoring, policies
├── pipeline/       # Composable data processing pipelines
├── codecs/         # Serialization, compression, content negotiation
├── transport/      # HTTP transport, pooling, retries, metrics
├── workers/        # Task queue, scheduling, result backends
├── emitters/       # Event bus, filtering, async dispatch
├── connectors/     # External service connectors with health checks
├── schema/         # Schema definitions, validation, migrations
├── fabric/         # REST API toolkit (views, serializers, auth)
├── scaffold/       # Project configuration helpers
├── resolvers/      # Data resolution and content processing
├── dispatchers/    # Request dispatching and routing logic
└── providers/      # Authentication and identity providers

Configuration

django-manager can be configured via Django settings:

# settings.py
DJANGO_MANAGER = {
    "DEFAULT_PAGINATION_SIZE": 25,
    "TRANSPORT_TIMEOUT": 30.0,
    "TRANSPORT_MAX_RETRIES": 3,
    "WORKER_POOL_SIZE": 4,
    "EVENT_HISTORY_SIZE": 10000,
    "ADAPTER_CIRCUIT_BREAKER_THRESHOLD": 5,
}

Requirements

• Python ≥ 3.12
• Django ≥ 6.0
• Django REST Framework ≥ 3.16

Changelog

1.3.2 (2026-02-14)

• Fixed schema backward-compatibility check for optional→required field transitions
• Improved transport retry budget token replenishment logic

1.3.1 (2026-02-01)

• Added AsyncBatchConnector for concurrent connector operations
• Fixed EventHistory.replay() ordering when using since parameter

1.3.0 (2026-01-18)

• New schema package with validation, migration, and diff support
• Added emitters.history module for event recording and replay
• Transport metrics now include p99 latency tracking

1.2.0 (2025-12-20)

• Connector framework with health checks, pooling, and factory pattern
• Worker result backends (in-memory, file-based)
• Added pipeline.branches for conditional and parallel stages

1.1.0 (2025-11-15)

• Event emitter system with async bus, filters, and decorators
• Worker task queue with priority scheduling
• Transport middleware chain support

1.0.0 (2025-10-01)

• Initial stable release
• Pipeline engine, adapter registry, codec layer
• REST API fabric with token auth and custom pagination

License

MIT License — see LICENSE for details.