🔗 daplug-sql (da•plug)

Schema-Driven SQL Normalization & Event Publishing for Python

daplug-sql wraps psycopg2 / mysql-connector with optimistic CRUD helpers and SNS event fan-out so your Postgres and MySQL services stay DRY and event-driven.

📎 Agents – a dedicated playbook lives in .agents/AGENTS.md.

---

✨ Key Features

• Single adapter factory – daplug_sql.adapter(**kwargs) returns a ready-to-go adapter configured for Postgres or MySQL based on the engine parameter.
• Optimistic CRUD – Identifier-aware insert, update, upsert, and delete guard against duplicates and emit SNS events automatically.
• Atomic upserts – upsert compiles to a single INSERT ... ON CONFLICT DO UPDATE (Postgres) or INSERT ... ON DUPLICATE KEY UPDATE (MySQL), safe under concurrent writers, with optional JSON deep-merge, key stripping, and an out-of-order guard column.
• JSON native – dict/list values are adapted automatically (psycopg2 Json on Postgres, json.dumps on MySQL), so JSONB/JSON columns just work.
• Connection reuse – Thread-safe cache reuses connections per endpoint/database/user/port/engine and lazily closes them.
• Integration-tested – pipenv run integration spins up both Postgres and MySQL via docker-compose and runs the real test suite.

---

🚀 Quick Start

Installation

pip install daplug-sql
# pipenv install daplug-sql
# poetry add daplug-sql
# uv pip install daplug-sql

Minimal Example

from daplug_sql import adapter

sql = adapter(
    endpoint="127.0.0.1",
    database="daplug",
    user="svc",
    password="secret",
    engine="postgres",  # "mysql" also supported
)

sql.connect()
sql.insert(
    data={"customer_id": "abc123", "name": "Ada"},
    table="customers",
    identifier="customer_id",
)
record = sql.get("abc123", table="customers", identifier="customer_id")
print(record)
sql.close()

---

⚙️ Configuration

Parameter	Type	Required	Description
endpoint	str	✅	Host/IP of the Postgres/MySQL server.
database	str	✅	Database/schema name.
user	str	✅	Database username.
password	str	✅	Database password.
engine	str	➖	'postgres' (default) or 'mysql'.
autocommit	bool	➖	Defaults to True; set False for manual transaction control.
sns_arn	str	➖	SNS topic ARN used when publishing CRUD events.
sns_endpoint	str	➖	Optional SNS endpoint URL (e.g., LocalStack).
sns_attributes	dict	➖	Default SNS message attributes merged into every publish.

Per-Call Options

Every CRUD/query helper expects the target table and identifier column at call time so one adapter can manage multiple tables:

Argument	Description
table	Table to operate on (customers, orders, etc.).
identifier	Column that uniquely identifies rows (customer_id).
commit	Override autocommit per call (True/False).
debug	Log SQL statements via the adapter logger when True.
sns_attributes	Per-call attributes merged with defaults before publish.
fifo_group_id / fifo_duplication_id	Optional FIFO metadata passed straight to SNS.
publish	Set to False to skip the SNS publish for this call only (default True).
publish_data	Replace the published payload entirely (the row write is unchanged).
merge	update only: set False to skip the read-and-merge and write the payload exactly as given (default True).
atomic	upsert only: set False to fall back to the legacy fetch-then-insert/update path (default True).
merge_columns	upsert only: list of JSON columns to deep-merge with the existing row instead of overwriting.
strip_paths	upsert only: {column: [dot.paths]} removed from the column after merge (e.g. prune stale keys).
guard_column	upsert only: column compared as incoming >= existing; stale rows are skipped and upsert returns None.

SNS Publishing

SQLAdapter inherits daplug-core's SNS publisher. Provide the topic details when constructing the adapter:

sql = adapter(
    endpoint="127.0.0.1",
    database="daplug",
    user="svc",
    password="secret",
    engine="postgres",
    sns_arn="arn:aws:sns:us-east-1:123456789012:sql-events",
    sns_endpoint="http://localhost:4566",  # optional (LocalStack)
    sns_attributes={"service": "billing"},
)

• sns_attributes passed to adapter(...) become defaults for every publish.
• Each CRUD helper accepts its own sns_attributes to overlay call-specific metadata.
• FIFO topics are supported via the fifo_group_id and fifo_duplication_id kwargs on individual calls.

Example:

sql.insert(
    data={"customer_id": "abc123", "name": "Ada"},
    table="customers",
    identifier="customer_id",
    sns_attributes={"event": "customer-created"},
    fifo_group_id="customers",
)

If sns_arn is omitted, publish calls are skipped automatically. To skip a single call while keeping defaults intact, pass publish=False. To publish a different payload than the row that was written, pass publish_data={...}.

sql.insert(data=row, table="customers", identifier="customer_id", publish=False)

sql.update(
    data=row,
    table="customers",
    identifier="customer_id",
    publish_data={"id": row["customer_id"], "event": "updated"},
)

---

🧭 Public API Cheat Sheet

