CODEWITHSUNDEEP
pythondatetimesqlalchemyepoch
John Boik
John Boik

Reputation: 11

How to cast integer (epoch) column to datetime in sqlalchemy?

I'm using sqlalchemy to query a database and have a Table.c.Field object from inspect, where Field is of type integer (e.g., 1596657600). How do I cast that field to a datetime? In the query I want to cast Field as datetime, then later extract and groupby dayofweek (or some other aspect of datetime). But first I need to cast Field as a date or datetime.

I have tried a few ways, but all fail. In the code below I first import sqlalchemy as sa.

I tried

sa.func.to_timestamp(Table.c.Field / 1.0).cast(sa.Date)

as suggested here. But it produces the error: no such function: to_timestamp.

I tried

Table.c.Field.cast(sa.DateTime)

as suggested here. But as in that link it produces the error: Couldn't parse datetime string '1596657600' - value is not a string. If it helps, the database I am querying is also Sqlite, but I would like my solution to work for any integer field from any database.

I also tried

sa.cast(Table.c.Field, SQA.Interval)

but it produced the same error: Couldn't parse datetime string '1596657600' - value is not a string.

If the solution is to create some special function (e.g., Fx) to do the conversion, to be used as

(Table.c.Field).Fx()

what would that function look like? Or what kind of function would do the trick? I cannot write to the database, and need to do all calculations (casting, grouping, filtering, etc.) via a single sqlalchemy query.

Upvotes: 0

Views: 1983

Answers (1)

John Boik
John Boik

Reputation: 11

I found a solution in the docs, but I wasn't sure at first how to apply it in my case.

First, create a new type that accepts integers, then simply cast Field to that type. I first import sqlalchemy as SQA and import tzinfo and datetime and timedelta from datetime.

class MyEpochType(SQA.types.TypeDecorator):
    impl = SQA.types.Integer
    epoch = datetime(1970, 1, 1, 0, 0, 0, tzinfo=pytz.timezone('UTC'))
    
    def process_bind_param(self, value, dialect):
        pass # not needed in my use case

    def process_result_value(self, value, dialect):
        # either of the two returns below, or similar ones
        #return self.epoch + timedelta(seconds=value )
        return datetime.fromtimestamp(value).isoformat()

Table.c.Field.cast(MyEpochType)

The above solution will not allow for a group_by on the transformed values (e.g., a datetime or string). For this, a better solution is to use the native functions of the underlying database. In my case it is Sqlite, and would be called similar to that below:

SQA.func.strftime('%H%M',  
  SQA.func.datetime(Table.c.Field, 'unixepoch')).label('someLabel')

Here as an example I am calling two different Sqlite functions and giving the result a label. Each database has its own set of particular functions that can be called.

Upvotes: 1

Related Questions