Определение таблицы с помощью sqlalchemy с временной меткой mysql unix
-
11-09-2019 - |
Вопрос
Предыстория: существует несколько способов хранения дат в MySQ.
- В виде строки, например.«09.09.2009».
- В качестве целого числа с использованием функции UNIX_TIMESTAMP() это предположительно традиционное представление времени Unix (вы знаете секунды с начала эпохи плюс/минус дополнительные секунды).
- В качестве MySQL TIMESTAMP — тип данных MySQL, отличный от временных меток Unix.
В качестве поля даты MySQL — еще один тип данных, специфичный для MySQL.
Очень важно не путать случай 2 со случаем 3 (или случаем 4).У меня есть существующая таблица с целочисленным полем даты (случай 2). Как я могу определить ее в sqlalchemy таким образом, чтобы мне не требовался доступ к функции MySQL «FROM_UNIXTIME»?
Для справки, простое использование sqlalchemy.types.DateTime и надежда, что он поступит правильно, когда обнаружит целочисленный столбец, не работает, это работает для полей меток времени и полей даты.
Решение
Я думаю, что с показанным вами декоратором шрифтов есть пара проблем.
impl
должно бытьsqlalchemy.types.Integer
вместоDateTime
.- Декоратор должен разрешать столбцы, допускающие значение NULL.
Вот что я имею в виду:
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))
Другие советы
Так что да, этот подход работает.И в итоге я ответил на свой вопрос :/, надеюсь, кто-нибудь найдет это полезным.
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)