Defining a table with sqlalchemy with a mysql unix timestamp

Background, there are several ways to store dates in MySQ.

  1. As a string e.g. "09/09/2009".
  2. As integer using the function UNIX_TIMESTAMP() this is supposedly the traditional unix time representation (you know seconds since the epoch plus/minus leap seconds).
  3. As a MySQL TIMESTAMP, a mysql specific data type not the same than unix timestamps.
  4. As a MySQL Date field, another mysql specific data type.

    It's very important not to confuse case 2 with case 3 (or case 4). I have an existing table with an integer date field (case 2) how can I define it in sqlalchemy in a way I don't have to access mysql's "FROM_UNIXTIME" function?

    For the record, just using sqlalchemy.types.DateTime and hoping it does the right thing when it detects an integer column doesn't work, it works for timestamp fields and date fields.

Asked by: Vanessa194 | Posted: 06-12-2021

Answer 1

I think there is a couple of issues with the type decorator you showed.

  1. impl should be sqlalchemy.types.Integer instead of DateTime.
  2. The decorator should allow nullable columns.

Here's the what I have in mind:

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))

Answered by: Rebecca442 | Posted: 07-01-2022

Answer 2

So yeah, this approach works. And I ended up answering my own question :/, hope somebody finds this useful.

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)

Answered by: Eric197 | Posted: 07-01-2022

Similar questions

python - How to convert SQLAlchemy TIMESTAMP column read-only buffer to datetime?

I have inherited an SQLAlchemy model which interfaces to an MSSQL db. Most tables have a modified column. The columns are calculated within the db so the model declares them like this: modified = Column(TIMESTAMP, FetchedValue()) When I call this property it comes back as a read-only buffer: >>> Thing.modified <read-only buffer for 0xa901fa0, size...

python - Insert MySQL timestamp column value with SqlAlchemy

I have a sqlalchemy class mapping to a database table in MySQL innoDB. The table has several columns and I am able to successfully populate them all except from a TIMESTAMP column: The mapping: class HarvestSources(Base): __table__ = Table('harvested', metadata, autoload=True) The column on MySQL is a TIMESTAMP which has CURRENT_TIMESTAMP as default value, but when I insert a r...

python - Select date from timestamp in SQLAlchemy

I'm using SQLAlchemy with python and i want to select date from column type timestamp, to do this query: SELECT DATE(`record_date`) FROM Users I made this code by sql alchemy but it will return the timestamp: session.query(Users.record_date).all() How can i do it?

python - SQLAlchemy MySQL Timestamp column value

