Source code for services.users.export

import concurrent.futures
from typing import TYPE_CHECKING

from django.apps import apps
from django.core import serializers
from orjson import orjson

from api.users.serializers import FollowingAndFollowersUserSerializer

if TYPE_CHECKING:
    from apps.users.models import User




class ExportUserData:
    @classmethod
    def _get_params(cls, model, user):
        params = {}
        if hasattr(model, "author"):
            params["author"] = user
        if hasattr(model, "is_deleted"):
            params["is_deleted"] = False
        if hasattr(model, "post"):
            params["post__is_deleted"] = False
        return params

    @classmethod
    def _serialize_batch(cls, model, params):
        qs = model.objects.filter(**params)
        if not qs.exists():
            return model.__name__, []
        serialized_data = serializers.serialize("python", qs)
        transformed_data = [{"id": obj["pk"], **obj["fields"]} for obj in serialized_data]
        return model.__name__, transformed_data



    @classmethod
    def export_data(cls, user: "User"):
        models = apps.get_models()
        data = {"User": FollowingAndFollowersUserSerializer(user).data}

        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = {
                executor.submit(cls._serialize_batch, model, cls._get_params(model, user)): model
                for model in models
                if hasattr(model, "author")
            }

            for future in concurrent.futures.as_completed(futures):
                model_name, result = future.result()
                if result:
                    data[model_name] = result

        return orjson.dumps(data).decode()