Hello,
I got exception when I try to insert numpy.int8 type data.
sqlalchemy.exc.ProgrammingError: (psycopg.ProgrammingError) cannot adapt
type 'int8' using placeholder '%s' (format: AUTO)
I'm trying to adapt it with code below, but it looks like a bit
comprehensive.
Maybe you can suggest another solution? And what is a proper place to call
connection.adapters.register_dumper, is it ok to call it in that listener?
*models/__init__.py*
@listens_for(Pool, "connect")
def pool_on_connect(connection, _):
connection.adapters.register_dumper(np.integer, NumPyIntDumper)
*models/adapters.py *
from typing import Any
import numpy as np
from psycopg import errors as e, postgres
from psycopg._wrappers import Int2, Int4, Int8, IntNumeric
from psycopg.adapt import Dumper, PyFormat
class _NumPyIntDumper(Dumper):
def dump(self, obj: Any) -> bytes:
t = type(obj)
if not issubclass(t, np.integer):
raise e.DataError(f"integer expected, got
{type(obj).__name__!r}")
return str(obj).encode()
def quote(self, obj: Any) -> bytes:
value = self.dump(obj)
return value if obj >= 0 else b" " + value
class Int2Dumper(_NumPyIntDumper):
oid = postgres.types["int2"].oid
class Int4Dumper(_NumPyIntDumper):
oid = postgres.types["int4"].oid
class Int8Dumper(_NumPyIntDumper):
oid = postgres.types["int8"].oid
class IntNumericDumper(_NumPyIntDumper):
oid = postgres.types["numeric"].oid
class NumPyIntDumper(Dumper):
def dump(self, obj: Any) -> bytes:
raise TypeError(
f"{type(self).__name__} is a dispatcher to other dumpers:"
" dump() is not supposed to be called"
)
def get_key(self, obj: int, format: PyFormat) -> type:
return self.upgrade(obj, format).cls
_int2_dumper = Int2Dumper(Int2) # smallint
_int4_dumper = Int4Dumper(Int4) # integer
_int8_dumper = Int8Dumper(Int8) # bigint
_int_numeric_dumper = IntNumericDumper(IntNumeric) # numeric
def upgrade(self, obj: int, format: PyFormat) -> Dumper:
if isinstance(obj, (np.int8, np.int16, np.uint8)):
return self._int2_dumper
elif isinstance(obj, (np.int32, np.uint16)):
return self._int4_dumper
elif isinstance(obj, (np.int64, np.uint32)):
return self._int8_dumper
elif isinstance(obj, np.uint64):
return self._int_numeric_dumper
raise e.DataError(f"{type(obj).__name__!r} not supported by
NumPyIntDumper")
--
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 [email protected].
To view this discussion on the web visit
https://groups.google.com/d/msgid/sqlalchemy/db4fd851-bf4f-4a1e-a984-2398c027ce83n%40googlegroups.com.
