On 10/05/2016 12:38 AM, Jonathan Schultz wrote:
Many thanks for the prompt reply.
if you are getting that error on INSERT it sounds like there are driver
issues involved (e.g. pymssql, pyodbc). For an INSERT you should not
get that error; using the VARBINARY SQLAlchemy type, SQLAlchemy calls
upon a construct provided by your DBAPI (e.g., the driver) called
Binary(). This is a wrapper that is intended to signal to the driver
that this value is not a character string, it's bytes. Internally the
driver should be doing this step.
For here we'd look to see what driver you're using, if changing drivers
resolves, and as always a small and self-contained reproducing test
case. For a driver issue I'd then look to convert the test case to be
using the DB driver alone, then we send a bug report to that driver.
OK in brief I'm using pymssql. I've never used pyodbc but if it's worth
investigating whether it produces the same error I'll find the time to
figure out how to use it.
pymssql is overall a much better driver these days as it is actively
maintained, pyodbc seems like it isn't maintained very often.
However, for this one, pymssql is failing and pyodbc isn't, so I'd
advise reporting this upstream to the pymssql driver. See attached. As
a workaround, you can supply CONVERT yourself as part of the datatype:
from sqlalchemy import *
e = create_engine("mssql+pymssql://scott:tiger@192.168.122.135:1213",
echo=True)
class MyVarBinary(TypeDecorator):
impl = VARBINARY
def bind_expression(self, bindvalue):
return cast(bindvalue, VARBINARY)
m = MetaData()
t = Table(
'test', m,
Column('id', Integer, primary_key=True),
Column('x', MyVarBinary())
)
conn = e.connect()
trans = conn.begin()
t.create(conn)
conn.execute(
t.insert().values({'x':bindparam('x')}),
{'x': b'a'}
)
output:
INSERT INTO test (x) OUTPUT inserted.id VALUES (CAST(%(x)s AS
VARBINARY(max))
In the meantime I have produced a small test script that demonstrates
the problem. You need a database already created with something like:
CREATE DATABASE test
USE test
CREATE USER user FROM LOGIN user
GRANT CONTROL TO USER
CREATE TABLE Test(Id UNIQUEIDENTIFIER PRIMARY KEY, Test
VARBINARY(MAX) not null)
Then if you run the attached script testvarbinary.py with a single
argument that references the database, for example:
.\testvarbinary.py mssql+pymssql://user:password@localhost/test
you'll get the following output:
Test 1 works
Test 2 fails
Traceback (most recent call last):
File "./testvarbinary.py", line 25, in <module>
{'Test':bytearray(b'a')})
<snip>
File "pymssql.pyx", line 467, in pymssql.Cursor.execute
(pymssql.c:7561)
sqlalchemy.exc.OperationalError: (pymssql.OperationalError) (257,
'Implicit conversion from data type varchar to varbinary(max) is not
allowed. Use the CONVERT function to run this query.DB-Lib error
message 20018, severity 16:\nGeneral SQL Server error: Check
messages from the SQL Server\n') [SQL: 'INSERT INTO [Test] ([Test])
OUTPUT inserted.[Id] VALUES (%(Test)s)'] [parameters: {'Test': 'a'}]
What I find curious is that although I have tried to cast the value to
insert as a bytearray, it seems to have been cast back to a string in
the error message. However in some other cases where I obtain a value
via SQLAlchemy from a BLOB column in an SQLite DB, which appears to be a
byte array, I am able to insert it into an MSSQL VARBINARY(MAX) column
without calling CONVERT.
Hope that's of some use in tracking down any issue, or perhaps you can
see where I am going wrong.
Thanks again... Jonathan
--
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+unsubscr...@googlegroups.com
<mailto:sqlalchemy+unsubscr...@googlegroups.com>.
To post to this group, send email to sqlalchemy@googlegroups.com
<mailto:sqlalchemy@googlegroups.com>.
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
--
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+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
def run_test(dbapi, conn):
cursor = conn.cursor()
cursor.execute(
"CREATE TABLE t(id INTEGER PRIMARY KEY, x VARBINARY(MAX) not null)"
)
try:
if dbapi.paramstyle == 'qmark':
cursor.execute(
"insert into t (id, x) values (?, ?)",
(1, dbapi.Binary(b'a'))
)
else:
cursor.execute(
"insert into t (id, x) values (%(id)s, %(x)s)",
{"id": 1, "x": dbapi.Binary(b'a')}
)
print("Passed w %s" % dbapi)
except Exception as err:
print("Failed w %s: %s" % (dbapi, err))
try:
if dbapi.paramstyle == 'qmark':
cursor.execute(
"insert into t (id, x) values (?, ?)",
(2, dbapi.Binary(bytearray(b'a')))
)
else:
cursor.execute(
"insert into t (id, x) values (%(id)s, %(x)s)",
{"id": 1, "x": dbapi.Binary(bytearray(b'a'))}
)
print("Passed w %s" % dbapi)
except Exception as err:
print("Failed w %s: %s" % (dbapi, err))
try:
import pymssql
except ImportError:
pass
else:
conn = pymssql.connect(
user='scott', password='tiger', host='192.168.122.135', port=1213)
run_test(pymssql, conn)
conn.close()
try:
import pyodbc
except ImportError:
pass
else:
conn = pyodbc.connect(user='scott', password='tiger', dsn="ms_2008")
run_test(pyodbc, conn)
conn.close()
