How to implement a conditional insert (MySQL)

[jens.t...@gmail.com]

Hi,

My question is based on this answer on Stackoverflow to the question MySQL Conditional Insert. In my current SA implementation I have the following code:

token = Token(                                                                

    user=user,                                                                            

    client_sig=sigstring,                                                                        

    )                                                                                           

session.add(token)                                                                        

# session.flush()                                                                               

Now I would like to prevent duplicate client_sig entries to prevent the same user having the same client_sig, but I can’t set that column to unique. It looks like the following SQL statement achieves what I want:

INSERT INTO tokens(id, user_id, client_sig)

SELECT '51…bb', 'fd…b3', 'some string'

FROM dual

WHERE NOT EXISTS (

  SELECT * 

  FROM tokens

  WHERE user_id='fd…b3'

  AND client_sig='some string'

  );

I found some documentation on insert … from select but can not quite put the two together (I don't think they're the same). How would I go about implementing the above SQL statement using SA, or will I have to issue raw SQL in this case?

Thank you!

Jens

[Mike Bayer]

On Fri, Nov 24, 2017 at 6:32 PM, <jens.t...@gmail.com> wrote:
> Hi,
>
> My question is based on this answer on Stackoverflow to the question MySQL
> Conditional Insert. In my current SA implementation I have the following
> code:
>
> token = Token(
> user=user,
> client_sig=sigstring,
> )
> session.add(token)
> # session.flush()
>
> Now I would like to prevent duplicate client_sig entries to prevent the same
> user having the same client_sig, but I can’t set that column to unique.

that's the first red flag here, why can't you put a unique constraint here?

It
> looks like the following SQL statement achieves what I want:
>
> INSERT INTO tokens(id, user_id, client_sig)
> SELECT '51…bb', 'fd…b3', 'some string'
> FROM dual
> WHERE NOT EXISTS (
> SELECT *
> FROM tokens
> WHERE user_id='fd…b3'
> AND client_sig='some string'
> );
>
> I found some documentation on insert … from select but can not quite put the
> two together (I don't think they're the same).

I doubt MySQL requires the "Dual" part, here is a demo

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class Token(Base):
__tablename__ = 'tokens'
id = Column(String(50), primary_key=True)
user_id = Column(String(50))
client_sig = Column(String(50))

e = create_engine("mysql://scott:tiger@localhost/test", echo=True)
Base.metadata.drop_all(e)
Base.metadata.create_all(e)

stmt = select([
literal_column("'abc'"),
literal_column("'def'"),
literal_column("'ghi'"),
]).where(
~exists().where(and_(Token.user_id == 'def', Token.client_sig == 'ghi'))
)

e.execute(
insert(Token).from_select(['id', 'user_id', 'client_sig'], stmt)
)

How would I go about
> implementing the above SQL statement using SA, or will I have to issue raw
> SQL in this case?
>
> Thank you!
> Jens
>

> --
> SQLAlchemy -
> The Python SQL Toolkit and Object Relational Mapper
>
> http://www.sqlalchemy.org/
>
> To post example code, please provide an MCVE: Minimal, Complete, and
> Verifiable Example. See http://stackoverflow.com/help/mcve for a full
> description.
> ---
> You received this message because you are subscribed to the Google Groups
> "sqlalchemy" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to sqlalchemy+...@googlegroups.com.
> To post to this group, send email to sqlal...@googlegroups.com.
> Visit this group at https://groups.google.com/group/sqlalchemy.
> For more options, visit https://groups.google.com/d/optout.

[jens.t...@gmail.com]

Thanks Mike!

that's the first red flag here, why can't you put a unique constraint here?

Ordinarily I’d agree. In this case, there’s an additional column called “deleted” which is NULL for active Tokens and contains utcnow() for deleted Tokens. That makes for deleted and active tokens which can have the same user_id/client_sig (one active, many deleted).

I doubt MySQL requires the "Dual" part, here is a demo

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Token(Base):
    __tablename__ = 'tokens'
    id = Column(String(50), primary_key=True)
    user_id = Column(String(50))
    client_sig = Column(String(50))

