redismap / redismap / converters.py

# -*- coding: utf-8 -*-
"""
redismap.converters
===================
Converters serialize data to and from bytestrings for storage in Redis.
They can be classes, instances, or whatever, just so long as they have two
methods: `to_redis` (takes a value and serializes it to `str`) and
`from_redis` (takes a `str` and deserializes it).

Converters can also convert their values to the scores for a sorted set using
`to_zscore` (takes a value and serializes it to `float`) and `from_zscore`
(takes a `float` and deserializes it).

:copyright: (C) 2011 Matthew Frazier
:license:   MIT/X11, see LICENSE for details
"""
import datetime
import decimal
import inspect
import time

def is_encoded(s, encoding):
    try:
        s.decode(encoding, "strict")
    except UnicodeDecodeError:
        return False
    else:
        return True


class TextConverter(object):
    """
    Instances of `TextConverter` can be used to store `unicode` text in an
    encoding of your choice. (Though `UTF8` and `UTF16` will suffice for most
    purposes.)
    
    :param encoding: The encoding to store the text in. This should be a valid
                     Python encoding.
    """
    def __init__(self, encoding):
        self.encoding = encoding
    
    def __repr__(self):
        return "TextConverter(%r)" % self.encoding
    
    def to_redis(self, value):
        encoding = self.encoding
        if isinstance(value, unicode):
            return value.encode(encoding)
        elif isinstance(value, str) and is_encoded(value, encoding):
            return value
        raise TypeError("can only convert unicode (and %s str) instances" %
                        encoding)
    
    def from_redis(self, value):
        return value.decode(self.encoding, 'replace')


#: This stores `unicode` by encoding it as UTF-8.
UTF8 = TextConverter('utf-8')

#: This stores `unicode` by encoding it as UTF-16, with a byte order mark.
#: Redis includes native support for UTF-8, so you should use `UTF8` instead
#: of this unless you have special requirements (i.e. your application
#: stores large amounts of CJK data).
UTF16 = TextConverter('utf-16')


class Bytes(object):
    """
    This just stores `str` objects as their constituent bytes.
    """
    @staticmethod
    def to_redis(value):
        return str(value)
    
    @staticmethod
    def from_redis(value):
        return value


class Integer(object):
    """
    This lets you store integers (and also longs).
    """
    @staticmethod
    def to_redis(value):
        return str(int(value))
    
    @staticmethod
    def from_redis(value):
        return int(value)
    
    @staticmethod
    def to_zscore(value):
        return float(int(value))
    
    @staticmethod
    def from_zscore(value):
        return int(value)


class Float(object):
    """
    This stores floating-point numbers. Not very useful for money - use
    `Decimal` instead.
    """
    @staticmethod
    def to_redis(value):
        return str(float(value))
    
    @staticmethod
    def from_redis(value):
        return float(value)
    
    @staticmethod
    def to_zscore(value):
        return float(value)
    
    @staticmethod
    def from_zscore(value):
        return value


class Decimal(object):
    """
    This stores `~decimal.Decimal` objects. Please use these instead of floats
    for money.
    """
    @staticmethod
    def to_redis(value):
        return str(decimal.Decimal(value))
    
    @staticmethod
    def from_redis(value):
        return decimal.Decimal(value)
    
    @staticmethod
    def to_zscore(value):
        return float(decimal.Decimal(value))
    
    @staticmethod
    def from_zscore(value):
        return decimal.Decimal(str(value))


class Boolean(object):
    """
    This lets you store Boolean values (`True` and `False`). It represents
    them as 0 and 1.
    """
    @staticmethod
    def to_redis(value):
        if value:
            return '1'
        else:
            return '0'
    
    @staticmethod
    def from_redis(value):
        return bool(int(value))


class _DateTimeBase(object):
    @staticmethod
    def to_zscore(value):
        if isinstance(value, datetime.datetime):
            return time.mktime(value.timetuple())
        elif isinstance(value, (int, long)) and value >= 0:
            return float(value)
        raise TypeError("can only convert datetime.datetime instances")
    
    @staticmethod
    def from_zscore(value):
        return datetime.datetime.fromtimestamp(int(value))


ISO_8601 = "%Y-%m-%dT%H:%M:%S"

class DateTime(_DateTimeBase):
    """
    This lets you store `datetime.datetime` instances, as ISO 8601 dates and
    times (``YYYY-MM-DDTHH:MM:SS``). They will be stored as UNIX timestamps
    in zset scores, however.
    """
    @staticmethod
    def to_redis(value):
        if isinstance(value, datetime.datetime):
            return value.strftime(ISO_8601)
        elif isinstance(value, (int, long)) and value >= 0:
            return datetime.datetime.fromtimestamp(value).strftime(ISO_8601)
        raise TypeError("can only convert datetime.datetime instances")
    
    @staticmethod
    def from_redis(value):
        return datetime.datetime.strptime(value, ISO_8601)


class Timestamp(_DateTimeBase):
    """
    Like `DateTime`, this lets you store `datetime.datetime` instances.
    Unlike `DateTime`, it stores them as UNIX timestamps, both in keys and
    in zset scores.
    """
    @staticmethod
    def to_redis(value):
        if isinstance(value, datetime.datetime):
            return str(int(time.mktime(value.timetuple())))
        elif isinstance(value, (int, long)) and value >= 0:
            return str(value)
        raise TypeError("can only convert datetime.datetime instances")
    
    @staticmethod
    def from_redis(value):
        return datetime.datetime.fromtimestamp(int(value))


class Date(object):
    """
    This stores `datetime.date` instances. It stores them as ISO 8601 dates
    in the database (``YYYY-MM-DD``), and as proleptic Gregorian ordinals
    as zset scores.
    """
    @staticmethod
    def to_redis(value):
        if isinstance(value, datetime.date):
            return value.strftime("%Y-%m-%d")
        elif isinstance(value, int) and value >= 0:
            return datetime.date.fromordinal(value).strftime("%Y-%m-%d")
        raise TypeError("can only convert datetime.date instances")
    
    @staticmethod
    def from_redis(value):
        return datetime.datetime.strptime(value, "%Y-%m-%d").date()
    
    @staticmethod
    def to_zscore(value):
        if isinstance(value, datetime.date):
            return float(value.toordinal())
        elif isinstance(value, int) and value >= 0:
            return float(value)
        raise TypeError("can only convert datetime.date instances")
    
    @staticmethod
    def from_zscore(value):
        return datetime.date.fromordinal(int(value))


#: This is a mapping of some basic Python types to the converters that will
#: convert those types.
BASIC_CONVERTERS = {
    unicode: UTF8,
    int: Integer,
    float: Float,
    decimal.Decimal: Decimal,
    bool: Boolean,
    datetime.datetime: DateTime,
    datetime.date: Date
}
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.