manyi/library/__init__.py

# -*- coding: utf-8 -*-
# !/usr/bin/env python

import random
import string
from contextlib import contextmanager

import six
from kombu.five import monotonic
from six.moves.urllib import parse as urlparse

import logging

__all__ = ('Map')

from typing import TYPE_CHECKING

logger = logging.getLogger(__name__)


class Map(dict):
    """
    提供字典的dot访问模式
    Example:
    m = Map({'first_name': 'Eduardo'}, last_name='Pool', age=24, sports=['Soccer'])
    """

    def __init__(self, *args, **kwargs):
        super(Map, self).__init__(*args, **kwargs)
        for arg in args:
            if isinstance(arg, dict):
                for k, v in arg.items():
                    if isinstance(v, dict):
                        v = Map(v)
                    self[k] = v

        if kwargs:
            for k, v in kwargs.items():
                if isinstance(v, dict):
                    v = Map(v)
                self[k] = v

    def __getattr__(self, attr):
        return self[attr]

    def __setattr__(self, key, value):
        self.__setitem__(key, value)

    def __getitem__(self, key):
        if key not in self.__dict__:
            super(Map, self).__setitem__(key, {})
            self.__dict__.update({key: Map()})
        return self.__dict__[key]

    def __setitem__(self, key, value):
        super(Map, self).__setitem__(key, value)
        self.__dict__.update({key: value})

    def __delattr__(self, item):
        self.__delitem__(item)

    def __delitem__(self, key):
        super(Map, self).__delitem__(key)
        del self.__dict__[key]


def to_text(value, encoding='utf-8'):
    """Convert value to unicode, default encoding is utf-8

    :param value: Value to be converted
    :param encoding: Desired encoding
    """
    if not value:
        return ''
    if isinstance(value, six.text_type):
        return value
    if isinstance(value, six.binary_type):
        return value.decode(encoding)
    return six.text_type(value)


def to_binary(value, encoding='utf-8'):
    """Convert value to binary string, default encoding is utf-8

    :param value: Value to be converted
    :param encoding: Desired encoding
    """
    if not value:
        return b''
    if isinstance(value, six.binary_type):
        return value
    if isinstance(value, six.text_type):
        return value.encode(encoding)
    return to_text(value).encode(encoding)


def timezone(zone):
    """Try to get timezone using pytz or python-dateutil

    :param zone: timezone str
    :return: timezone tzinfo or None
    """
    try:
        # noinspection PyUnresolvedReferences
        import pytz
        return pytz.timezone(zone)
    except ImportError:
        pass
    try:
        from dateutil.tz import gettz
        return gettz(zone)
    except ImportError:
        return None


class ObjectDict(dict):
    """Makes a dictionary behave like an object, with attribute-style access.
    """

    def __getattr__(self, key):
        if key in self:
            return self[key]
        return None

    def __setattr__(self, key, value):
        self[key] = value


def random_string(length=16):
    rule = string.ascii_letters + string.digits
    rand_list = random.sample(rule, length)
    return ''.join(rand_list)


def get_querystring(uri):
    """Get Querystring information from uri.

    :param uri: uri
    :return: querystring info or {}
    """
    parts = urlparse.urlsplit(uri)
    return urlparse.parse_qs(parts.query)


def byte2int(c):
    if six.PY2:
        return ord(c)
    return c


if TYPE_CHECKING:
    class MemcacheProxy(object):
        def add(self, key, value, timeout=0, version=None):
            pass

        def get(self, key, default=None, version=None):
            pass

        def set(self, key, value, timeout=0, version=None):
            pass

        def delete(self, key, version=None):
            pass

        def get_many(self, keys, version=None):
            pass

        def incr(self, key, delta=1, version=None):
            pass

        def decr(self, key, delta=1, version=None):
            pass


@contextmanager
def my_memcache_lock(mc, key, value, expire=60 * 10):
    # type: (MemcacheProxy, str, str, int)->None

    """
    Example usage
    ```
     with my_memcache_lock(cache, lock_id, self.app.oid) as acquired:
        if acquired:
            return Feed.objects.import_feed(feed_url).url
            logger.debug(
                'Feed %s is already being imported by another worker', feed_url)
    ```
    :param mc:
    :param key:
    :param value:
    :param expire
    :return:
    """

    value = str(value)

    timeout_at = monotonic() + expire - 3

    status = mc.add(key, value, expire)

    logger.debug('memcache_lock add result is:  {}; key is: {}'.format(status, key))

    try:
        yield status
    finally:
        # memcache delete is very slow, but we have to use it to take
        # advantage of using add() for atomic locking
        if monotonic() < timeout_at and status:
            # don't release the lock if we exceeded the timeout
            # to lessen the chance of releasing an expired lock
            # owned by someone else
            # also don't release the lock if we didn't acquire it
            result = mc.delete(key)

            logger.debug('memcache_lock delete result is:  {}; key is: {}'.format(result, key))