Вставка данных в PostgreSQL с функциями plpg sql с использованием sqlAlchemy и python

Question

Я новичок в базе данных и sqlAlchemy, возможно, это простой или глупый вопрос. Но я не смог найти никакого решения.

У меня есть существующий проект, и я пытаюсь изменить текущую реализацию для использования SQLAlchemy.

Моя проблема заключается в том, что код работает без ошибок и Я получаю уникальный идентификатор работы, который я вижу в возвращенных данных. Но у меня нет записи в базе данных. Поскольку я использую выполнение без установления соединения, я думаю, что могу вызывать функцию следующим образом.

Существующий. sql файл:

CREATE TYPE job_status_type AS ENUM (
    'SCHEDULED', 'SUCCESS', 'ERROR', 'CANCELLED', 'MISSED'
);

CREATE TABLE job (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES user(id),
    param_id TEXT REFERENCES param(id),
    proc_name TEXT,
    status job_status_type DEFAULT 'SCHEDULED',
    run_date TIMESTAMPTZ DEFAULT current_timestamp
);

CREATE OR REPLACE FUNCTION check_job_run_date() RETURNS TRIGGER AS $check_job_run_date$
BEGIN
    if NEW.run_date < (current_timestamp - INTERVAL '1 second') THEN
        RAISE 'job_run_date_in_past';
    END IF;
    RETURN NEW;
END;
$check_job_run_date$ LANGUAGE plpgsql;

CREATE TRIGGER check_job_run_date BEFORE INSERT ON job
    FOR EACH ROW EXECUTE PROCEDURE check_job_run_date();


CREATE OR REPLACE FUNCTION add_job(INTEGER, TEXT, TEXT, TIMESTAMPTZ) RETURNS SETOF job AS $$
DECLARE
    new_id INTEGER;
BEGIN
    INSERT INTO job (user_id, param_id, proc_name, run_date) 
    VALUES ($1, $2, $3, $4) RETURNING id INTO new_id;

    RETURN QUERY SELECT * FROM job WHERE id = new_id;
END;

Старая реализация, была успешно запущена:

def insert_job(db, user_id: int, param_id: str, proc_name: str, run_date: datetime) -> Job:
    try:
        db.execute('SELECT * FROM add_job(%s, %s, %s, %s)', (user_id, param_id, proc_name, run_date))
    except RaiseException as e:
        raise

    db.commit()
    job = db.fetchone()
    return job

Новая реализация:

def insert_job(db, user_id: int, param_id: str, proc_name: str, run_date: datetime) -> Job:
    query = select([func.add_job(user_id, param_id, proc_name, run_date)])
    try:
        result = engine.execute(query).fetchone()
    except RaiseException as e:
        raise
    return result

Answer 1

Прошло много времени с тех пор, как я использовал SQLAlchemy, но в тот день вам нужно было выдать коммит для транзакции, которая будет зафиксирована на стороне БД. Это, безусловно, будет иметь симптом, который вы видите, если это проблема. Попробуйте изменить новую реализацию на:

def insert_job(db, user_id: int, param_id: str, proc_name: str, run_date: datetime) -> Job:
    query = select([func.add_job(user_id, param_id, proc_name, run_date)])
    with engine.begin() as conn:
        result = conn.execute(query).fetchone()
    return result

(блок with просто устанавливает транзакцию и фиксирует ее, когда блок завершается без исключения)