e = create_engine("mysql://scott:tiger@localhost/test", echo=True)
Base.metadata.drop_all(e)
Base.metadata.create_all(e)

stmt = select([
    literal_column("'abc'"),
    literal_column("'def'"),
    literal_column("'ghi'"),
]).where(
    ~exists().where(and_(Token.user_id == 'def', Token.client_sig == 'ghi'))
)

e.execute(
    insert(Token).from_select(['id', 'user_id', 'client_sig'], stmt)
) 

 Ah ok, so from_select() would indeed be the correct approach here. Thank you, I’ll tinker with this :-)

One more question in this context though (without having tried the above yet). The INSERT fails if the SELECT returns an existing row (i.e. an active Token exists already in my example above). I suspect that the failure happens upon commit() of the transaction, not upon flush()?

Because the transaction and session are bound to a Pyramid request and the commit happens outside of the view handler function, I’ll catch that failure in a generic exception view which I’ll have to register?

Cheers,

Jens

[Mike Bayer]

flush() has no relationship to an INSERT..SELECT construct as above.

I"m not sure what "fails" means here, are you saying the INSERT
actually raises an error if the SELECT returns no rows (I've never
tried this, so not sure) ? if that is the case it would happen when
you invoke the statement. when some kind of error happens upon
COMMIT that is usually within a narrow range of deadlock conditions
that for MySQL usually occur for a certain kind of DELETE scenario,
for an INSERT I've only seen fail-on-commit as a behavior of Galera
when a writeset conflicts with another node.

>
> Because the transaction and session are bound to a Pyramid request and the
> commit happens outside of the view handler function, I’ll catch that failure
> in a generic exception view which I’ll have to register?

I'd need to see more detail about this failure mode since I haven't
tried doing an empty insert...select on MySQL.

>
> Cheers,

[jens.t...@gmail.com]

Thank you, Mike!

I’ve tried to implement what we talked about above, but had a few problems. First, I was unable to use MySQLdb because of compilation errors; instead I’ve used pymysql for a while now. When I tried to run the test code from your previous answer I received a SQL syntax error near WHERE. It looks like the FROM is missing.

As for “fails” I was mistaken. The conditional insert should always succeed, but may not insert a row into the table. While this maintains data consistency, it’s not quite what I need. It’ll do for now though.

My current working solution is the following hack which I need to revisit again later:

# Create a proxy object to return to the caller.

token = Token(
    id=uuid4(),

    user=user_id,

    client_sig=sigstring,

    )

# If I don't expunge then the token is automatically added/inserted?

# I thought I'd always have to dbsession.add().

dbsession.expunge(token)

# Now issue the SQL statement much like the on from my initial question.

sql = (

    "INSERT INTO tokens(id, user_id, client_sig) "                                       

    "SELECT '" + token.id + "', '" + token.user_id + "', '" + token.client_sig + "' "

    "FROM dual "                                                                             

    "WHERE NOT EXISTS "                                                                     

    "(SELECT * FROM tokens WHERE user_id='" + token.user_id + "' AND client_sig='" + token.client_sig + "')"

    )

dbsession.execute(sql)

# Return the proxy object to the Pyramid view function.

return token

I realize that this is a poor solution because the returned proxy object may in rare cases not represent what's committed to the db. However, that's easier to handle than committing a duplicate. I would also prefer SQLA functions over raw SQL code.

Jens

[Mike Bayer]

On Sun, Nov 26, 2017 at 9:39 PM, <jens.t...@gmail.com> wrote:
> Thank you, Mike!
>
> I’ve tried to implement what we talked about above, but had a few problems.
> First, I was unable to use MySQLdb because of compilation errors; instead
> I’ve used pymysql for a while now. When I tried to run the test code from
> your previous answer I received a SQL syntax error near WHERE. It looks like
> the FROM is missing.

OK I don't recall what conditions MySQL needs "FROM dual" can you share with me:

1. the full version of MySQL

2. the output of "SHOW VARIABLES LIKE '%SQL_MODE%'

3. stack trace + error message

[jens.t...@gmail.com]

No problem, here it is. To work with your initial code example...

