sqlalchemy: how to customize standard type like DateTime() param binding processing for dialect?

Question

Given the following snippet

t = Table(
    "foo",
    MetaData(),
    Column("bar", DateTime()),
)
engine.execute(t.insert((datetime(1900, 1, 1),)))
engine.execute(t.insert(("1900-01-01",))) 

the last statement works well for postgresql, while failing for Spark e.g.

Cannot safely cast 'bar': string to timestamp
[SQL: INSERT INTO TABLE `foo` VALUES (%(bar)s)]
[parameters: {'bar': '1900-01-01'}]

I can manage it with custom type like

class MyDateTime(TypeDecorator):
    impl = DateTime
    def process_bind_param(self, value, dialect):
        if dialect.name == "hive" and isinstance(value, str):
            return datetime.strptime(value, "%Y-%m-%d")
        return value


t = Table(
    "foo",
    MetaData(),
    Column("bar", MyDateTime()),
)

but solution

• seems hacky while we directly checks for dialect name
• I need to customize existing type for dialect, not implementing new one, because we have code base with DateTime type

Is there any solution for sqlalchemy to customize existing type?

Answer 1

There is no built-in way to customize the parameter binding processing for a standard SQLAlchemy type, such as DateTime(), for a specific dialect. However, there are a few workarounds that you can use.

One workaround is to create a custom type that wraps the standard type and overrides the process_bind_param() method. For example, the following code shows a custom type that overrides the process_bind_param() method to convert a string to a datetime object for the Hive dialect:

class MyDateTime(TypeDecorator):
    impl = DateTime

    def process_bind_param(self, value, dialect):
        if dialect.name == "hive" and isinstance(value, str):
            return datetime.strptime(value, "%Y-%m-%d")
        return value

You can then use this custom type in your SQLAlchemy schema instead of the standard DateTime() type. For example:

t = Table(
    "foo",
    MetaData(),
    Column("bar", MyDateTime()),
)

Another workaround is to use a custom dialect that overrides the bind_param() method for the standard DateTime() type. For example, the following code shows a custom dialect that overrides the bind_param() method to convert a string to a datetime object for the Hive dialect:

class MyHiveDialect(postgresql.PGDialect):
    def bind_param(self, value, type_):
        if type_ == DateTime and isinstance(value, str):
            return datetime.strptime(value, "%Y-%m-%d")
        return super().bind_param(value, type_)

You can then use this custom dialect when creating your SQLAlchemy engine. For example:

engine = create_engine("postgresql://localhost/foo", dialect=MyHiveDialect())

Which workaround you choose depends on your specific needs. If you only need to customize the parameter binding processing for a specific type for a single dialect, then the first workaround is probably the simplest solution. If you need to customize the parameter binding processing for multiple types or for multiple dialects, then the second workaround may be a better solution.

Please note that both of these workarounds are considered to be "hacky" solutions. There is no official way to customize the parameter binding processing for a standard SQLAlchemy type for a specific dialect. If you need to do this, then you should be aware that your code may be brittle and may not work with future versions of SQLAlchemy.

Answer 2

I think your code will be simple like this with adding "datetime.strptime", i like this

t = Table(
    "foo",
    MetaData(),
    Column("bar", DateTime()),
)
engine.execute(t.insert((datetime(1900, 1, 1),)))
engine.execute(t.insert(datetime.strptime("1900-01-01", "%Y-%m-%d")))