I need to execute a query which compares only the year and month value from TIMESTAMP column where the records look like this: 2015-01-01 08:33:06 The SQL Query is very simple (the interesting part is the year(timestamp) and month(timestamp) which extracts the year and the month so I can use them for comparison: SELECT model, COUNT(model) A...

python - SqlAlchemy TIMESTAMP 'on update' extra

I am using SqlAlchemy on python3.4.3 to manage a MySQL database. I was creating a table with: from datetime import datetime from sqlalchemy import Column, text, create_engine from sqlalchemy.types import TIMESTAMP from sqlalchemy.dialects.mysql import BIGINT from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class MyClass(Base): __tablename__ = 'my_class' id = Column(B...

python - Convert datetime to unix timestamp in SQLAlchemy model before executing query?

I am using SQLAlchemy to work with a remote database that uses a strange timestamp format--it stores timestamps as double-precision milliseconds since epoch. I'd like to work with python datetime objects, so I wrote getter/setter methods in my model, following this gist: from sqlalchemy.ext.declarative import declarative_base from sqlalchem...

python - SQLAlchemy fails to insert timestamp into MSSQL

sqlalchemy.exc.IntegrityError: (pyodbc.IntegrityError) ('23000', '[23000] [FreeTDS][SQL Server]Cannot insert an explicit value into a timestamp column. Use INSERT with a column list to exclude the timestamp column, or insert a D EFAULT into the timestamp column. (273) (SQLExecDirectW)') [SQL: 'INSERT INTO task_events (task_id, event_name, ts) OUTP UT VALUES (?, ?, ?)'] [parameters: (...

python - SQLAlchemy group by day in timestamp column

I have an ORM (Model) defined in SQLAlchemy as below: class StoreView(Base): __tablename__ = 'store_views' id = Column(Integer, primary_key=True) store_id = Column(Integer) started_from = Column(TIMESTAMP) end_to = Column(TIMESTAMP) average_watch_time = Column(Float) total_watch_time = Column(Float) total_views = Column(Float) I'm planning to get the sum of a...

python - Creation timestamp of a row in SQLAlchemy

Using SQLAlchemy ORM, what is a good way to record the creation time of a record, without overwriting the creation time on subsequent updates? Here's a sample table definition: class Car: model = Column(String, primary_key=True) latest_model_year = Column(Integer) created = Column(DateTime) I would like to use this code to add/update a Car row in the database:

python - How can I use UUIDs in SQLAlchemy?

Is there a way to define a column (primary key) as a UUID in SQLAlchemy if using PostgreSQL (Postgres)?

python - Getting random row through SQLAlchemy

How do I select one or more random rows from a table using SQLAlchemy?

python - How to add an automatic filter to a relation with SQLAlchemy?

I'm using SQLAlchemy 0.5rc, and I'd like to add an automatic filter to a relation, so that every time it tries to fetch records for that relation, it ignores the "remote" ones if they're flagged as "logically_deleted" (a boolean field of the child table) For example, if an object "parent" has a "children" relation that has 3 records, but one of them is logically deleted, when I query for "Parent" I'd like SQLA to f...

python - What is the sqlalchemy equivalent column type for 'money' and 'OID' in Postgres?

What is the sqlalchemy equivalent column type for 'money' and 'OID' column types in Postgres?

python - SQLAlchemy and empty columns

When I try to insert a new record into the database using SQLAlchemy and I don't fill out all values, it tries to insert them as "None" (instead of omitting them). It then complains about "can't be null" errors. Is there a way to have it just omit columns from the sql query if I also omitted them when declaring the instance?

python - SQLAlchemy DateTime timezone

SQLAlchemy's DateTime type allows for a timezone=True argument to save a non-naive datetime object to the database, and to return it as such. Is there any way to modify the timezone of the tzinfo that SQLAlchemy passes in so it could be, for instance, UTC? I realize that I could just use default=datetime.datetime.utcnow; however, this is a naive time that would happily ac...

python - How can I get all rows with keys provided in a list using SQLalchemy?

I have sequence of IDs I want to retrieve. It's simple: session.query(Record).filter( Is there a better way to do it?

python - How can I order objects according to some attribute of the child in sqlalchemy?

Here is the situation: I have a parent model say BlogPost. It has many Comments. What I want is the list of BlogPosts ordered by the creation date of its' Comments. I.e. the blog post which has the most newest comment should be on top of the list. Is this possible with SQLAlchemy?

python - SQLAlchemy - INSERT OR REPLACE equivalent

does anybody know what is the equivalent to SQL "INSERT OR REPLACE" clause in SQLAlchemy and its SQL expression language? Many thanks -- honzas

python - How to generate a file with DDL in the engine's SQL dialect in SQLAlchemy?

Suppose I have an engine pointing at MySQL database: engine = create_engine('mysql://arthurdent:answer42@localhost/dtdb', echo=True) I can populate dtdb with tables, FKs, etc by: metadata.create_all(engine) Is there an easy way to generate the SQL file that contains all the DDL statements instead of actually applying these DDL sta...

Still can't find your answer? Check out these communities...

PySlackers | Full Stack Python | NHS Python | Pythonist Cafe | Hacker Earth | Discord Python