andrewshkovskii/companies_cache.py Secret

## companies_cache.py
# coding: utf-8
"""
Стояла задача - реализовать дерево гос. учреждений для РФ.
Иметь возможность перемещать учреждения и получать актуальные поддеревья(списки)
дочерних учреждение в режиме "реального времени".

А так же обеспечить кэш дерева.

Учреждение может быть привязано к субъекту РФ и к другому учреждению.
Глубина дерева может быть бесконечно большой. т.е. несколько десятков тысяч
учреждений.

Модель (упрощенная):
class Company:
    administrative_unit_id: ID субъекта.
    parent_id: ID вышестоящего Company. Может быть null.

Задача решалась в несколько этапов.

Первый этап был решением через django-ORM(проект написан на django)
и простом юнионом кверисетов рекурсивно.
Проработало это где-то пару месяцев пока не появилась более-менее настоящая
нагрузка и все умирало.

Вторым этапом было решено взять django-mptt
(https://github.com/django-mptt/django-mptt).
Это "полноценная" Реализация дерева на SQL с своими менеджероми,
пасьянсом и поэтесами.
Это проработало где-то полтора месяца и вылезли баги:
Деревья очень долго перестраивались (постгрес вообще мог отвалиться в
не-операбельный режим). Постоянно множились ошибки из-за которых приходилось
пересобирать дерево и система в эти моменты была недоступа.
В общем - было не сильно лучше.

Далее мы подумали и решили, что можно это дерево разложить в плоской структуре
и собирать его онлайн. И использовать то, что уже есть в стеке - редис.
Сначала это показалось не лучшим решением на фоне потребляемой памяти и
скорости работы питона - но в итоге это было неплохо.
Все дерево по РФ можно собрать где-то за 5секунд.
Для "большого" мира софта это много, но для нашей задачи это было приемлимо.
Тем более - сбор всего дерева РФ очень редкая операция.

В ходе разработки были произведены некоторые оптимизации(или микро-оптимизации).
В общем - это работает и нас устраивает то, как это работает сейчас.

Пока производительности хватает, но есть идеи для перехода на Neo4j (
https://ru.wikipedia.org/wiki/Neo4j).
Не переходили изначально из-за того, что не хотелось увеличивать стек технологий
для поддержки. ну и конечно перед переходом надо будет все это профилировать,
тестировать.

Текущая статистика по редису:

uptime_in_days:53 (Аптайм маленький из-за недавних ребутов серверов.
Сервера стоят не на нашей площадке)

# Stats
total_connections_received:15002
total_commands_processed:357683126
instantaneous_ops_per_sec:80
rejected_connections:0
sync_full:0
sync_partial_ok:0
sync_partial_err:0
expired_keys:0
evicted_keys:0
keyspace_hits:350306471
keyspace_misses:71
pubsub_channels:0
pubsub_patterns:0
latest_fork_usec:0

# Memory
used_memory:135802672
used_memory_human:129.51M
used_memory_rss:154419200
used_memory_peak:138898064
used_memory_peak_human:132.46M
used_memory_lua:33792

# Keyspace
db0:keys=125378,expires=0,avg_ttl=0, из них только 95 - Субъекты.

Дерево меняется несколько сотен раз за сутки (добавление, удаление и
перемещение нод).

Для пущей производительности был отключен дамп базы редиса в файл, что
позволило избежать Timeout Error'ов. Если что-то упадет - все может
относителньо быстро (~2 минуты) собраться опять. В общем - это было не критично.

"""
from __future__ import absolute_import, unicode_literals

# third party
import ujson

# project
from cache.administrative_units_cache import set_unit_companies_cache
from cache import TREE_CLIENT as redis_client

# {administrative_unit_id}:{company_id}
COMPANY_KEY_TEMPLATE = "cmp:{0}:{1}"


def make_company_key(company_id, company_unit_id):
    """Создаем ключ для редиса на основе COMPANY_KEY_TEMPLATE.

    :param company_id: айди учреждения
    :param company_unit_id: айди юнита учреждения.

    :return Созданный ключ.

    """
    return COMPANY_KEY_TEMPLATE.format(company_unit_id, company_id)


def get_company_hash(company_instance):
    """Генерирует хэш для записи в редис.

    :param company_instance: Инстанс учреждения для записи

    :return Словарь с данными по учреждению для записи в редис

    """
    children = [
        cid for cid in company_instance.children.values_list("id", flat=True)
        ]
    return {
        "id": company_instance.id,
        "parent_id": company_instance.parent_id,
        "children": children,
        "name": company_instance.name,
        "adm_unit": company_instance.administrative_unit_id,
        "is_auth": company_instance.is_authorized_organization
    }


