/usr/lib/python3/dist-packages/gnocchi/indexer/sqlalchemy_types.py is in python3-gnocchi 4.2.0-0ubuntu5.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | # -*- encoding: utf-8 -*-
#
# Copyright © 2016 Red Hat, Inc.
# Copyright © 2014-2015 eNovance
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import calendar
import datetime
import decimal
import iso8601
import sqlalchemy
from sqlalchemy.dialects import mysql
from sqlalchemy import types
from gnocchi import utils
class PreciseTimestamp(types.TypeDecorator):
"""Represents a timestamp precise to the microsecond.
Deprecated in favor of TimestampUTC.
Still used in alembic migrations.
"""
impl = sqlalchemy.DateTime
@staticmethod
def _decimal_to_dt(dec):
"""Return a datetime from Decimal unixtime format."""
if dec is None:
return None
integer = int(dec)
micro = (dec - decimal.Decimal(integer)) * decimal.Decimal(1000000)
daittyme = datetime.datetime.utcfromtimestamp(integer)
return daittyme.replace(microsecond=int(round(micro)))
@staticmethod
def _dt_to_decimal(utc):
"""Datetime to Decimal.
Some databases don't store microseconds in datetime
so we always store as Decimal unixtime.
"""
if utc is None:
return None
decimal.getcontext().prec = 30
return (decimal.Decimal(str(calendar.timegm(utc.utctimetuple()))) +
(decimal.Decimal(str(utc.microsecond)) /
decimal.Decimal("1000000.0")))
def load_dialect_impl(self, dialect):
if dialect.name == 'mysql':
return dialect.type_descriptor(
types.DECIMAL(precision=20,
scale=6,
asdecimal=True))
return dialect.type_descriptor(self.impl)
def compare_against_backend(self, dialect, conn_type):
if dialect.name == 'mysql':
return issubclass(type(conn_type), types.DECIMAL)
return issubclass(type(conn_type), type(self.impl))
def process_bind_param(self, value, dialect):
if value is not None:
value = utils.normalize_time(value)
if dialect.name == 'mysql':
return self._dt_to_decimal(value)
return value
def process_result_value(self, value, dialect):
if dialect.name == 'mysql':
value = self._decimal_to_dt(value)
if value is not None:
return utils.normalize_time(value).replace(
tzinfo=iso8601.iso8601.UTC)
class TimestampUTC(types.TypeDecorator):
"""Represents a timestamp precise to the microsecond."""
impl = sqlalchemy.DateTime
def load_dialect_impl(self, dialect):
if dialect.name == 'mysql':
return dialect.type_descriptor(mysql.DATETIME(fsp=6))
return self.impl
def process_bind_param(self, value, dialect):
if value is not None:
return utils.normalize_time(value)
def process_result_value(self, value, dialect):
if value is not None:
return value.replace(tzinfo=iso8601.iso8601.UTC)
|