Method	Description
connect()	Opens a connection + cursor using the engine-specific connector.
close()	Closes the cursor/connection and evicts the cached connector.
commit(commit=True)	Commits the underlying DB connection when commit is truthy.
insert(data, table, identifier, **kwargs)	Validates data, enforces uniqueness on the provided identifier, inserts the row, and publishes SNS.
update(data, table, identifier, **kwargs)	Fetches the existing row, merges via dict_merger (skip with merge=False), runs UPDATE, publishes SNS.
upsert(data, table, identifier, **kwargs)	Single atomic ON CONFLICT/ON DUPLICATE KEY write (default); supports merge_columns, strip_paths, guard_column. Returns the written row, or None when the guard rejects it. atomic=False restores the legacy fetch-then-write path.
get(identifier_value, table, identifier, **kwargs)	Returns the first matching row or None.
read(identifier_value, table, identifier, **kwargs)	Alias of get.
query(query, params, table, identifier, **kwargs)	Executes a read-only statement (SELECT) and returns all rows as dictionaries.
delete(identifier_value, table, identifier, **kwargs)	Deletes the row, publishes SNS, and ignores missing rows.
create_index(table_name, index_columns)	Issues CREATE INDEX index_col1_col2 ON table_name (col1, col2) using safe identifiers.
create_table(query, **kwargs)	Executes DDL that must start with CREATE TABLE; anything else raises CreateTableException.
install_json_merge(**kwargs)	Postgres only: installs the daplug_json_merge deep-merge function used by merge_columns (no-op on MySQL, which uses native JSON_MERGE_PATCH). Run once per database, e.g. in migrations.

All identifier-based helpers sanitize names with SAFE_IDENTIFIER to prevent SQL injection through table/column inputs.

---

📚 Usage Examples

Insert + Query (Postgres)

sql = adapter(
    endpoint="127.0.0.1",
    database="daplug",
    user="svc",
    password="secret",
    engine="postgres",
)
sql.connect()

sql.insert(data={"sku": "W-1000", "name": "Widget", "cost": 99}, table="inventory", identifier="sku")
rows = sql.query(
    query="SELECT sku, name FROM inventory WHERE cost >= %(min_cost)s",
    params={"min_cost": 50},
    table="inventory",
    identifier="sku",
)
print(rows)
sql.close()

Transactions (MySQL)

sql = adapter(
    endpoint="127.0.0.1",
    database="daplug",
    user="svc",
    password="secret",
    engine="mysql",
    autocommit=False,
)
sql.connect()

try:
    sql.insert(data={"order_id": "O-1", "status": "pending"}, table="orders", identifier="order_id", commit=False)
    sql.update(data={"order_id": "O-1", "status": "shipped"}, table="orders", identifier="order_id", commit=False)
    sql.commit(True)
finally:
    sql.close()

Per-call Table Overrides

# Share one adapter across multiple tables by overriding table + identifier per call
sql.insert(data=payload, table="orders", identifier="order_id")
sql.create_index("orders", ["status", "created_at"])

Atomic Upserts with JSON Merge (event projections)

Project events into one row per entity: merge payloads additively, prune stale keys, and skip out-of-order deliveries, all in a single atomic statement.

sql.create_table(
    query="CREATE TABLE IF NOT EXISTS business_workers ("
          " entity_key VARCHAR(64) PRIMARY KEY,"
          " payload JSONB,"
          " last_event_at BIGINT)"
)
sql.install_json_merge()  # once per Postgres database; MySQL uses native JSON_MERGE_PATCH

row = sql.upsert(
    data={"entity_key": "worker-123", "payload": event_payload, "last_event_at": occurred_at},
    table="business_workers",
    identifier="entity_key",
    merge_columns=["payload"],                       # deep-merge instead of overwrite
    strip_paths={"payload": ["eye_color", "preferences.music"]},  # prune stale keys
    guard_column="last_event_at",                    # ignore older events
)
if row is None:
    print("stale event skipped")

Engine notes: the row is returned via RETURNING * on Postgres and re-fetched on MySQL (JSON columns come back as strings there). MySQL deep-merge follows JSON_MERGE_PATCH semantics, so a JSON null removes its key; Postgres keeps it. The atomic path requires the identifier column to be the primary key or a unique index, and MySQL 8.0.19+ for the row-alias syntax.

---

🧪 Testing & Tooling

Command	Description
pipenv run lint	Runs pylint and exports HTML/JSON to coverage/lint.
pipenv run type-check	Runs mypy using the new Protocol types.
pipenv run test	Executes the unit suite (mocks only).
pipenv run integration	Starts Postgres + MySQL via docker-compose and runs tests/integration.
pipenv run test_ci	Runs unit tests and integration tests sequentially (no Docker management).
pipenv run coverage	Full coverage run producing HTML, XML, JUnit, and pretty reports.

Integration tests rely on tests/integration/docker-compose.yml. The CircleCI pipeline mirrors this by launching Postgres and MySQL sidecars, waiting for them to be reachable, and then executing pipenv run coverage so artifacts are published automatically.

---

🗂 Project Layout

daplug-sql/
├── daplug_sql/
│   ├── adapter.py           # SQLAdapter implementation
│   ├── exception.py         # Adapter-specific exceptions
│   ├── sql_connector.py     # Engine-aware connector wrapper
│   ├── sql_connection.py    # Connection caching decorators
│   ├── types/__init__.py    # Shared typing helpers (Protocols, aliases)
│   └── __init__.py          # Adapter factory export
├── tests/
│   ├── unit/                # Pure unit tests (mocks only)
│   └── integration/         # Integration tests (Postgres + MySQL)
├── tests/integration/docker-compose.yml
├── Pipfile / Pipfile.lock   # Runtime + dev dependencies
├── setup.py                 # Packaging metadata
├── README.md
└── .agents/AGENTS.md        # Automation/Triage playbook for agents

---

🤝 Contributing

1. Fork / branch (git checkout -b feature/amazing)
2. pipenv install --dev
3. Add/change code + tests
4. Run pipenv run lint && pipenv run type-check && pipenv run test && pipenv run integration
5. Open a pull request and tag @dual

---

📄 License

Apache License 2.0 – see LICENSE.

---

Built to keep SQL integrations event-driven and zero-boilerplate.