>>> e = create_engine("mysql+pymysql://jens@localhost/test?charset=utf8&unix_socket=/opt/local/var/run/mysql56/mysqld.sock", echo=True)

>>> Base.metadata.drop_all(e) 

2017-11-28 06:47:21,171 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'sql_mode'

2017-11-28 06:47:21,171 INFO sqlalchemy.engine.base.Engine {}

2017-11-28 06:47:21,173 INFO sqlalchemy.engine.base.Engine SELECT DATABASE()

2017-11-28 06:47:21,173 INFO sqlalchemy.engine.base.Engine {}

2017-11-28 06:47:21,174 INFO sqlalchemy.engine.base.Engine show collation where `Charset` = 'utf8' and `Collation` = 'utf8_bin'

2017-11-28 06:47:21,174 INFO sqlalchemy.engine.base.Engine {}

2017-11-28 06:47:21,175 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS CHAR(60)) AS anon_1

2017-11-28 06:47:21,175 INFO sqlalchemy.engine.base.Engine {}

2017-11-28 06:47:21,176 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS CHAR(60)) AS anon_1

2017-11-28 06:47:21,176 INFO sqlalchemy.engine.base.Engine {}

2017-11-28 06:47:21,176 INFO sqlalchemy.engine.base.Engine SELECT CAST('test collated returns' AS CHAR CHARACTER SET utf8) COLLATE utf8_bin AS anon_1

2017-11-28 06:47:21,176 INFO sqlalchemy.engine.base.Engine {}

2017-11-28 06:47:21,177 INFO sqlalchemy.engine.base.Engine DESCRIBE `tokens`

2017-11-28 06:47:21,177 INFO sqlalchemy.engine.base.Engine {}

2017-11-28 06:47:21,179 INFO sqlalchemy.engine.base.Engine ROLLBACK

>>> Base.metadata.create_all(e) 

2017-11-28 06:47:23,490 INFO sqlalchemy.engine.base.Engine DESCRIBE `tokens`

2017-11-28 06:47:23,490 INFO sqlalchemy.engine.base.Engine {}

2017-11-28 06:47:23,495 INFO sqlalchemy.engine.base.Engine ROLLBACK

2017-11-28 06:47:23,496 INFO sqlalchemy.engine.base.Engine 

CREATE TABLE tokens (

id VARCHAR(50) NOT NULL, 

user_id VARCHAR(50), 

client_sig VARCHAR(50), 

PRIMARY KEY (id)

)

2017-11-28 06:47:23,496 INFO sqlalchemy.engine.base.Engine {}

2017-11-28 06:47:23,507 INFO sqlalchemy.engine.base.Engine COMMIT

>>> stmt = select([ 

...     literal_column("'abc'"), 

...     literal_column("'def'"), 

...     literal_column("'ghi'"), 

... ]).where( 

...     ~exists().where(and_(Token.user_id == 'def', Token.client_sig == 'ghi')) 

... ) 

>>> 

>>> print(stmt)

SELECT 'abc', 'def', 'ghi' 

WHERE NOT (EXISTS (SELECT * 

FROM tokens 

WHERE tokens.user_id = :user_id_1 AND tokens.client_sig = :client_sig_1))

OK I don't recall what conditions MySQL needs "FROM dual" can you share with me:

1. the full version of MySQL

 

>>> e.dialect.name, e.dialect.driver, e.dialect.server_version_info

('mysql', 'pymysql', (5, 6, 34))

2. the output of "SHOW VARIABLES LIKE '%SQL_MODE%'

mysql> show variables like '%sql_mode%';

+---------------+------------------------+

| Variable_name | Value                  |

+---------------+------------------------+

| sql_mode      | NO_ENGINE_SUBSTITUTION |

+---------------+------------------------+

1 row in set (0.00 sec)

3. stack trace + error message

>>> e.execute( 

...     insert(Token).from_select(['id', 'user_id', 'client_sig'], stmt) 

... ) 

2017-11-28 06:47:49,489 INFO sqlalchemy.engine.base.Engine INSERT INTO tokens (id, user_id, client_sig) SELECT 'abc', 'def', 'ghi' 

