Metadata-Version: 2.5
Name: outbox-sqs
Version: 0.1.0
Summary: Transactional outbox for AWS SQS and SNS — publish events atomically with your database writes.
Project-URL: Homepage, https://github.com/BasukiNathKumar/outbox-sqs
Project-URL: Issues, https://github.com/BasukiNathKumar/outbox-sqs/issues
Author: Basuki Nath Kumar
License: MIT
License-File: LICENSE
Keywords: aws,event-driven,microservices,outbox,sns,sqlalchemy,sqs,transactional-outbox
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Database
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.9
Requires-Dist: sqlalchemy>=1.4
Provides-Extra: aws
Requires-Dist: boto3>=1.26; extra == 'aws'
Provides-Extra: dev
Requires-Dist: boto3>=1.26; extra == 'dev'
Requires-Dist: pytest-cov>=4.0; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Requires-Dist: ruff>=0.1; extra == 'dev'
Description-Content-Type: text/markdown

# outbox-sqs

Transactional outbox for AWS SQS and SNS. Publish events atomically with your database writes.

```bash
pip install outbox-sqs[aws]
```

## The problem

This is the **dual-write problem**: any time a request has to update a database *and* notify another system (a queue, a topic, a webhook), and those two writes aren't the same transaction, there is a window where one succeeds and the other doesn't.

```python
session.add(Order(id=1, total=1000))
session.commit()
sqs.send_message(QueueUrl=QUEUE, MessageBody=...)   # <-- process dies here
```

The order exists, the event never went out. Downstream systems never hear about it — inventory isn't reserved, the confirmation email never sends, a sharded read-model never catches up. Swap the two lines and you get the opposite bug: the event fires, the transaction rolls back, and consumers act on an order that doesn't exist.

It doesn't take a crash to trigger this. A network blip to SQS, a throttling response, a deploy that kills the process mid-request — anything that lands between the two calls produces the same gap. And "just retry the SQS call" doesn't fix it either: now you need to know whether the first attempt actually got through before you send it again, which is the same problem one level down.

The database transaction and the SQS/SNS call are two independent systems with no shared coordinator. MySQL can't roll back an SQS publish, and SQS has no idea a MySQL transaction exists. Distributed transactions (XA/2PC) could theoretically bridge that, but SQS and SNS don't support them, and even where a broker does (Kafka doesn't either, for what it's worth), holding a DB transaction open across a network call to a broker is its own outage waiting to happen.

This isn't a new problem or one specific to AWS — it shows up anywhere a service owns both a database and a message broker. Chris Richardson's write-up on the pattern this library implements is the canonical reference: [microservices.io — Transactional Outbox](https://microservices.io/patterns/data/transactional-outbox.html).

## The fix

Turn the dual write into a single write. Instead of calling SQS directly, write the event as a row in a table in the *same* database transaction as your data — that's one write to one system, so it's atomic for free, no XA/2PC required. A separate relay process then reads that table and does the actual SQS/SNS call outside of any transaction. If your transaction rolls back, the row rolls back with it and nothing is ever published. If SQS is down or slow, the row just waits and gets picked up on the next poll — your original request already returned.

```python
from outbox_sqs import save

with Session() as session:
    session.add(Order(id=1, total=1000))
    save(
        session,
        destination=QUEUE_URL,
        payload={"event": "OrderCreated", "order_id": 1, "total": 1000},
    )
    session.commit()          # order and event commit together, or neither does
```

Then run the relay as its own process:

```bash
outbox-sqs create-table --dsn mysql+pymysql://user:pass@host/db
outbox-sqs relay        --dsn mysql+pymysql://user:pass@host/db
```

Or embed it:

```python
from outbox_sqs import Relay, SQSPublisher

Relay(Session, SQSPublisher()).run_forever()
```

That's the whole API.

## What it handles

- **Partial batch failures.** SQS `send_message_batch` can succeed for 7 entries and fail for 3. Each row is settled independently.
- **Exponential backoff.** Failed rows get pushed forward, capped at 5 minutes, then marked `failed` after `max_attempts` so a poison message can't spin forever.
- **Multiple relay instances.** Rows are claimed with `SELECT ... FOR UPDATE SKIP LOCKED` on MySQL/MariaDB/Postgres, so instances don't fight over the same rows or block each other.
- **Crashed relays.** Rows claimed but never settled are reclaimed after `claim_timeout` instead of sitting stuck forever.
- **Short lock windows.** Claim, publish, settle are three separate transactions. No row locks are held across the network call to AWS.
- **FIFO queues.** Pass `group_id` and `dedup_id` through to `MessageGroupId` / `MessageDeduplicationId`.

## Delivery semantics

**At-least-once.** If the relay publishes successfully but crashes before recording that, the message is sent again on the next cycle. This is inherent to the pattern — the database commit and the SQS ack cannot be made atomic either.

**Your consumers must be idempotent.** Include an event ID in the payload and have consumers skip IDs they've already processed. For FIFO queues, `dedup_id` gives you a five-minute dedup window from SQS itself, which helps but does not replace idempotent consumers.

Ordering is by `available_at, id` — approximately insertion order, but retries reorder things. If you need strict ordering, use a FIFO queue with a consistent `group_id`.

## Configuration

```python
Relay(
    session_factory,          # a sessionmaker
    publisher,                # SQSPublisher() / SNSPublisher()
    batch_size=10,            # rows claimed per cycle (SQS caps batches at 10)
    max_attempts=8,           # then the row is marked `failed`
    poll_interval=1.0,        # sleep only when there was nothing to do
    claim_timeout=300.0,      # seconds before an unsettled row is reclaimed
    backoff=exponential_backoff,
)
```

`save()` takes `headers` (sent as message attributes), `delay` (defer first publish), and `serializer` (defaults to `json.dumps`).

## The table

`outbox-sqs create-table` will create it, but in production you should own the schema. `outbox_sqs.MYSQL_DDL` gives you the exact DDL to paste into a migration.

The relay's hot query is covered by `ix_outbox_dispatch (status, available_at, id)`.

**Housekeeping:** the relay never deletes rows. Add a job that trims `published` rows older than a few days, or the table grows forever:

```sql
DELETE FROM outbox_events
WHERE status = 'published' AND published_at < NOW() - INTERVAL 7 DAY
LIMIT 10000;
```

## Operating it

`outbox-sqs status` counts rows by state. The two things worth alerting on:

- rows in `failed` — something needs a human
- oldest `pending` row age — the relay is down or falling behind

## Limitations

- Polling, not CDC. Expect sub-second to a few seconds of publish latency. If you need lower, Debezium is the right tool.
- Sync SQLAlchemy only. Async support is planned.
- One relay does one backend at a time. Run two if you publish to both SQS and SNS.

## Comparison

| | outbox-sqs | Debezium | roll your own |
|---|---|---|---|
| Setup | pip install | Kafka Connect cluster | a weekend, then bugs |
| Latency | ~1s | ~ms | depends |
| SQS/SNS native | yes | via Kafka | yes |
| Ops burden | one process | a cluster | yours |

## License

MIT
