Django Object Streams

django-object-streams is a Django library for lightweight, permission-aware, filter-aware subscriptions to model objects and querysets.

It is designed to tell clients what changed and whether they should refetch through the canonical REST API. It is not a replacement for REST, a generic Channels wrapper, or an activity feed library.

Status

Pre-alpha. The current package includes the core registration API, a replayable outbox table, generic source-to-event producer helpers, and a connection-local subscription session runtime. A minimal Channels JSON websocket consumer is available with channel-layer fanout and PostgreSQL LISTEN/NOTIFY wakeups. A pruning command enforces outbox retention limits. History integrations are not implemented yet.

Database Support

This library targets PostgreSQL. That is a feature choice, not just a reduced support matrix: replayable cursors, low-latency wakeups, and future producer adapters are expected to use PostgreSQL behavior directly.

Install

uv add django-object-streams

pip install django-object-streams

Trigger-based capture needs the triggers extra, described in Trigger-Based Capture.

Django Setup

Add the app to INSTALLED_APPS:

INSTALLED_APPS = [
    "django.contrib.contenttypes",
    "object_streams",
]

Run migrations after installing:

python manage.py migrate object_streams

What a working install needs

Installing and migrating creates the outbox table. It does not capture anything or deliver anything on its own. Four more pieces make a working deployment:

1. Register the models you want subscribable, with a FilterSet and a visibility policy. See Usage Sketch.
2. Capture writes into the outbox, either by calling the producers from application code or by declaring Postgres triggers. See Trigger-Based Capture.
3. Route the consumer into your ASGI application behind auth middleware. See ASGI routing and Authentication.
4. Run the listener process, object_streams_listen, and give it a process-shared channel layer. See Listener process and Channel layer.

Step 4 is the one that fails quietly. Without a running listener the application still writes outbox rows and still sends NOTIFY, but nothing fans out to the ASGI workers, so subscriptions are accepted and no events ever arrive.

Core Concepts

• Subject: the user-visible object subscribers care about.
• Source: the row, event, or history record that caused the stream event.
• Facet: the part of a subject that changed, such as object, workflow_state, permissions, or related.
• Cursor: the global commit-visible delivery sequence used for reconnect and replay. It is separate from the outbox row id used by internal wakeups.
• Visibility policy: the permission-scoped queryset for a user and model.
• Subscription: an object, filtered queryset, or model-level watch.

Usage Sketch

Register models that can be subscribed to:

import django_filters

from object_streams import AllowAllVisibilityPolicy
from object_streams import ModelSource
from object_streams import register

from store.models import CustomerOrder


class CustomerOrderSubscriptionFilter(django_filters.FilterSet):
    class Meta:
        model = CustomerOrder
        fields = ["status"]


register(
    CustomerOrder,
    filterset=CustomerOrderSubscriptionFilter,
    visibility=AllowAllVisibilityPolicy(),
    sources=[ModelSource(CustomerOrder)],
)

Produce outbox events after application writes:

from object_streams import EventOperation
from object_streams.producers import enqueue_source_events


order.status = "open"
order.save(update_fields=["status"])
enqueue_source_events(
    order,
    op=EventOperation.UPDATED,
    changed_fields=["status"],
)

For scripts, tests, or jobs that need the outbox row immediately, use create_source_events(...) instead.

Capture and delivery are separate. A row created inside an open transaction has no public cursor until that transaction commits and the listener can observe it. This prevents concurrent transactions that commit in a different order from creating gaps a reconnecting client could skip.

Outbox writes send a PostgreSQL notification with the committed outbox id by default. Pass notify=False to create_source_events(...), enqueue_source_events(...), create_outbox_event(...), or enqueue_outbox_event(...) when creating rows that should not wake connected consumers.

Trigger-Based Capture

Calling the producers from application code works, but it depends on every write site remembering to do it. Postgres triggers do not:

• The outbox row is written inside the same transaction as the write that caused it, so nothing is lost between commit and a callback.
• DELETE is captured. post_save cannot see deletes, and a post_delete receiver is a second thing to remember.
• Bulk writes are captured. QuerySet.update() and bulk_create() bypass save() and signals entirely.
• changed_fields is computed by comparing the old and new row, so it is populated whether or not the write passed update_fields.

Install the extra:

uv add "django-object-streams[triggers]"

Add pgtrigger to INSTALLED_APPS, then declare capture in model state so migrations pick it up. On a model you own:

from object_streams.triggers import ObjectStreamTrigger


class Order(models.Model):
    class Meta:
        triggers = [ObjectStreamTrigger(name="order_stream")]

On a model you do not own, declare it on a proxy in your own app so the migration lands in your app rather than in the one that defines the model:

