I'm getting a 'ghost' ROLLBACK from sqlalchemy but not when use psql with postgres

Question

I have a weird behaviour with postgres + sqlalchemy.

I call a function that insert into a table, but when called from sqlalchemy it roolback at the end, and when called from psql it succeed:

Logs when called by sqlalchemy (as reported by the logs):

Jan 21 13:17:28 intersec.local postgres[3466]: [18-9] STATEMENT:  SELECT name, suffix 
Jan 21 13:17:28 intersec.local postgres[3466]: [18-10]  FROM doc_codes('195536d95bd155b9ea412154b3e920761495681a')
Jan 21 13:17:28 intersec.local postgres[3466]: [19-9] STATEMENT:  ROLLBACK
Jan 21 13:17:28 intersec.local postgres[3465]: [13-9] STATEMENT:  COMMIT

If using psql:

Jan 21 13:28:47 intersec.local postgres[3561]: [20-9] STATEMENT:  SELECT name, suffix FROM doc_codes('195536d95bd155b9ea412154b3e920761495681a');

Note not transaction stuff at all.

This is my python code:

def getCon(self):
    conStr = "postgresql+psycopg2://%(USER)s:%(PASSWORD)s@%(HOST)s/%(NAME)s"
    config = settings.DATABASES['default']
    #print conStr % config
    con = sq.create_engine(
        conStr % config,
        echo=ECHO
    )

    event.listen(con, 'checkout', self.set_path)

    self.con = con
    self.meta.bind = con

    return con

def getDocPrefixes(self, deviceId):
    f = sq.sql.func.doc_codes(deviceId, type_=types.String)

    columns = [
        sq.Column('name', types.String),
        sq.Column('suffix', types.String)
    ]

    return [dict(x.items()) for x in self.con.execute
            (
                select(columns).
                select_from(f)
            ).fetchall()]

sync = dbSync('malab')
for k in sync.getDocPrefixes('195536d95bd155b9ea412154b3e920761495681a'):
    print k['name'], '=', k['suffix']

What could trigger the ROLLBACK?

P.D: My DB functions:

CREATE OR REPLACE FUNCTION next_letter (
  table_name TEXT,
  OUT RETURNS TEXT
)
AS
$$
DECLARE
    result TEXT = 'A';
    nextLetter TEXT;
    num INTEGER;
BEGIN
    SELECT INTO num nextval('letters');
    nextLetter := chr(num);
    result := nextLetter;

    WHILE true LOOP
        --RAISE NOTICE '%', result;
        IF EXISTS(SELECT 1 FROM DocPrefix WHERE Name=result AND TableName=table_name) THEN
            SELECT max(SUBSTRING(name FROM '\d+'))
                FROM DocPrefix WHERE Name=result AND TableName=table_name
                INTO num;

            result := nextLetter || (coalesce(num,0) + 1);
        ELSE
            EXIT;
        END IF;
    END LOOP;

    RETURNS = result;
END;
$$
LANGUAGE 'plpgsql';

-- Retorna el prefijo unico para la tabla/dispositivo.
CREATE OR REPLACE FUNCTION prefix_fordevice (
  table_name TEXT,
  device_id TEXT,
  OUT RETURNS TEXT
)
AS
$$
DECLARE
    result TEXT = NULL;
    row RECORD;
BEGIN
    IF NOT(EXISTS(SELECT 1 FROM DocPrefix WHERE MachineId=device_id AND TableName=table_name)) THEN
        INSERT INTO DocPrefix
            (Name, MachineId, TableName)
        VALUES
            (next_letter(table_name),  device_id, table_name);
    END IF;

    SELECT name FROM DocPrefix WHERE
        MachineId=device_id AND TableName=table_name
    INTO result;

    RETURNS =  result;
END;
$$
LANGUAGE 'plpgsql';

--Retornar los prefijos exclusivos para el ID de dispositvo
CREATE OR REPLACE FUNCTION doc_codes(device_id TEXT) RETURNS TABLE("name" TEXT, "suffix" TEXT) AS $$
    SELECT name, prefix_fordevice(name,  device_id) AS suffix FROM doccode;
$$ LANGUAGE SQL;

Answer 1

I finally found the answer here:

Make SQLAlchemy COMMIT instead of ROLLBACK after a SELECT query

def getDocPrefixes(self, deviceId):
    f = sq.sql.func.doc_codes(deviceId, type_=types.String)

    columns = [
        sq.Column('name', types.String),
        sq.Column('sufix', types.String)
    ]

    with self.con.begin():
        return [dict(x.items()) for x in self.con.execute
            (
                select(columns).
                select_from(f)
            ).fetchall()]

The thing is, the function can insert data + also return a SELECT, so, sqlalchemy think this is a normal SELECT when in fact the function also change data and need commit.

Answer 2

the antipattern here is that you're confusing a SQLAlchemy Engine for a connection, when you do something like this:

con = sq.create_engine(<url>)
result = con.execute(statement)

the Engine is associated with a connection pool as a source of connections. When you call the execute() method on Engine, it checks out a connection from the pool, runs the statement, and returns the results; when the result set is exhausted, it returns the connection to the pool. At that stage, the pool will either close the connection fully, or it will re-pool it. Storing the connection in the pool means that any remaining transactional state must be cleared (note that DBAPI connections are always in a transaction when they are used), so it emits a rollback.

Your program should create a single Engine per URL at the module level, and when it needs a connection, should call upon engine.connect().

the document Working with Engines and Connections explains all of this.