WHERE NOT (EXISTS (SELECT * 

FROM tokens 

WHERE tokens.user_id = %(user_id_1)s AND tokens.client_sig = %(client_sig_1)s))

2017-11-28 06:47:49,489 INFO sqlalchemy.engine.base.Engine {'client_sig_1': 'ghi', 'user_id_1': 'def'}

2017-11-28 06:47:49,491 INFO sqlalchemy.engine.base.Engine ROLLBACK

Traceback (most recent call last):

  File "/…/lib/python3.5/site-packages/sqlalchemy/engine/base.py", line 1182, in _execute_context

    context)

  File "/…/lib/python3.5/site-packages/sqlalchemy/engine/default.py", line 470, in do_execute

    cursor.execute(statement, parameters)

  File "/…/lib/python3.5/site-packages/pymysql/cursors.py", line 166, in execute

    result = self._query(query)

  File "/…/lib/python3.5/site-packages/pymysql/cursors.py", line 322, in _query

    conn.query(q)

  File "/…/lib/python3.5/site-packages/pymysql/connections.py", line 856, in query

    self._affected_rows = self._read_query_result(unbuffered=unbuffered)

  File "/…/lib/python3.5/site-packages/pymysql/connections.py", line 1057, in _read_query_result

    result.read()

  File "/…/lib/python3.5/site-packages/pymysql/connections.py", line 1340, in read

    first_packet = self.connection._read_packet()

  File "/…/lib/python3.5/site-packages/pymysql/connections.py", line 1014, in _read_packet

    packet.check_error()

  File "/…/lib/python3.5/site-packages/pymysql/connections.py", line 393, in check_error

    err.raise_mysql_exception(self._data)

  File "/…/lib/python3.5/site-packages/pymysql/err.py", line 107, in raise_mysql_exception

    raise errorclass(errno, errval)

pymysql.err.ProgrammingError: (1064, "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'WHERE NOT (EXISTS (SELECT * \nFROM tokens \nWHERE tokens.user_id = 'def' AND token' at line 2")

The above exception was the direct cause of the following exception:

Traceback (most recent call last):

  File "<stdin>", line 2, in <module>

  File "/…/lib/python3.5/site-packages/sqlalchemy/engine/base.py", line 2064, in execute

    return connection.execute(statement, *multiparams, **params)

  File "/…/lib/python3.5/site-packages/sqlalchemy/engine/base.py", line 945, in execute

    return meth(self, multiparams, params)

  File "/…/lib/python3.5/site-packages/sqlalchemy/sql/elements.py", line 263, in _execute_on_connection

    return connection._execute_clauseelement(self, multiparams, params)

  File "/…/lib/python3.5/site-packages/sqlalchemy/engine/base.py", line 1053, in _execute_clauseelement

    compiled_sql, distilled_params

  File "/…/lib/python3.5/site-packages/sqlalchemy/engine/base.py", line 1189, in _execute_context

    context)

  File "/…/lib/python3.5/site-packages/sqlalchemy/engine/base.py", line 1402, in _handle_dbapi_exception

    exc_info

  File "/…/lib/python3.5/site-packages/sqlalchemy/util/compat.py", line 203, in raise_from_cause

    reraise(type(exception), exception, tb=exc_tb, cause=cause)

  File "/…/lib/python3.5/site-packages/sqlalchemy/util/compat.py", line 186, in reraise

    raise value.with_traceback(tb)

  File "/…/lib/python3.5/site-packages/sqlalchemy/engine/base.py", line 1182, in _execute_context

    context)

  File "/…/lib/python3.5/site-packages/sqlalchemy/engine/default.py", line 470, in do_execute

    cursor.execute(statement, parameters)

  File "/…/lib/python3.5/site-packages/pymysql/cursors.py", line 166, in execute

    result = self._query(query)

  File "/…/lib/python3.5/site-packages/pymysql/cursors.py", line 322, in _query

    conn.query(q)

  File "/…/lib/python3.5/site-packages/pymysql/connections.py", line 856, in query

    self._affected_rows = self._read_query_result(unbuffered=unbuffered)

  File "/…/lib/python3.5/site-packages/pymysql/connections.py", line 1057, in _read_query_result

    result.read()

  File "/…/lib/python3.5/site-packages/pymysql/connections.py", line 1340, in read

    first_packet = self.connection._read_packet()

  File "/…/lib/python3.5/site-packages/pymysql/connections.py", line 1014, in _read_packet

    packet.check_error()

  File "/…/lib/python3.5/site-packages/pymysql/connections.py", line 393, in check_error

    err.raise_mysql_exception(self._data)

  File "/…/lib/python3.5/site-packages/pymysql/err.py", line 107, in raise_mysql_exception

    raise errorclass(errno, errval)