def get_company_dict_from_cache(cache):
    """Создает словарь с аттрибутами из кеша редиса. Десериализует.

    :param cache: Кэш, колченный из редиса

    :return Словарь с данными

    """
    p_id = cache["parent_id"]
    # редис хранит boolean как строку 'False' или 'True'
    # Аналогично и с None
    auth = True if cache["is_auth"] == "True" else False
    _dict = {
        "id": int(cache["id"]),
        "parent_id": int(p_id) if p_id != "None" else None,
        "children": ujson.loads(cache["children"]) if cache["children"] else [],
        "name": cache["name"],
        "administrative_unit_id": int(cache["adm_unit"]),
        "is_authorized_organization": auth
    }
    return _dict


def get_company_node(company_id, company_unit_id):
    """Получить ноду учреждения из кеша.

    :param company_id: айди учреждения.
    :param company_unit_id: айди юнита учреждения.

    :return Ноду учреждения - словарь.

    """
    key = make_company_key(company_id, company_unit_id)
    cache = redis_client.hgetall(key)
    return get_company_dict_from_cache(cache)


def get_companies_nodes(companies_id, unit_id):
    """Возвращает список нод учреждений.

    :param companies_id: Список айди учреждений.
    :param unit_id: айди юнита

    :return Список содержащий ноды учреждений.

    """
    keys = []
    nodes = []
    for company_id in companies_id:
        keys.append(make_company_key(company_id, unit_id))
    pipe = redis_client.pipeline()
    for key in keys:
        pipe.hgetall(key)
    for result in pipe.execute():
        nodes.append(get_company_dict_from_cache(result))
    return nodes


def add_company_node(company_instance):
    """Создает ноду учреждения в кэше редиса.

    :param company_instance: инстанс учреждения

    :return результат выполнения команды редиса hmset

    """
    key = make_company_key(
        company_instance.id, company_instance.administrative_unit_id)
    return redis_client.hmset(key, get_company_hash(company_instance))


def delete_company_node(company_instance):
    """Удаляет ноду учреждения из кэша редиса.

    :param company_instance: инстанс учреждения

    :return True

    """
    company_key = make_company_key(
            company_instance.id, company_instance.administrative_unit_id)
    pipe = redis_client.pipeline()
    id_to_delete = get_company_node_descendants(
            company_instance.pk, company_instance.administrative_unit_id)
    for cid in id_to_delete:
        key = make_company_key(cid, company_instance.administrative_unit_id)
        pipe.delete(key)
    pipe.delete(company_key)
    parent = company_instance.parent
    if parent:
        parent_key = make_company_key(parent.id, parent.administrative_unit_id)
        pipe.delete(parent_key)
        pipe.hmset(parent_key, get_company_hash(parent))
    pipe.execute()
    set_unit_companies_cache(company_instance.administrative_unit_id)
    return True


def update_company_node(company_instance):
    """Пересоздает ноду учреждения в кэше редиса

    :param company_instance: инстанс учреждения

    :return None

    """
    pipe = redis_client.pipeline()
    key = make_company_key(
        company_instance.id, company_instance.administrative_unit_id)
    pipe.delete(key)
    pipe.hmset(key, get_company_hash(company_instance))
    pipe.execute()
    return None


def get_company_node_descendants(company_id, company_unit_id):
    """Возвращает список потомков учреждения.

    :param company_id: айди учреждения
    :param company_unit_id: айди юнита учреждения

    :return генератор содержащий результат выборки

    """
    key = make_company_key(company_id, company_unit_id)
    node = redis_client.hgetall(key)
    # стек заполняем первичными чилдами
    stack = ujson.loads(node["children"])
    # микро-оптимизация доступа
    pop = stack.pop
    extend = stack.extend
    # пока у нас что-то есть в стеке
    while stack:
        # получаем из стека айди чилда
        child_id = pop()
        # отдаем его через yield
        yield child_id
        key = COMPANY_KEY_TEMPLATE.format(company_unit_id, child_id)
        child_list = ujson.loads(redis_client.hget(key, "children") or "[]")
        # заполняем стек чилдами чилда
        extend(child_list)


