def persistOperation(self, operation):
    """ update or insert request into db
        Also release the assignment tag

    :param operation: FTS3Operation instance
    """

    session = self.dbSession(expire_on_commit=False)

    # set the assignment to NULL
    # so that another agent can work on the request
    operation.assignment = None
    # because of the merge we have to explicitely set lastUpdate
    operation.lastUpdate = func.utc_timestamp()
    try:

      # Merge it in case it already is in the DB
      operation = session.merge(operation)
      session.add(operation)
      session.commit()
      session.expunge_all()

      return S_OK(operation.operationID)

    except SQLAlchemyError as e:
      session.rollback()
      self.log.exception("persistOperation: unexpected exception", lException=e)
      return S_ERROR("persistOperation: unexpected exception %s" % e)
    finally:
      session.close()

  def kickStuckJobs(self, limit=20, kickDelay=2):
    """finds jobs that have not been updated for more than a given
      time but are still assigned and resets the assignment

    :param int limit: number of jobs to treat
    :param int kickDelay: age of the lastUpdate in hours
    :returns: S_OK/S_ERROR with number of kicked jobs

    """

    session = self.dbSession(expire_on_commit=False)

    try:

      ftsJobs = session.query(FTS3Job.jobID)\
          .filter(FTS3Job.lastUpdate < (func.date_sub(func.utc_timestamp(),
                                                      text('INTERVAL %d HOUR' % kickDelay
                                                           ))))\
          .filter(~FTS3Job.assignment.is_(None))\
          .limit(limit)

      jobIDs = [jobTuple[0] for jobTuple in ftsJobs]
      rowCount = 0

      if jobIDs:
        result = session.execute(
            update(FTS3Job) .where(
                FTS3Job.jobID.in_(jobIDs)) .where(
                FTS3Job.lastUpdate < (
                    func.date_sub(
                        func.utc_timestamp(), text(
                            'INTERVAL %d HOUR' %
                            kickDelay)))) .values(
                {
                    'assignment': None}))
        rowCount = result.rowcount

      session.commit()
      session.expunge_all()

      return S_OK(rowCount)

    except SQLAlchemyError as e:
      session.rollback()
      return S_ERROR("kickStuckJobs: unexpected exception : %s" % e)
    finally:
      session.close()

    def define_tables(cls, metadata):
        Table(
            "t",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("x", Integer),
            Column("data", DateTime),
        )

        Table(
            "t_default",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("x", Integer),
            Column("idata", DateTime, default=func.utc_timestamp()),
            Column("udata", DateTime, onupdate=func.utc_timestamp()),
        )

def clearOldNotifications(conn):
    """Clear old notifications

  :param conn: SQLAlchemy connection object
  :type conn: sqlalchemy.engine.base.Connection
  """
    delete = schema.notification.delete().where(  # pylint: disable=E1120
        func.date_add(schema.notification.c.timestamp, text("INTERVAL 30 DAY")) < func.utc_timestamp()
    )
    conn.execute(delete)

def deleteStaleNotificationDevices(conn, days):
  """Deletes devices from notifications if they haven't been active recently.

  :param conn: SQLAlchemy connection object
  :type conn: sqlalchemy.engine.base.Connection
  :param days: the number of days of absense before considering a device stale
  :type days: int
  """
  query = schema.notification_settings.delete().where(  # pylint: disable=E1120
      func.date_add(schema.notification_settings.c.last_timestamp,
                    text("INTERVAL %i DAY" % days)) <
      func.utc_timestamp())
  conn.execute(query)

  def updateJobStatus(self, jobStatusDict):
    """ Update the job Status and error
        The update is only done if the job is not in a final state
        The assignment flag is released

       :param jobStatusDict : { jobID : { status , error, completeness } }
    """
    session = self.dbSession()
    try:

      for jobID, valueDict in jobStatusDict.iteritems():

        updateDict = {FTS3Job.status: valueDict['status']}

        # We only update error if it is specified
        if 'error' in valueDict:
          newError = valueDict['error']
          # Replace empty string with None
          if not newError:
            newError = None
          updateDict[FTS3Job.error] = newError

        if 'completeness' in valueDict:
          updateDict[FTS3Job.completeness] = valueDict['completeness']

        if valueDict.get('lastMonitor'):
          updateDict[FTS3Job.lastMonitor] = func.utc_timestamp()

        updateDict[FTS3Job.assignment] = None

        session.execute(update(FTS3Job)
                        .where(and_(FTS3Job.jobID == jobID,
                                    ~ FTS3Job.status.in_(FTS3Job.FINAL_STATES)
                                    )
                               )
                        .values(updateDict)
                        )
      session.commit()

      return S_OK()

    except SQLAlchemyError as e:
      session.rollback()
      self.log.exception("updateJobStatus: unexpected exception", lException=e)
      return S_ERROR("updateJobStatus: unexpected exception %s" % e)
    finally:
      session.close()

