oh, I though you were asking how to use psycopg's API. If you want to use SQLAlchemy techniques, you would use TypeDecorator:
https://docs.sqlalchemy.org/en/14/core/custom_types.html#augmenting-existing-types your type would look just like https://docs.sqlalchemy.org/en/14/core/custom_types.html#coercing-encoded-strings-to-unicode: from sqlalchemy.types import TypeDecorator, Integer class CoerceNumpyInt(TypeDecorator): impl = Integer cache_ok = True def process_bind_param(self, value, dialect): if value is not None: value = int(value) return value that would be it On Sun, Sep 11, 2022, at 8:09 AM, sector119 wrote: > Thank You, Mike, I thought that I can adapt types with sqlalchemy, not with > db adapter like psycopg, to make it not db adapter dependent solution. > > воскресенье, 11 сентября 2022 г. в 04:14:02 UTC+3, Mike Bayer: >> __ >> maybe email on the psycopg2 list / github tracker >> >> On Sat, Sep 10, 2022, at 4:15 PM, sector119 wrote: >>> 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 sqlalchemy+...@googlegroups.com. >>> To view this discussion on the web visit >>> https://groups.google.com/d/msgid/sqlalchemy/db4fd851-bf4f-4a1e-a984-2398c027ce83n%40googlegroups.com >>> >>> <https://groups.google.com/d/msgid/sqlalchemy/db4fd851-bf4f-4a1e-a984-2398c027ce83n%40googlegroups.com?utm_medium=email&utm_source=footer>. >> > > > -- > 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+unsubscr...@googlegroups.com. > To view this discussion on the web visit > https://groups.google.com/d/msgid/sqlalchemy/25a58769-42e5-4c43-87d3-0cfeefa8f192n%40googlegroups.com > > <https://groups.google.com/d/msgid/sqlalchemy/25a58769-42e5-4c43-87d3-0cfeefa8f192n%40googlegroups.com?utm_medium=email&utm_source=footer>. -- 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+unsubscr...@googlegroups.com. To view this discussion on the web visit https://groups.google.com/d/msgid/sqlalchemy/c18a95cb-542e-478f-8338-8248404957ce%40www.fastmail.com.