class OrderStream(ThirdPartyOrder):
    class Meta:
        proxy = True
        triggers = [ObjectStreamTrigger(name="order_stream")]

Events captured through a proxy use the concrete model's label, so they match a registration for ThirdPartyOrder rather than the proxy itself.

Run makemigrations and migrate afterwards. A declared trigger that was never migrated is not installed, and capture silently does nothing.

Every captured row records the Postgres transaction id in metadata:

{"transaction_id": "3208276"}

Events written by one database transaction share it, so a client can coalesce them into a single refetch instead of one per row. This is not an application action identifier: one request or audit context may span several transactions.

The trigger sends its own pg_notify, which is transactional: the notification arrives when the transaction commits and is discarded when it rolls back. It reads the channel from a database setting so that changing it needs no migration:

ALTER DATABASE mydb SET object_streams.notify_channel = 'my_channel';

Without that setting the trigger uses object_streams_events, the same default as OBJECT_STREAMS_NOTIFY_CHANNEL. Set both when you override either.

Three limits are worth knowing:

• The captured row is its own subject, which suits models whose own writes are what subscribers care about. A source whose subject is a different object, such as a workflow state row pointing at the object it governs, needs its subject mapping written as SQL or left to a Python producer.
• changed_fields reports database column names, so a foreign key appears as supplier_id rather than supplier.
• Composite primary keys are not supported. Trigger compilation raises a focused error instead of generating a subject id that cannot represent the key.

Producers and triggers can coexist, but not on the same table: a table with a capture trigger should not also have producers called against it, or each write lands in the outbox twice.

Handle connection-local subscriptions with a transport object:

from object_streams.sessions import SubscriptionSession


session = SubscriptionSession(
    user=request.user,
    request=request,
    transport=transport,
)

session.handle_message(
    {
        "op": "subscribe",
        "kind": "filter",
        "model": "store.CustomerOrder",
        "filter": {"status": "open"},
        "cursor": 120000,
    }
)

Filter subscriptions use the registered django-filter compatible FilterSet. The FilterSet receives the visibility-scoped queryset and the request passed to SubscriptionSession, so request-aware method filters can use the same context as normal Django view filters. Clients may send either filter or filters. An empty filter object is valid and means no FilterSet constraints beyond visibility. Object and model subscriptions use visibility only.

Public Import Surface

The package root exports only names that are safe to import before django.setup() runs:

from object_streams import AllowAllVisibilityPolicy
from object_streams import DenyAllVisibilityPolicy
from object_streams import EventOperation
from object_streams import ListAction
from object_streams import ModelSource
from object_streams import ObjectRef
from object_streams import ObjectStreamRegistration
from object_streams import ObjectStreamRegistry
from object_streams import ResyncRequired
from object_streams import Source
from object_streams import SourceRef
from object_streams import StreamEvent
from object_streams import SubscriptionKind
from object_streams import SubscriptionRequest
from object_streams import VisibilityPolicy
from object_streams import register
from object_streams import registry

Registration errors (ObjectStreamsError, RegistrationError, AlreadyRegistered, NotRegistered, FilterValidationError) are exported from the root as well.

Import the rest from their own modules, because they load Django models: object_streams.models, object_streams.outbox, object_streams.postgres, object_streams.producers, object_streams.retention, object_streams.sessions, and object_streams.transports.channels.

Protocol Reference

The protocol is JSON messages in both directions. Client messages carry an op. Server messages carry a type.

Client to server

subscribe registers one subscription and returns a subscribed acknowledgement.

Field	Required	Meaning
op	yes	"subscribe".
kind	no	"object", "filter", or "model". Defaults to "object".
model	yes	Model label, such as "store.CustomerOrder".
pk	for object	Primary key of the subscribed object.
filter	no	FilterSet parameters. filters is accepted as an alias.
cursor	no	Last cursor the client processed. Triggers replay.
subscription_id	no	Client-chosen id. The server assigns sub_N when omitted.
search	no	Rejected with unsupported_search for now.
ordering	no	Accepted and echoed. Does not affect membership.
shape	no	Accepted and echoed. Transport payload hint.

{"op": "subscribe", "kind": "filter", "model": "store.CustomerOrder", "filter": {"status": "open"}, "cursor": 120000}

unsubscribe removes one subscription.

{"op": "unsubscribe", "subscription_id": "sub_7"}

Server to client

subscribed acknowledges a subscription. Its cursor is the outbox cursor the subscription starts from, which is the global cursor at subscribe time, not the cursor the client requested.

{
  "type": "subscribed",
  "subscription_id": "sub_7",
  "kind": "filter",
  "model": "store.CustomerOrder",
  "filter": {"status": "open"},
  "cursor": 120044
}