def updateNotificationDeviceTimestamp(conn, deviceId):
  """Updates last access timestamp for the specified device.

  :param conn: SQLAlchemy connection object
  :type conn: sqlalchemy.engine.base.Connection
  :param deviceId: Device uid
  :type deviceId: str
  :raises: ObjectNotFoundError when there is no device with deviceId configured
  """
  query = (schema.notification_settings
           .update()
           .where(schema.notification_settings.c.uid == deviceId)
           .values(last_timestamp=func.utc_timestamp()))
  result = conn.execute(query)
  if result.rowcount == 0:
    raise ObjectNotFoundError("No notification settings for device: %s" %
                              deviceId)

    def test_update_executemany_w_default(self):
        with testing.db.connect() as conn:
            timestamp = datetime.datetime(2015, 4, 17, 18, 5, 2)
            conn.execute(
                self.tables.t_default.insert(),
                [
                    {"x": 5, "idata": timestamp},
                    {"x": 6, "idata": timestamp},
                    {"x": 7, "idata": timestamp},
                ],
            )

            conn.execute(
                self.tables.t_default.update()
                .values(idata=func.utc_timestamp())
                .where(self.tables.t_default.c.x == bindparam("xval")),
                [{"xval": 5}, {"xval": 6}, {"xval": 7}],
            )

def _getCloudwatchMetricReadinessPredicate(conn):
  """ Generate an sqlAlchemy predicate that determines whether the metric is
  ready for data collection.

  :returns: sqlAlchemy predicate for use in `where` clause
  """
  # NOTE: the time arithmetic must be coordinated with
  # grok.app.aws.cloudwatch_utils.getMetricCollectionBackoffSeconds()

  # NOTE: the time difference logic here is a heuristic fine-tuned for
  # cloudwatch-based metrics as follows:
  #   * Cloudwatch metrics aggregated over a specific period (e.g., 5 minutes)
  #     appear to be arranged in contiguous time buckets each with a specific
  #     start and end time; we don't have visibility into the actual start time
  #     of any bucket, which appears to depend on when the cloudwatch metric was
  #     created.
  #   * In the higher-level logic, the first time we query a metric, we pick the
  #     metric's virtual starting time based on a 14-day backoff from current
  #     time (which is not the actual metric time bucket's start time) and
  #     subsequently add the metric's period to arrive at the next metric
  #     value's virtual start time, and so on.
  #   * Based on experiments with 5-minute-aggregated metrics, it appears that a
  #     metric's aggregated value becomes availabe one period after the true end
  #     of the metric value's time bucket. If you don't wait long enough, you
  #     either don't get any value from cloudwatch (which is a wasted slow call
  #     that contributes to API throttling) or you might get a non-final
  #     value.
  #   * Since we don't know the true start time of the time bucket, we
  #     compensate for it: first, we add the metric period to the virtual start
  #     time, which should get us at least to the end of the true time bucket;
  #     then we add another period to get us at least to the point in time where
  #     the time-bucket's data becomes available, and finally add a fudge value
  #     (60 seconds at time of this writing) for the metric value to stabilize
  return (
    (schema.metric.c.last_timestamp == None)
    | (func.timestampdiff(text("SECOND"),
                          schema.metric.c.last_timestamp,
                          func.utc_timestamp())
       >= (schema.metric.c.poll_interval +
           (schema.metric.c.poll_interval + 60))))

  def deleteFinalOperations(self, limit=20, deleteDelay=180):
    """deletes operation in final state that are older than given time

    :param int limit: number of operations to treat
    :param int deleteDelay: age of the lastUpdate in days
    :returns: S_OK/S_ERROR with number of deleted operations
    """

    session = self.dbSession(expire_on_commit=False)

    try:

      ftsOps = session.query(
          FTS3Operation.operationID) .filter(
          FTS3Operation.lastUpdate < (
              func.date_sub(
                  func.utc_timestamp(),
                  text(
                      'INTERVAL %d DAY' %
                      deleteDelay)))) .filter(
          FTS3Operation.status.in_(
              FTS3Operation.FINAL_STATES)) .limit(limit)

      opIDs = [opTuple[0] for opTuple in ftsOps]
      rowCount = 0
      if opIDs:
        result = session.execute(delete(FTS3Operation)
                                 .where(FTS3Operation.operationID.in_(opIDs)))
        rowCount = result.rowcount

      session.commit()
      session.expunge_all()

      return S_OK(rowCount)

    except SQLAlchemyError as e:
      session.rollback()
      return S_ERROR("deleteFinalOperations: unexpected exception : %s" % e)
    finally:
      session.close()

 def is_stuck(self, mins=10):
     diff = func.timestampdiff(text('minute'),
                               self.updated_at, func.utc_timestamp())
     return (self.updated_at == None) | (diff > mins)

