A Python implementation of Craig Kerstiens's blog post, "Working with Time"

working_with_pg_time.py

"""
This is a python implementation of Craig Kerstiens blog post:
   http://www.craigkerstiens.com/2017/06/08/working-with-time-in-postgres/
"""

from random import randint
import functools
from datetime import datetime
import numpy as np
from arrow import Arrow
from faker import Faker
from sqlalchemy import (DateTime, BigInteger, Table, Column, create_engine,
                        text, MetaData, Text, select, cast, func, Interval,
                        literal, desc, and_)

db_url = "postgres://username:password@localhost:5432/db"
engine = create_engine(db_url, echo=True)
metadata = MetaData()

t_user = Table(
    'tmp_user', metadata,
    Column('pk_id', BigInteger, primary_key=True, autoincrement=True),
    Column('handle', Text, nullable=False, unique=True),
    Column('created_at', DateTime(True), nullable=False,
           server_default=text("now()::timestamptz::timestamptz")),
    schema='public')


def setup_users():
    conn = engine.connect()
    metadata.drop_all(conn)
    metadata.create_all(conn)

    fake = Faker()

    for day in Arrow.range('day', datetime(2017, 1, 1), datetime.now()):
        if np.random.choice([True, False], p=[0.8, 0.2]):
            users = [{'handle': fake.first_name()+fake.last_name()+fake.user_name(),
                      'created_at': day.datetime} for x in range(randint(1, 50))]
            conn.execute(t_user.insert().values(users))


def runnable(fn):
    @functools.wraps(fn)
    def wrap(*args, **kwargs):
        conn = engine.connect()
        query = fn(*args, **kwargs)
        return conn.execute(query).fetchall()
    return wrap


@runnable
def interval_example(interval):
    """
    Finding all users that have signed up for service within the interval

    If we wanted to find the users that signed up within the last week:
        SELECT *
        FROM tmp_user
        WHERE created_at >= now() - '1 week'::interval
    """
    return (select([t_user]).
            where(t_user.c.created_at >= func.now() - cast(interval, Interval())))


@runnable
def date_trunc_example(field):
    """
    date_trunc will truncate a date to some interval level

    Valid values for field are:
        microseconds, milliseconds, second, minute, hour, day, week, month,
        quarter, year, decade, century, millennium

    If we wanted to find the count of users that signed up per week:
        SELECT date_trunc('week', created_at),
               count(*)
        FROM tmp_user
        GROUP BY 1
        ORDER BY 1 DESC;
    """
    return (select([func.date_trunc(field, t_user.c.created_at).label('field'),
                    func.count().label('total_users')]).
            group_by(literal(1)).order_by(desc(literal(1))))


@runnable
def weekly_signups_example():
    """
    WITH weeks as (
        SELECT week
        FROM generate_series('2017-01-01'::date, now()::date, '1 week'::interval) week
    )
    SELECT weeks.week, count(*)
    FROM weeks, tmp_user
    WHERE
        tmp_user.created_at >= weeks.week and
        tmp_user.created_at < (weeks.week + '1 week'::interval)
    GROUP BY week;
    """
    weekly_series = func.generate_series(datetime(2017, 1, 1),
                                         datetime.now(),
                                         cast('1 week', Interval()))
    weeks = (select([weekly_series.label('week')])).cte(name="weeks")

    return (select([weeks.c.week, func.count()]).
            select_from(weeks).
            where(and_(t_user.c.created_at >= weeks.c.week,
                       t_user.c.created_at < (weeks.c.week + cast('1 week', Interval())))).
            group_by(weeks.c.week).order_by(desc(literal(1))))


if __name__ == '__main__':
    setup_users()