unsubscribed acknowledges removal.

{
  "type": "unsubscribed",
  "subscription_id": "sub_7"
}

event reports one subscription-relative change.

{
  "type": "event",
  "subscription_id": "sub_7",
  "cursor": 120044,
  "subject": {"model": "store.CustomerOrder", "pk": "123"},
  "facet": "workflow_state",
  "op": "updated",
  "list_action": "changed",
  "changed_fields": ["workflow_state"],
  "fetch": true
}

op is the source operation: created, updated, or deleted.

list_action is the effect on this subscription:

list_action	Meaning
added	The object now belongs in the subscribed set.
changed	The object was in the set and still is.
removed	The object left the set but still exists.
deleted	The object was deleted.

Object subscriptions can usually ignore list_action. Filter and model subscriptions need it. Events with no effect on a subscription are not sent.

resync_required is a subscription-level message for cases where the server cannot safely compute the delta. It carries no subject. The client should refetch through REST and resubscribe from the supplied cursor.

{
  "type": "resync_required",
  "subscription_id": "sub_7",
  "cursor": 120044,
  "reason": "cursor_replay_unavailable"
}

reason	Cause
cursor_replay_unavailable	A filter or model subscription asked to replay from an older cursor. Collection membership cannot be recomputed from outbox rows alone.
object_replay_limit_exceeded	An object subscription had more missed events than replay_limit, which defaults to 1000.
cursor_pruned	The requested cursor is older than the retention watermark, so the missed rows are gone.

error reports a rejected message. details is present only when the error carries structured data, such as FilterSet validation errors.

{
  "type": "error",
  "code": "invalid_filter",
  "message": "Subscription filters are invalid.",
  "details": {"status": ["Select a valid choice."]}
}

code	Cause
invalid_request	Unsupported op, missing model, missing subscription_id, or a non-object JSON message.
invalid_cursor	The requested cursor is newer than the outbox.
invalid_filter	FilterSet validation failed. details carries the errors.
not_found	An object subscription target does not exist or is not visible.
not_subscribed	Unsubscribe named an inactive subscription.
unsupported_search	The request included search.
duplicate_subscription	The subscription_id is already active on this connection.
subscription_limit_exceeded	The connection already holds max_subscriptions subscriptions.
subscription_too_large	The subscription matches more than max_member_pks objects.
invalid_event	A fanout message arrived without an outbox id.
event_not_found	A fanout message named an outbox row that does not exist.

Deployment

Process topology

A production deployment runs three kinds of process against one PostgreSQL database and one process-shared channel layer:

ASGI workers      run ObjectStreamConsumer, hold connection-local subscriptions
listener process  runs object_streams_listen, turns NOTIFY into channel fanout
application       writes model changes and outbox rows

The path from a write to a client is:

application transaction commits
outbox row is created and NOTIFY sends its internal id
listener assigns a commit-visible delivery cursor
listener fans the row out to Channels groups
each ASGI worker evaluates the row against its own subscriptions
consumer sends subscription-relative messages

Subscription state is connection-local. Every worker evaluates visibility and filter membership for its own connections, so no permission decision is shared between processes. Only outbox ids cross the channel layer.

Run one active listener for each database and notification channel. The durable watermark makes a restarted listener safe, but concurrent listeners are not a supported high-availability topology because they can fan out cursors out of order.

ASGI routing

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter
from channels.routing import URLRouter
from django.urls import path

from object_streams.transports.channels import ObjectStreamConsumer


application = ProtocolTypeRouter(
    {
        "websocket": AuthMiddlewareStack(
            URLRouter(
                [
                    path("ws/object-streams/", ObjectStreamConsumer.as_asgi()),
                ]
            )
        ),
    }
)

The consumer joins deterministic Channels groups for active subscriptions and publishes outbox rows when it receives an object.stream.event ASGI message containing id or outbox_id. Object subscriptions join object groups. Filter and model subscriptions join model groups, then each consumer evaluates the event against its own connection-local subscriptions and visibility policy.

Your ASGI workers can run under Uvicorn, Daphne, or another ASGI server.

Authentication

The consumer reads scope["user"] and passes it to every visibility policy. It does not authenticate connections itself, so the ASGI stack must populate that key before the consumer runs. AuthMiddlewareStack does this from the session cookie. Substitute your own middleware for token or header authentication.

Without auth middleware, scope["user"] is unset and every policy receives None. The default AllowAllVisibilityPolicy ignores its user argument, so a registration that does not set visibility delivers every event for the model to every connection. Register a visibility policy for any model whose events are not public, and write that policy to deny unauthenticated users:

class OwnerVisibilityPolicy:
    def get_queryset(self, user, model, action="read"):
        if user is None or not user.is_authenticated:
            return model._default_manager.none()
        return model._default_manager.filter(owner=user)

Rejecting unauthenticated connections at the transport is the stronger option. Subclass the consumer and close the socket in connect() when the scope has no authenticated user.

Connection limits

Every subscription costs memory and query work for the life of the connection. A filter or model subscription holds the primary key of every matching object so it can classify later events as added, changed, or removed, and each event on a registered model runs one membership check per active subscription on that model. A client that opens many subscriptions therefore multiplies the cost of every write.

The consumer bounds this with three class attributes:

Attribute	Default	Effect
max_subscriptions	100	Active subscriptions per connection. None disables the limit.
max_member_pks	10000	Objects one subscription may match. None disables the limit.
replay_limit	1000	Object replay events sent before falling back to resync_required.

Subclass the consumer to change them:

class WideObjectStreamConsumer(ObjectStreamConsumer):
    max_subscriptions = 500
    max_member_pks = 50000

These are backstops, not a substitute for authentication. Raise them deliberately, and keep in mind that max_member_pks is checked with a bounded query before the membership set is materialized.

Channel layer

The listener and the ASGI workers are separate processes, so the in-memory channel layer will not deliver between them. Production deployments need a process-shared layer such as Redis:

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": ["redis://127.0.0.1:6379/0"],
        },
    },
}

Install channels-redis alongside this package to use that backend. Every ASGI worker and the listener process must point at the same layer.

Listener process

python manage.py object_streams_listen

Option	Meaning
--database	Database alias to listen on. Defaults to default.
--channel	PostgreSQL notification channel. Defaults to the configured channel.
--timeout	Maximum seconds to wait for a notification before returning.
--once	Broadcast one notification and exit.
--retry-delay	Seconds to wait before retrying after a database error. Defaults to 1.
--max-retries	Maximum reconnect attempts. Defaults to retrying forever.

The listener retries database errors by default, so a PostgreSQL restart does not need a supervisor restart. After it starts listening, and after every reconnect, it drains committed rows whose successful fanout was not recorded. PostgreSQL notifications are therefore wakeups rather than the durable delivery ledger. Use --once --timeout <seconds> for smoke tests or supervisor health checks.

Settings

Setting	Default	Meaning
OBJECT_STREAMS_NOTIFY_CHANNEL	object_streams_events	PostgreSQL LISTEN/NOTIFY channel carrying committed outbox ids. Must be an unquoted identifier of 63 bytes or fewer.
OBJECT_STREAMS_RETENTION_DAYS	None	Age limit for outbox rows, in days. None keeps every row.
OBJECT_STREAMS_RETENTION_MAX_ROWS	None	Row limit for the outbox. None keeps every row.

Give each deployment sharing a database its own OBJECT_STREAMS_NOTIFY_CHANNEL so listeners do not wake on ids they cannot load.

Outbox Retention

The outbox grows without bound until it is pruned. Configure a limit and run the pruning command on a schedule:

OBJECT_STREAMS_RETENTION_DAYS = 30
OBJECT_STREAMS_RETENTION_MAX_ROWS = 5_000_000

python manage.py object_streams_prune

When both limits are set, the stricter one wins. --days and --max-rows override the settings, --dry-run reports what would be deleted, and --database selects a database alias.

Retention sets the cursor contract. Pick a window longer than the longest client disconnect you want to replay rather than resync.

Two properties keep pruning safe:

• Pruning does not delete captured rows until the listener has recorded their successful channel-layer fanout.
• Pruning never deletes the newest retained row. The global cursor never moves backwards, so a returning client never sees invalid_cursor for a cursor it legitimately holds.
• Pruning records how far it deleted. A client that reconnects with a cursor older than that watermark receives resync_required with reason cursor_pruned rather than a silent, empty catch-up.

Development

Set up the local environment:

just bootstrap

Tests expect a local PostgreSQL server and a role that can create test databases. By default, tests.settings connects through the local PostgreSQL socket as the postgres role and uses a database named django_object_streams.

The Redis-backed live listener integration test runs when Redis is available. It uses OBJECT_STREAMS_TEST_REDIS_URL, defaulting to redis://localhost:6379/15, and skips when Redis is not reachable.

Create the local database if needed:

createdb -U postgres django_object_streams

To override the connection, copy .env.local.example to .env.local and set the database variables there.

just check
just test
just coverage

CI enforces a coverage floor of 80 percent.

Package code follows standard uv, setuptools, ruff, pytest-django, and Justfile conventions for a focused Django library.

License

BSD 3-Clause. See LICENSE.