Bridge Adapters¶

Insert Awa jobs within existing transactions from non-sqlx Postgres libraries.

The core awa::insert and awa::insert_with functions require sqlx's PgExecutor trait. Bridge adapters let users of other libraries enqueue jobs without depending on sqlx directly. All adapters share the same preparation logic (validation, state determination, unique key computation) as the sqlx path — no semantic drift between drivers.

Rust: external adapter API¶

External Rust integration crates can reuse Awa's canonical insert preparation without depending on sqlx executors. The stable surface is awa::adapter::postgres:

use awa::adapter::postgres::{
    prepare_job_insert, INSERT_JOB_SQL, UNIQUE_VIOLATION_SQLSTATE,
};
use awa::InsertOpts;

let prepared = prepare_job_insert(&args, InsertOpts::default())?;

// Execute INSERT_JOB_SQL with your driver's transaction or connection type.
// Bind values in this order:
// 1. prepared.kind()
// 2. prepared.queue()
// 3. prepared.args()
// 4. prepared.state_db_str()
// 5. prepared.priority()
// 6. prepared.max_attempts()
// 7. prepared.run_at()
// 8. prepared.metadata()
// 9. prepared.tags()
// 10. prepared.unique_key()
// 11. prepared.unique_states_bit_string()
// 12. prepared.ordering_key()

prepare_job_insert and prepare_raw_job_insert apply the same validation, scheduled-state selection, unique-key computation, unique-state bitmask formatting, and sharded enqueue ordering-key propagation as Awa's built-in sqlx insert path. Adapters should map Postgres SQLSTATE UNIQUE_VIOLATION_SQLSTATE to AwaError::UniqueConflict.

This API is intentionally single-row. High-throughput bulk ingestion remains on Awa's native SQLx-backed APIs (insert_many_copy_from_pool) and the queue-storage-native COPY path (QueueStorage::enqueue_params_copy), because COPY support, row return semantics, and uniqueness handling are driver-specific.

Worker polling, heartbeating, claiming, and completion remain on the Awa runtime. That does not prevent applications from using their existing database stack inside job handlers: pass a SeaORM connection, Diesel pool, or other app dependency through Client::builder(...).state(...) and extract it from JobContext. Integration crates can provide ergonomic helpers for this handler dependency wiring, but should not reimplement Awa's lease/runtime storage engine.

Rust: tokio-postgres¶

Dependencies¶

