view weather_server/common.py @ 11:52ef21607b31

server: Create endpoint to get some recent readings.
author Paul Fisher <paul@pfish.zone>
date Sun, 06 Oct 2019 13:11:11 -0400
parents efe7a1eff167
children
line wrap: on
line source

import datetime
import json
import typing as t

import bson
import pytz

BSON_OPTIONS = bson.DEFAULT_CODEC_OPTIONS.with_options(
    tz_aware=True, tzinfo=pytz.UTC)


def bson_encode(data: t.Dict[str, t.Any]) -> bytes:
    return bson.BSON.encode(data, codec_options=BSON_OPTIONS)


class DateEncoder(json.JSONEncoder):

    def default(self, o: t.Any) -> t.Any:
        if not isinstance(o, datetime.datetime):
            return super().default(o)
        return o.timestamp()


JSON_ENCODER = DateEncoder(sort_keys=True)


def json_dumps(data: t.Any) -> str:
    return JSON_ENCODER.encode(data)


def utc_now():
    return datetime.datetime.utcnow().replace(tzinfo=pytz.UTC)