sqlalchemy.exc.ProgrammingError: (pymysql.err.ProgrammingError) (1064, "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'WHERE NOT (EXISTS (SELECT * \nFROM tokens \nWHERE tokens.user_id = 'def' AND token' at line 2") [SQL: "INSERT INTO tokens (id, user_id, client_sig) SELECT 'abc', 'def', 'ghi' \nWHERE NOT (EXISTS (SELECT * \nFROM tokens \nWHERE tokens.user_id = %(user_id_1)s AND tokens.client_sig = %(client_sig_1)s))"] [parameters: {'client_sig_1': 'ghi', 'user_id_1': 'def'}]

 Please let me know of anything else you need from me!

Jens

[Mike Bayer]

Well in standard MySQL / MariaDB fashion, they have made this as fun
as possible (well, more fun would be one database *rejects* FROM DUAL,
at least it isn't that bad):

1. MySQL 5.6 requires "FROM DUAL"

2. MySQL 5.7 does not require FROM DUAL but accepts it.

3. MariaDB 10.1 does not require FROM DUAL but accepts it.

4. MariaDB 10.2 *does* require FROM DUAL. The two vendors have
**flip-flopped** on their preference of this issue.

Anyway, here's your dual, as is typical, to make it work completely we
need an esoteric trick to avoid quoting the "dual" word:

from sqlalchemy.sql import quoted_name
dual = table(quoted_name("dual", quote=False))

then your statement:

stmt = select([
literal_column("'abc'"),
literal_column("'def'"),
literal_column("'ghi'"),

]).select_from(dual).where(

~exists().where(and_(Token.user_id == 'def', Token.client_sig == 'ghi'))

)

[jens.t...@gmail.com]

Hah 🤔 Boy this is (not really) funny. Thank you for digging into this, Mike!

I had to make two minor import adjustments

from sqlalchemy.sql.elements import quoted_name                                                     

from sqlalchemy.sql.expression import literal_column                                                

but otherwise the code works now. I still think it’s a poor solution to my problem. The proper solution, I think, would be a CHECK constraint across the columns and a simple INSERT which can fail the constraint.

Alas, MySQL doesn’t do CHECK constraints—another reason to migrate to PostgreSQL as soon as possible. 

Jens

[Mike Bayer]

On Nov 28, 2017 12:13 AM, <jens.t...@gmail.com> wrote:

Hah 🤔 Boy this is (not really) funny. Thank you for digging into this, Mike!

I had to make two minor import adjustments

from sqlalchemy.sql.elements import quoted_name                                                     

from sqlalchemy.sql.expression import literal_column                                                

but otherwise the code works now. I still think it’s a poor solution to my problem. The proper solution, I think, would be a CHECK constraint across the columns and a simple INSERT which can fail the constraint.

Alas, MySQL doesn’t do CHECK constraints—another reason to migrate to PostgreSQL as soon as possible. 

MariaDB 10.2 does 

To unsubscribe from this group and stop receiving emails from it, send an email to sqlalchemy+unsubscribe@googlegroups.com.

[jens.t...@gmail.com]

I am now thoroughly confused.

My understanding of the above conditional insert statement was that it won’t persist a token if there is already a token with the same user_id and client_sig in the table. Alas, today once again I see an exception “MultipleResultsFound: Multiple rows were found for one_or_none()” and two token were in the table. To requests arrived 70ms apart and it seems that both were successful in creating tokens.

I expected that one would “outrace” the other and one would succeed to persist the token which the other would see when its insert runs.

What am I missing here?

[Mike Bayer]

You would need to use serializable isolation or table locks to ensure
two conflicting INSERT operations don't overlap, if your operation
depends upon SELECTing those rows after the fact.

[jens.t...@gmail.com]

Thank you!

I found this elderly thread:  https://groups.google.com/forum/?fromgroups=#!topic/sqlalchemy/8WLhbsp2nls

If injecting the SQL statements directly is still the way to go, then I’ll wrap the conditional insert with a table lock.

Jens

[jens.t...@gmail.com]

Once again I'm at a loss.

I tried this:

    dbsession.execute("LOCK TABLES tokens WRITE")

    dbsession.execute(cond_insert) # As per above discussion.

    dbsession.execute("UNLOCK TABLES")

which raises an error: 1100, "Table 'tokens' was not locked with LOCK TABLES".

Then I read that the temporary table created by an inner select needs to be locked as well, but it doesn't exist at the time when LOCK TABLES executes. So using stmt.alias("foo") for the INSERT, and adding "foo" to the LOCK TABLES doesn't work. The "dual" table can't be locked either, as it doesn't actually exist. No other table is being used by the conditional INSERT as far as I can see.

I can't quite make sense of the error message, and how to resolve it.

[Mike Bayer]

is that session using "autocommit" mode? it looks like the error is
raised on the UNLOCK ?

[jens.t...@gmail.com]

is that session using "autocommit" mode?  

print("--> autocommit", dbsession.autocommit) gives a False.

 

it looks like the error is raised on the UNLOCK ? 

 When I comment out the UNLOCK, the exception still raises. Here is SQLA’s verbose logging:

--> autocommit False

2017-12-07 11:23:52,101 INFO  [sqlalchemy.engine.base.Engine][MainThread] LOCK TABLES tokens WRITE

2017-12-07 11:23:52,101 INFO  [sqlalchemy.engine.base.Engine][MainThread] {}

2017-12-07 11:23:52,102 INFO  [sqlalchemy.engine.base.Engine][MainThread] INSERT INTO tokens (id, user_id, client_sig) SELECT '84…20', '39…30', '0b…87' 

FROM dual 

WHERE NOT (EXISTS (SELECT * 

FROM tokens 

WHERE tokens.user_id = %(user_id_1)s AND tokens.client_sig = %(client_sig_1)s))

2017-12-07 11:23:52,102 INFO  [sqlalchemy.engine.base.Engine][MainThread] {'user_id_1': '39…30', 'client_sig_1': '0b…87'}

2017-12-07 11:23:52,103 ERROR [srv.views.exceptions][MainThread] Internal server error detected, stacktrace follows.

[Mike Bayer]

the answer is here:

https://stackoverflow.com/a/12943155/34549

so "Token" has to be aliased and both token and the alias need to be
locked individually, demo is below. But keep in mind, since you are
LOCKing the whole table, the whole "INSERT..SELECT" approach is
unnecessary, you could just SELECT first, get the result back, then
INSERT. Also make sure you unlock in a try/finally block.

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.sql import quoted_name
dual = table(quoted_name("dual", quote=False))

Base = declarative_base()


class Token(Base):
__tablename__ = 'tokens'
id = Column(String(50), primary_key=True)
user_id = Column(String(50))
client_sig = Column(String(50))

e = create_engine("mysql://scott:tiger@localhost/test", echo=True)
Base.metadata.drop_all(e)
Base.metadata.create_all(e)

ta = aliased(Token, name='ta')

stmt = select([
literal_column("'abc'"),
literal_column("'def'"),
literal_column("'ghi'"),
]).select_from(dual).where(

~exists().where(and_(ta.user_id == 'def', ta.client_sig == 'ghi'))
)


with e.begin() as conn:
conn.execute("LOCK TABLES tokens WRITE, tokens as ta WRITE")
try:
conn.execute(stmt)
conn.execute(

insert(Token).from_select(['id', 'user_id', 'client_sig'], stmt)

)
finally:
conn.execute("UNLOCK TABLES")