Module `event_processor.util.object_hash`

Expand source code

import hashlib
import pickle

class ObjectHash:
    @staticmethod
    def load():
        try:
            with open('/tmp/hashes', 'rb') as f:
                return pickle.load(f)
        except FileNotFoundError:
            return dict()
    
    @staticmethod
    def write(hashes):
        with open('/tmp/hashes', 'wb') as f:
            pickle.dump(hashes, f)

    @staticmethod
    def create_hash(obj):
        str_obj = str(obj).encode('utf-8')
        return hashlib.md5(str_obj).hexdigest()

    @staticmethod
    def get(key):
        hashes = ObjectHash.load()
        return hashes[key] if key in hashes else ''
    
    @staticmethod
    def set(key, value):
        hashes = ObjectHash.load()
        hashes[key] = value
        ObjectHash.write(hashes)

Classes

class ObjectHash (*args, **kwargs)

Expand source code

class ObjectHash:
    @staticmethod
    def load():
        try:
            with open('/tmp/hashes', 'rb') as f:
                return pickle.load(f)
        except FileNotFoundError:
            return dict()
    
    @staticmethod
    def write(hashes):
        with open('/tmp/hashes', 'wb') as f:
            pickle.dump(hashes, f)

    @staticmethod
    def create_hash(obj):
        str_obj = str(obj).encode('utf-8')
        return hashlib.md5(str_obj).hexdigest()

    @staticmethod
    def get(key):
        hashes = ObjectHash.load()
        return hashes[key] if key in hashes else ''
    
    @staticmethod
    def set(key, value):
        hashes = ObjectHash.load()
        hashes[key] = value
        ObjectHash.write(hashes)

Static methods

def create_hash(obj)

Expand source code

@staticmethod
def create_hash(obj):
    str_obj = str(obj).encode('utf-8')
    return hashlib.md5(str_obj).hexdigest()

def get(key)

Expand source code

@staticmethod
def get(key):
    hashes = ObjectHash.load()
    return hashes[key] if key in hashes else ''

def load()

Expand source code

@staticmethod
def load():
    try:
        with open('/tmp/hashes', 'rb') as f:
            return pickle.load(f)
    except FileNotFoundError:
        return dict()

def set(key, value)

Expand source code

@staticmethod
def set(key, value):
    hashes = ObjectHash.load()
    hashes[key] = value
    ObjectHash.write(hashes)

def write(hashes)

Expand source code

@staticmethod
def write(hashes):
    with open('/tmp/hashes', 'wb') as f:
        pickle.dump(hashes, f)