SQLAlchemy berichtet "Invalid utf8mb4 Zeichenfolge" für Binärspalte

Question

Unter der Annahme dieses MySQL Tabellenschema:

CREATE TABLE `user` (
    `id` int(11) NOT NULL AUTO_INCREMENT, 
    `uuid` binary(16) NOT NULL, 
    `email` varchar(255) NOT NULL, 
    `name` varchar(255) DEFAULT NULL, 
    `photo` binary(16) DEFAULT NULL, 
    PRIMARY KEY (`id`), 
    UNIQUE KEY `uuid` (`uuid`), 
    UNIQUE KEY `email` (`email`) 
) ENGINE=InnoDB AUTO_INCREMENT=8 DEFAULT CHARSET=utf8mb4; 

Wenn ich die execute() API von SQLAlchemy Verbindungsklasse als solche:

with self.engine.begin() as connection: 
    user_uuid = uuid.UUID("...") 
    result = connection.execute("SELECT email, name, photo FROM user WHERE uuid=%s", user_uuid.bytes) 

Wenn die UUID ist F393A167-A919-4B50-BBB7-4AD356E89E6B, dann druckt SQLAlchemy diese Warnung:

/site-packages/sqlalchemy/engine/def ault.py:450: Warnung: Ungültige utf8mb4 Zeichenfolge: ‚F393A1‘

Die uuid Spalte ist eine BINARY Spalte, also warum erwägt SQLAlchemy diesen Parametern einen Text ein, anstatt eine binäre Eins und wie dies zu verhindern ?

Answer 1

Die Erklärung und Lösung ist eigentlich in dieser bug report in MySQL:

ersetzen:

cursor.execute ("" " INSERT INTO user (UUID) VALUES (% s) """ , my_uuid)

mit

cursor.execute (""“ INSERT INTO user (UUID) VALUES (_binary% s) ""“, my_uuid)

Geist der Unterstrich. Es ist "_binary", nicht "binary". Dieses "_binary" teilt MySQL mit, dass die folgende Zeichenfolge als binär zu interpretieren ist und nicht als utf8 interpretiert/validiert werden soll.

Answer 2

Das Problem tritt auf Python 3 nicht, also denke ich, dass das Problem, dass die den Python 2 str Art gegeben zu unterscheiden, was sollte Datenbank-Treiber ist nicht in der Lage ist Byte sein.

Unabhängig davon scheint es mit SQLAlchemy Kern funktioniert direkt richtig, vermutlich weil es den Spaltentyp direkt kennt.

from sqlalchemy import MetaData, Table, select 

meta = MetaData() 
user = Table('user', meta, autoload_with=engine) 
result = select([user]).where(user.c.uuid == user_uuid.bytes) 

Wenn Sie einen String Ausführung fortsetzen möchten, können Sie zu gieße bytesarray wie SQLAlchemy zu tun scheint:

with self.engine.begin() as connection: 
    user_uuid = uuid.UUID("...") 
    result = connection.execute(
     "SELECT email, name, photo FROM user WHERE uuid=%s", 
     bytearray(user_uuid.bytes)) 

Oder SQLAlchemy zu sagen, welche Art der gebundene Parameter dies automatisch zu erhalten:

from sqlalchemy import text, bindparam, BINARY 

with self.engine.begin() as connection: 
    user_uuid = uuid.UUID("...") 
    stmt = text("SELECT email, name, photo FROM user WHERE uuid = :uuid") 
    stmt = stmt.bindparams(bindparam('uuid', user_uuid.bytes, type_=BINARY)) 
    result = connection.execute(stmt)