import math import json from django.core.cache import cache from django.core.serializers.json import DjangoJSONEncoder from channels.channel import Group from channels.generic.websockets import JsonWebsocketConsumer def to_ukf(balance, is_cof=False): """Convert euro to UKF.""" from .config import kfet_config subvention = kfet_config.subvention_cof grant = (1 + subvention / 100) if is_cof else 1 return math.floor(balance * 10 * grant) # Storage class CachedMixin: """Object with cached properties. Attributes: cached (dict): Keys are cached properties. Associated value is the returned default by getters in case the key is missing from cache. cache_prefix (str): Used to prefix keys in cache. """ cached = {} cache_prefix = '' def __init__(self, cache_prefix=None, *args, **kwargs): super().__init__(*args, **kwargs) if cache_prefix is not None: self.cache_prefix = cache_prefix def cachekey(self, attr): return '{}__{}'.format(self.cache_prefix, attr) def __getattr__(self, attr): if attr in self.cached: return cache.get(self.cachekey(attr), self.cached.get(attr)) elif hasattr(super(), '__getattr__'): return super().__getattr__(attr) else: raise AttributeError("can't get attribute") def __setattr__(self, attr, value): if attr in self.cached: cache.set(self.cachekey(attr), value) elif hasattr(super(), '__setattr__'): super().__setattr__(attr, value) else: raise AttributeError("can't set attribute") def clear_cache(self): cache.delete_many([ self.cachekey(attr) for attr in self.cached.keys() ]) # Consumers class DjangoJsonWebsocketConsumer(JsonWebsocketConsumer): """Custom Json Websocket Consumer. Encode to JSON with DjangoJSONEncoder. """ @classmethod def encode_json(cls, content): return json.dumps(content, cls=DjangoJSONEncoder) class PermConsumerMixin: """Add support to check permissions on consumers. Attributes: perms_connect (list): Required permissions to connect to this consumer. message.user is appended as argument to each connection_groups method call. """ http_user = True # Enable message.user perms_connect = [] def connect(self, message, **kwargs): """Check permissions on connection.""" if message.user.has_perms(self.perms_connect): super().connect(message, **kwargs) else: self.close() def raw_connect(self, message, **kwargs): # Same as original raw_connect method of JsonWebsocketConsumer # We add user to connection_groups call. groups = self.connection_groups(user=message.user, **kwargs) for group in groups: Group(group, channel_layer=message.channel_layer).add(message.reply_channel) self.connect(message, **kwargs) def raw_disconnect(self, message, **kwargs): # Same as original raw_connect method of JsonWebsocketConsumer # We add user to connection_groups call. groups = self.connection_groups(user=message.user, **kwargs) for group in groups: Group(group, channel_layer=message.channel_layer).discard(message.reply_channel) self.disconnect(message, **kwargs) def connection_groups(self, user, **kwargs): """`message.user` is available as `user` arg. Original behavior.""" return super().connection_groups(user=user, **kwargs)