sqlalchemy_DateTimeTrucated.py

from sqlalchemy import create_engine, Column, types
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.sql import functions, expression, func
from sqlalchemy.ext.compiler import compiles

from datetime import datetime

Base = declarative_base()

class DateTimeTrucated(types.DateTime):
    def _result_processor(self, dialect, coltype):
        def process(value):
            func.DATE_TRUNC('second', value)
        if dialect.name == 'postgresql':
            return process
        return None
        
    def column_expression(self, colexpr):
        return func.DATE_TRUNC('second', colexpr)

    def _adapt(self, *args, **kw):
        return DateTimeTrucated(**kw)

class Model(Base):
    __tablename__ = "model"
    id = Column(types.Integer, primary_key=True)
    col1 = Column(DateTimeTrucated, default=datetime.now)
    col2 = Column(DateTimeTrucated, default=datetime.now)
    
    def __repr__(self):
        return "Model(col1=%r, col2=%r)" % (self.col1, self.col2)

sqlite_engine = create_engine('sqlite:///:memory:', echo=True)
mysql_engine = create_engine('mysql:///test?unix_socket=/tmp/mysql/socket', echo=True)
psql_engine = create_engine('postgresql:///postgres?host=/tmp/pg', echo=True)

def create_session(engine=sqlite_engine, drop=True, create=True):
    if drop:
        Base.metadata.drop_all(engine)
    if create:
        Base.metadata.create_all(engine)
    return scoped_session(sessionmaker(bind=engine)) 

session = create_session(psql_engine)

    
session.add(Model(col1=datetime(2016, 01, 01, 18, 00, 00, 100), col2=datetime(2016, 01, 01, 18, 00, 01,  50)))
session.add(Model(col1=datetime(2016, 01, 01, 18, 00, 00, 100), col2=datetime(2016, 01, 01, 18, 00, 03, 100)))
#result=session.query(Model.id, Model.col1, Model.col2, func.sum(func.age(Model.col2, Model.col1))).group_by(Model.id).all()
#session.rollback()
#result