Reputation: 362
Background, there are several ways to store dates in MySQ.
As a MySQL Date field, another mysql specific data type.
It's very important not to confuse case 2 with case 3 (or case 4). I have an existing table with an integer date field (case 2) how can I define it in sqlalchemy in a way I don't have to access mysql's "FROM_UNIXTIME" function?
For the record, just using sqlalchemy.types.DateTime and hoping it does the right thing when it detects an integer column doesn't work, it works for timestamp fields and date fields.
Upvotes: 4
Views: 5627
Reputation: 30963
I think there is a couple of issues with the type decorator you showed.
impl
should be sqlalchemy.types.Integer
instead of DateTime
.Here's the what I have in mind:
import datetime, time
from sqlalchemy.types import TypeDecorator, DateTime, Integer
class IntegerDateTime(TypeDecorator):
"""a type that decorates DateTime, converts to unix time on
the way in and to datetime.datetime objects on the way out."""
impl = Integer # In schema, you want these datetimes to
# be stored as integers.
def process_bind_param(self, value, _):
"""Assumes a datetime.datetime"""
if value is None:
return None # support nullability
elif isinstance(value, datetime.datetime):
return int(time.mktime(value.timetuple()))
raise ValueError("Can operate only on datetime values. "
"Offending value type: {0}".format(type(value).__name__))
def process_result_value(self, value, _):
if value is not None: # support nullability
return datetime.datetime.fromtimestamp(float(value))
Upvotes: 8
Reputation: 362
So yeah, this approach works. And I ended up answering my own question :/, hope somebody finds this useful.
import datetime, time
from sqlalchemy.types import TypeDecorator, DateTime
class IntegerDateTime(TypeDecorator):
"""a type that decorates DateTime, converts to unix time on
the way in and to datetime.datetime objects on the way out."""
impl = DateTime
def process_bind_param(self, value, engine):
"""Assumes a datetime.datetime"""
assert isinstance(value, datetime.datetime)
return int(time.mktime(value.timetuple()))
def process_result_value(self, value, engine):
return datetime.datetime.fromtimestamp(float(value))
def copy(self):
return IntegerDateTime(timezone=self.timezone)
Upvotes: 3