def move_company_node(
        company_instance, company_from_instance, company_to_instance,
        from_unit_id):
    """Воспроизводит алгоритм перемещения ноды учреждения для кэша редиса.

    :param company_instance: инстанс учреждения кот. перенесли
    :param company_from_instance: инстанс учреждения откуда перенесли
    :param company_to_instance: инстанс учреждения куда перенесли
    :param from_unit_id: айди юнита из которого переносят.
           Передается. т.к. может отсутствовать company_from_instance.

    :return None

    """
    key = make_company_key(company_instance.id, from_unit_id)
    to_unit_id = company_to_instance.administrative_unit_id
    new_key = make_company_key(company_instance.id, to_unit_id)
    to_key = make_company_key(company_to_instance.id, to_unit_id)
    pipe = redis_client.pipeline()
    if company_from_instance:
        from_key = make_company_key(company_from_instance.id, from_unit_id)
        pipe.delete(from_key)
        pipe.hmset(from_key, get_company_hash(company_from_instance))
    # from_unit_id потому, что учреждение может быть уже перенесено под
    # другой юнит
    pipe.delete(key)
    pipe.delete(to_key)
    pipe.hmset(new_key, get_company_hash(company_instance))
    pipe.hmset(to_key, get_company_hash(company_to_instance))
    # если у нас изменился юнит
    # здесь необходимо инвалидировать кеш для всех юнитов (их чилдов)
    # так же инвалидировать ключи всех потомков перемещенного учреждения
    if from_unit_id != to_unit_id:
        # у нас же остались старые ключи, так что используется from_unit_id
        ids = get_company_node_descendants(company_instance.id, from_unit_id)
        for c_id in ids:
            # переименовываем все ключи, кот у нас были в потомках
            child_key = make_company_key(c_id, from_unit_id)
            pipe.hset(child_key, "adm_unit", to_unit_id)
            pipe.rename(child_key, make_company_key(c_id, to_unit_id))
        # изменяем список прямых чилдов юнита откуда перекинули учреждение
        set_unit_companies_cache(from_unit_id)
        # изменяем список прямых чилдов юнита куда перекинули учреждение
        set_unit_companies_cache(to_unit_id)
    pipe.execute()
    return None
	# coding: utf-8
	"""
	Стояла задача - реализовать дерево гос. учреждений для РФ.
	Иметь возможность перемещать учреждения и получать актуальные поддеревья(списки)
	дочерних учреждение в режиме "реального времени".

	А так же обеспечить кэш дерева.

	Учреждение может быть привязано к субъекту РФ и к другому учреждению.
	Глубина дерева может быть бесконечно большой. т.е. несколько десятков тысяч
	учреждений.

	Модель (упрощенная):
	class Company:
	administrative_unit_id: ID субъекта.
	parent_id: ID вышестоящего Company. Может быть null.

	Задача решалась в несколько этапов.

	Первый этап был решением через django-ORM(проект написан на django)
	и простом юнионом кверисетов рекурсивно.
	Проработало это где-то пару месяцев пока не появилась более-менее настоящая
	нагрузка и все умирало.

	Вторым этапом было решено взять django-mptt
	(https://github.com/django-mptt/django-mptt).
	Это "полноценная" Реализация дерева на SQL с своими менеджероми,
	пасьянсом и поэтесами.
	Это проработало где-то полтора месяца и вылезли баги:
	Деревья очень долго перестраивались (постгрес вообще мог отвалиться в
	не-операбельный режим). Постоянно множились ошибки из-за которых приходилось
	пересобирать дерево и система в эти моменты была недоступа.
	В общем - было не сильно лучше.

	Далее мы подумали и решили, что можно это дерево разложить в плоской структуре
	и собирать его онлайн. И использовать то, что уже есть в стеке - редис.
	Сначала это показалось не лучшим решением на фоне потребляемой памяти и
	скорости работы питона - но в итоге это было неплохо.
	Все дерево по РФ можно собрать где-то за 5секунд.
	Для "большого" мира софта это много, но для нашей задачи это было приемлимо.
	Тем более - сбор всего дерева РФ очень редкая операция.

	В ходе разработки были произведены некоторые оптимизации(или микро-оптимизации).
	В общем - это работает и нас устраивает то, как это работает сейчас.

	Пока производительности хватает, но есть идеи для перехода на Neo4j (
	https://ru.wikipedia.org/wiki/Neo4j).
	Не переходили изначально из-за того, что не хотелось увеличивать стек технологий
	для поддержки. ну и конечно перед переходом надо будет все это профилировать,
	тестировать.

	Текущая статистика по редису:

	uptime_in_days:53 (Аптайм маленький из-за недавних ребутов серверов.
	Сервера стоят не на нашей площадке)

	# Stats
	total_connections_received:15002
	total_commands_processed:357683126
	instantaneous_ops_per_sec:80
	rejected_connections:0
	sync_full:0
	sync_partial_ok:0
	sync_partial_err:0
	expired_keys:0
	evicted_keys:0
	keyspace_hits:350306471
	keyspace_misses:71
	pubsub_channels:0
	pubsub_patterns:0
	latest_fork_usec:0

	# Memory
	used_memory:135802672
	used_memory_human:129.51M
	used_memory_rss:154419200
	used_memory_peak:138898064
	used_memory_peak_human:132.46M
	used_memory_lua:33792

	# Keyspace
	db0:keys=125378,expires=0,avg_ttl=0, из них только 95 - Субъекты.

	Дерево меняется несколько сотен раз за сутки (добавление, удаление и
	перемещение нод).

	Для пущей производительности был отключен дамп базы редиса в файл, что
	позволило избежать Timeout Error'ов. Если что-то упадет - все может
	относителньо быстро (~2 минуты) собраться опять. В общем - это было не критично.

	"""
	from __future__ import absolute_import, unicode_literals

	# third party
	import ujson

	# project
	from cache.administrative_units_cache import set_unit_companies_cache
	from cache import TREE_CLIENT as redis_client

	# {administrative_unit_id}:{company_id}
	COMPANY_KEY_TEMPLATE = "cmp:{0}:{1}"


	def make_company_key(company_id, company_unit_id):
	"""Создаем ключ для редиса на основе COMPANY_KEY_TEMPLATE.

	:param company_id: айди учреждения
	:param company_unit_id: айди юнита учреждения.

	:return Созданный ключ.

	"""
	return COMPANY_KEY_TEMPLATE.format(company_unit_id, company_id)


	def get_company_hash(company_instance):
	"""Генерирует хэш для записи в редис.

	:param company_instance: Инстанс учреждения для записи

	:return Словарь с данными по учреждению для записи в редис

	"""
	children = [
	cid for cid in company_instance.children.values_list("id", flat=True)
	]
	return {
	"id": company_instance.id,
	"parent_id": company_instance.parent_id,
	"children": children,
	"name": company_instance.name,
	"adm_unit": company_instance.administrative_unit_id,
	"is_auth": company_instance.is_authorized_organization
	}


	def get_company_dict_from_cache(cache):
	"""Создает словарь с аттрибутами из кеша редиса. Десериализует.

	:param cache: Кэш, колченный из редиса

	:return Словарь с данными

	"""
	p_id = cache["parent_id"]
	# редис хранит boolean как строку 'False' или 'True'
	# Аналогично и с None
	auth = True if cache["is_auth"] == "True" else False
	_dict = {
	"id": int(cache["id"]),
	"parent_id": int(p_id) if p_id != "None" else None,
	"children": ujson.loads(cache["children"]) if cache["children"] else [],
	"name": cache["name"],
	"administrative_unit_id": int(cache["adm_unit"]),
	"is_authorized_organization": auth
	}
	return _dict


	def get_company_node(company_id, company_unit_id):
	"""Получить ноду учреждения из кеша.

	:param company_id: айди учреждения.
	:param company_unit_id: айди юнита учреждения.

	:return Ноду учреждения - словарь.

	"""
	key = make_company_key(company_id, company_unit_id)
	cache = redis_client.hgetall(key)
	return get_company_dict_from_cache(cache)


	def get_companies_nodes(companies_id, unit_id):
	"""Возвращает список нод учреждений.

	:param companies_id: Список айди учреждений.
	:param unit_id: айди юнита

	:return Список содержащий ноды учреждений.

	"""
	keys = []
	nodes = []
	for company_id in companies_id:
	keys.append(make_company_key(company_id, unit_id))
	pipe = redis_client.pipeline()
	for key in keys:
	pipe.hgetall(key)
	for result in pipe.execute():
	nodes.append(get_company_dict_from_cache(result))
	return nodes


	def add_company_node(company_instance):
	"""Создает ноду учреждения в кэше редиса.

	:param company_instance: инстанс учреждения

	:return результат выполнения команды редиса hmset

	"""
	key = make_company_key(
	company_instance.id, company_instance.administrative_unit_id)
	return redis_client.hmset(key, get_company_hash(company_instance))


	def delete_company_node(company_instance):
	"""Удаляет ноду учреждения из кэша редиса.

	:param company_instance: инстанс учреждения

	:return True

	"""
	company_key = make_company_key(
	company_instance.id, company_instance.administrative_unit_id)
	pipe = redis_client.pipeline()
	id_to_delete = get_company_node_descendants(
	company_instance.pk, company_instance.administrative_unit_id)
	for cid in id_to_delete:
	key = make_company_key(cid, company_instance.administrative_unit_id)
	pipe.delete(key)
	pipe.delete(company_key)
	parent = company_instance.parent
	if parent:
	parent_key = make_company_key(parent.id, parent.administrative_unit_id)
	pipe.delete(parent_key)
	pipe.hmset(parent_key, get_company_hash(parent))
	pipe.execute()
	set_unit_companies_cache(company_instance.administrative_unit_id)
	return True


	def update_company_node(company_instance):
	"""Пересоздает ноду учреждения в кэше редиса

	:param company_instance: инстанс учреждения

	:return None

	"""
	pipe = redis_client.pipeline()
	key = make_company_key(
	company_instance.id, company_instance.administrative_unit_id)
	pipe.delete(key)
	pipe.hmset(key, get_company_hash(company_instance))
	pipe.execute()
	return None


	def get_company_node_descendants(company_id, company_unit_id):
	"""Возвращает список потомков учреждения.

	:param company_id: айди учреждения
	:param company_unit_id: айди юнита учреждения

	:return генератор содержащий результат выборки

	"""
	key = make_company_key(company_id, company_unit_id)
	node = redis_client.hgetall(key)
	# стек заполняем первичными чилдами
	stack = ujson.loads(node["children"])
	# микро-оптимизация доступа
	pop = stack.pop
	extend = stack.extend
	# пока у нас что-то есть в стеке
	while stack:
	# получаем из стека айди чилда
	child_id = pop()
	# отдаем его через yield
	yield child_id
	key = COMPANY_KEY_TEMPLATE.format(company_unit_id, child_id)
	child_list = ujson.loads(redis_client.hget(key, "children") or "[]")
	# заполняем стек чилдами чилда
	extend(child_list)


	def move_company_node(
	company_instance, company_from_instance, company_to_instance,
	from_unit_id):
	"""Воспроизводит алгоритм перемещения ноды учреждения для кэша редиса.

	:param company_instance: инстанс учреждения кот. перенесли
	:param company_from_instance: инстанс учреждения откуда перенесли
	:param company_to_instance: инстанс учреждения куда перенесли
	:param from_unit_id: айди юнита из которого переносят.
	Передается. т.к. может отсутствовать company_from_instance.

	:return None

	"""
	key = make_company_key(company_instance.id, from_unit_id)
	to_unit_id = company_to_instance.administrative_unit_id
	new_key = make_company_key(company_instance.id, to_unit_id)
	to_key = make_company_key(company_to_instance.id, to_unit_id)
	pipe = redis_client.pipeline()
	if company_from_instance:
	from_key = make_company_key(company_from_instance.id, from_unit_id)
	pipe.delete(from_key)
	pipe.hmset(from_key, get_company_hash(company_from_instance))
	# from_unit_id потому, что учреждение может быть уже перенесено под
	# другой юнит
	pipe.delete(key)
	pipe.delete(to_key)
	pipe.hmset(new_key, get_company_hash(company_instance))
	pipe.hmset(to_key, get_company_hash(company_to_instance))
	# если у нас изменился юнит
	# здесь необходимо инвалидировать кеш для всех юнитов (их чилдов)
	# так же инвалидировать ключи всех потомков перемещенного учреждения
	if from_unit_id != to_unit_id:
	# у нас же остались старые ключи, так что используется from_unit_id
	ids = get_company_node_descendants(company_instance.id, from_unit_id)
	for c_id in ids:
	# переименовываем все ключи, кот у нас были в потомках
	child_key = make_company_key(c_id, from_unit_id)
	pipe.hset(child_key, "adm_unit", to_unit_id)
	pipe.rename(child_key, make_company_key(c_id, to_unit_id))
	# изменяем список прямых чилдов юнита откуда перекинули учреждение
	set_unit_companies_cache(from_unit_id)
	# изменяем список прямых чилдов юнита куда перекинули учреждение
	set_unit_companies_cache(to_unit_id)
	pipe.execute()
	return None