[dependencies]
awa = { version = "0.6", features = ["tokio-postgres"] }
tokio-postgres = { version = "0.7", features = ["with-chrono-0_4", "with-serde_json-1", "with-uuid-1"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

Basic insert¶

use awa::bridge::tokio_pg;
use awa::JobArgs;
use serde::{Deserialize, Serialize};
use tokio_postgres::NoTls;

#[derive(Debug, Serialize, Deserialize, JobArgs)]
struct SendEmail {
    to: String,
    subject: String,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (client, connection) =
        tokio_postgres::connect("postgres://localhost/mydb", NoTls).await?;
    tokio::spawn(connection);

    let job = tokio_pg::insert_job(
        &client,
        &SendEmail { to: "alice@example.com".into(), subject: "Welcome".into() },
    ).await?;

    println!("inserted job {} (kind={}, state={:?})", job.id, job.kind, job.state);
    Ok(())
}

Transactional enqueue¶

The primary use case: insert app data and an Awa job atomically in the same transaction. If the transaction rolls back, both the app row and the job disappear.

use awa::bridge::tokio_pg;
use awa::InsertOpts;

let mut client = client; // from connect()
let txn = client.transaction().await?;

// App logic
txn.execute(
    "INSERT INTO orders (id, total) VALUES ($1, $2)",
    &[&order_id, &total],
).await?;

// Awa job in the same transaction
let job = tokio_pg::insert_job_with(
    &txn,
    &SendEmail { to: "alice@example.com".into(), subject: "Order confirmed".into() },
    InsertOpts {
        queue: "email".into(),
        priority: 1,
        ..Default::default()
    },
).await?;

txn.commit().await?;
// Both the order and the job are now visible. Rollback would discard both.

Supported types¶

insert_job and insert_job_with accept any C: tokio_postgres::GenericClient. This trait is implemented for:

tokio_postgres::Client — direct connection
tokio_postgres::Transaction<'_> — active transaction

Pool wrappers like deadpool_postgres::Client or bb8::PooledConnection typically Deref to tokio_postgres::Client but do not implement GenericClient directly. To use them, call .transaction() on the wrapper and pass the resulting tokio_postgres::Transaction:

// deadpool-postgres
let mut pool_client = pool.get().await?;
let txn = pool_client.transaction().await?;
tokio_pg::insert_job(&txn, &args).await?;
txn.commit().await?;

Raw insert¶

When you don't have a JobArgs impl (e.g. forwarding from a dynamic source):

let job = tokio_pg::insert_job_raw(
    &txn,
    "send_email".into(),
    serde_json::json!({"to": "alice@example.com", "subject": "Welcome"}),
    InsertOpts::default(),
).await?;

Return value¶

All functions return awa::JobRow with the full row from RETURNING * — same type as awa::insert_with. The only field not populated is unique_states (BIT(8), no direct tokio-postgres mapping). All other fields, including errors, are decoded from the database row.

Rust: SeaORM¶

SeaORM already sits on top of SQLx, so the integration is deliberately thin. A sea_orm::DatabaseConnection wraps a sqlx::PgPool, so for building a client, running migrations, or reading job state you can reach the pool directly (awa_seaorm::pool(&db) / db.awa_pool()) and use Awa's existing APIs unchanged.

The part that needs a real adapter is transactional enqueue. get_postgres_connection_pool() hands back a separate pooled connection, so a job inserted through it commits independently of your ORM writes. The insert / insert_with / insert_raw helpers instead run Awa's canonical insert SQL through SeaORM's ConnectionTrait, so they bind to whatever you pass — a DatabaseConnection or a DatabaseTransaction — letting a job commit atomically with the rest of a transaction.

The adapter lives in the optional awa-seaorm crate:

[dependencies]
awa = "0.6"
awa-seaorm = "0.6"
sea-orm = { version = "=2.0.0-rc.38", default-features = false, features = [
    "sqlx-postgres",
    "runtime-tokio-rustls",
] }

use awa::JobArgs;
use awa_seaorm::{insert, migrate};
use sea_orm::{Database, TransactionTrait};
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize, JobArgs)]
struct SendWelcomeEmail {
    user_id: i64,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let db = Database::connect(&std::env::var("DATABASE_URL")?).await?;
    migrate(&db).await?;

    // The user row and its welcome-email job commit together, or not at all.
    let txn = db.begin().await?;
    // ... insert your application rows via SeaORM on `txn` ...
    insert(&txn, &SendWelcomeEmail { user_id: 42 }).await?;
    txn.commit().await?;

    Ok(())
}

Without a transaction, pass the connection directly (insert(&db, &job)) to enqueue immediately, or use client_builder(&db) to build a worker client.

Why the adapter runs SQL rather than reusing `awa::insert_with`¶

Awa's native insert_with is a pull API: hand it a &mut sqlx::PgConnection and it runs on it. A SeaORM DatabaseTransaction can't satisfy that. It owns the connection inside an Arc<Mutex<…>> and must keep it so a later commit()/rollback() runs on that same connection — so it never lends out a &mut PgConnection, and exposes no Into/AsMut/accessor for one. The only way it lets you use that connection is its ConnectionTrait, a push API: you give it a Statement and it runs it for you.

So the adapter inverts the direction — it pushes Awa's canonical insert SQL (awa::adapter::postgres::INSERT_JOB_SQL, the same statement and bind order the native path and the tokio-postgres bridge use) through ConnectionTrait. The result row, however, is surrendered: QueryResult::try_as_pg_row yields the underlying sqlx::PgRow, which Awa's own JobRow: FromRow decodes — so the returned job is identical to one from awa::insert, with no parallel decoder to drift. This is also why the workspace enables SeaORM's with-chrono and with-json features: only so the timestamp and JSONB bind values can become sea_orm::Value.

Python: psycopg3, asyncpg, SQLAlchemy, Django¶

See Python getting started — ORM Transaction Bridging.

Rust feature flags¶

Feature	Crate	What it enables
`tokio-postgres`	`awa` or `awa-model`	`awa::bridge::tokio_pg` adapter
`cel`	`awa` or `awa-model`	CEL expression evaluation for callback filtering
`anyhow`	`awa`	`From<anyhow::Error>` for `JobError`