Skip to content

Instantly share code, notes, and snippets.

@zzzeek
Created November 14, 2013 17:31
Show Gist options
  • Save zzzeek/7470863 to your computer and use it in GitHub Desktop.
Save zzzeek/7470863 to your computer and use it in GitHub Desktop.
from sqlalchemy import types
from sqlalchemy import event, create_engine, MetaData, Table
import datetime
class MyEpochType(types.TypeDecorator):
impl = types.Integer
epoch = datetime.datetime(1970, 1, 1, 0, 0, 0)
def process_bind_param(self, value, dialect):
return (value / 1000 - self.epoch).total_seconds()
def process_result_value(self, value, dialect):
return self.epoch + datetime.timedelta(seconds=value / 1000)
@event.listens_for(Table, "column_reflect")
def setup_epoch(inspector, table, column_info):
if isinstance(column_info['type'], types.DateTime):
column_info['type'] = MyEpochType()
engine = create_engine("sqlite://")
engine.execute("""
CREATE TABLE "test" (
"date" datetime
)""")
engine.execute("INSERT INTO test VALUES(1384008289278);")
metadata = MetaData()
metadata.reflect(bind=engine)
print engine.execute(metadata.sorted_tables[0].select()).fetchall()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment