airlock-py 0.1.0a1

Explicit, lifecycle-scoped control of side effects

airlock

Express side effects anywhere. Control whether & when they escape.

tl;dr

import airlock

class Order:
    def process(self):
        self.status = "processed"
        airlock.enqueue(notify_warehouse, self.id)
        airlock.enqueue(send_confirmation_email)

The execution context decides when (and whether) your side effects actually get dispatched:

# Production API endpoint: flush at end of request
with airlock.scope():
    order.process()
# side effects dispatch here

# Migration: suppress everything
with airlock.scope(policy=airlock.DropAll()):
    order.process()
# nothing dispatched

# Test: fail if anything tries to escape
with airlock.scope(policy=airlock.AssertNoEffects()):
    order.hopefully_pure_function() # raises if any enqueue() called

# Test: surface the side effects
with airlock.scope(policy=airlock.DropAll()) as scope:
    order.process()
    assert len(scope.intents) == 2
    print((intent.name, intent.args, intent.kwargs) for intent in scope.intents)

# Admin API endpoint: selective control
with airlock.scope(policy=airlock.BlockTasks({"send_confirmation_email"})) as scope:
    order.process()
    assert len(scope.intents) == 2  # the blocked task remains enqueued while we're in the scope
# side effects dispatch or discard here -- warehouse notified, but no confirmation email sent 

Using Django? Maybe with Celery?

# settings.py
MIDDLEWARE = [
    # ... other middleware ...
    "airlock.integrations.django.AirlockMiddleware",
]

# models.py
import airlock
from . import tasks

class Order(models.Model):
    def process(self):
        self.status = "processed"
        self.save()
        airlock.enqueue(tasks.send_confirmation_email, order_id=self.id)
        airlock.enqueue(tasks.notify_warehouse, order_id=self.id)

# views.py
def checkout(request):
    order = Order.objects.get(id=request.POST['order_id'])
    order.process()
    return HttpResponse("OK")
# Celery tasks dispatch in middleware, after transaction has committed

Read more: Django integration

Installation

pip install airlock-py

Documentation

Full documentation

Key pages:

• The Problem - Why airlock exists
• Core Model - The 3 concerns (Policy/Executor/Scope)
• Nesting - Nested scopes and safety
• Alternatives - Do I really need this...?

Contributing

See CONTRIBUTING.md for development setup.

License

MIT