Multiple operations in the same transaction

[rafaelsq]

Hi, I'm trying to better understand the project and the best feature I can think of is being able to perform multiple operations in the same transaction. But I don't know if I misunderstood something or if this option doesn't exist natively.

As an example with pyscopg;
(in this case, the worker/consumer would run in another "app" just like mentioned in the documentation)

import json
from psycopg_pool import ConnectionPool
from procrastinate import jobs, manager

_pool = ConnectionPool(
    "postgres://{user}:{password}@{host}:{port}/{dbname}".format(),
    open=True,
)

# @app.task(queue="sums")
# def sum(a, b):
#     with open("myfile", "w") as f:
#         f.write(str(a + b))


with _pool.connection() as c:
    # _ = c.execute("select...")
    # [code]

    # just to be safe and don't have to write the query myself
    # I don't want to break my  app in case of an updated on procrastinate
    payload = manager.JobManager(None)._defer_job_query_kwargs(
        jobs.Job(
            queue="sums",
            task_name="sum",
            lock=None,
            queueing_lock=None,
            task_kwargs=dict(a="a", b="b"),
        )
    )

    payload["args"] = json.dumps(payload["args"])
    _ = c.execute(payload["query"], payload).fetchone()

    # _ = c.execute("insert into ....")
    # commit()
    

And pretty much the same with sqlalchemy;

import json
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from procrastinate import jobs, manager

engine = create_engine(
    "postgres://{user}:{password}@{host}:{port}/{dbname}".format(),
    echo=True,
)

tx = sessionmaker(autocommit=False, autoflush=False, bind=engine)


with tx() as c:
    # c.query(Something).all()
    # [code]

    payload = manager.JobManager(None)._defer_job_query_kwargs(
        jobs.Job(
            queue="sums",
            task_name="sum",
            lock=None,
            queueing_lock=None,
            task_kwargs=dict(a="a", b="b"),
        )
    )

    payload["args"] = json.dumps(payload["args"])
    _ = c.execute(payload["query"], payload).fetchone()

    # c.add(Something)
    # c.commit()

So my question is; Would there be a native way to do the same? If I misunderstood, could you show me an example of how this is could be made?

[ewjoachim]

Yes, I don't think there's a native way to do it.
I'm not opposed to adding a .defer_query() next to .defer() or .defer_async() (if anyone wants to tackle it, please open an issue first). I don't think there should be a way of passing your own connector to defer, as it would make typing much harder.