from DIRAC.DataManagementSystem.Client.FTS3Job import FTS3Job
from DIRAC.ConfigurationSystem.Client.Utilities import getDBParameters

__RCSID__ = "$Id$"


metadata = MetaData()


fts3FileTable = Table('Files', metadata,
                      Column('fileID', Integer, primary_key=True),
                      Column('operationID', Integer,
                             ForeignKey('Operations.operationID', ondelete='CASCADE'),
                             nullable=False),
                      Column('attempt', Integer, server_default='0'),
                      Column('lastUpdate', DateTime, onupdate=func.utc_timestamp()),
                      Column('rmsFileID', Integer, server_default='0'),
                      Column('lfn', String(1024)),
                      Column('checksum', String(255)),
                      Column('size', BigInteger),
                      Column('targetSE', String(255), nullable=False),
                      Column('error', String(2048)),
                      Column('ftsGUID', String(255)),
                      Column('status', Enum(*FTS3File.ALL_STATES),
                             server_default=FTS3File.INIT_STATE,
                             index=True),
                      mysql_engine='InnoDB',
                      )

mapper(FTS3File, fts3FileTable)

 def test_utc_timestamp(self):
     self.assert_compile(func.utc_timestamp(), "UTC_TIMESTAMP")

 def test_utc_timestamp_fsp(self):
     self.assert_compile(
         func.utc_timestamp(5), "utc_timestamp(%s)",
         checkparams={"utc_timestamp_1": 5})

        )

    annotation_types = Table('annotation_types', meta,
        Column('annotation_type_id', Integer, primary_key=True),
        Column('name', String(255), nullable=False),
        UniqueConstraint('name', name='uc_annotation_type_name')
        )


    annotations = Table('annotations', meta,
        Column('annotation_id', Integer, primary_key=True),
        Column('document_id', Integer, nullable=False),
        Column('annotation_type_id', Integer, nullable=False),
        Column('identifier_id', Integer, nullable=False),
        Column('creator_id', Integer, nullable=False),
        Column('created_ts', DateTime(timezone=True), default=func.utc_timestamp()),
        Column('comment', Text, nullable=False),
        ForeignKeyConstraint(['document_id'], ['documents.document_id']),
        ForeignKeyConstraint(['annotation_type_id'], ['annotation_types.annotation_type_id']),
        ForeignKeyConstraint(['identifier_id'], ['identifiers.identifier_id']),
        ForeignKeyConstraint(['creator_id'], ['creators.creator_id']),
        )


    scanners = Table('scanners', meta,
        Column('scanner_id', Integer, primary_key=True),
        Column('name', String(255), nullable=False),
        UniqueConstraint('name', name='uc_scanner_name')
        )

    packages_scans = Table('packages_scans', meta,

#!/usr/bin/env python
# -*- coding:utf8 -*-

from datetime import datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, Date
from sqlalchemy.types import TypeDecorator
from sqlalchemy import DateTime
import time

from sqlalchemy import func
from sqlalchemy.sql.expression import select, text

dsn = "mysql://xxxxx:[email protected]/xxxxx?charset=utf8"
engine = create_engine(dsn, convert_unicode=True, echo=False)
db_session = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))

Base = declarative_base()
Base.query = db_session.query_property()

if __name__ == "__main__":

    r = db_session.execute(select([(func.utc_timestamp() - text("INTERVAL 2 DAY"))])).first()
    print r[0].strftime("%Y/%m/%d %H:%I:%S")

    # func => SQLの関数
    # text => SQLの文

 def test_insert_executemany(self):
     with testing.db.connect() as conn:
         conn.execute(
             self.tables.t.insert().values(data=func.utc_timestamp()),
             [{"x": 5}, {"x": 6}, {"x": 7}],
         )