Organizando e iniciando novo populate em comandos

Author: Winzen

backend/apps/core/management/commands/constants.py

@@ -0,0 +1,10 @@
+# -*- coding: utf-8 -*-
+from enum import Enum
+from pathlib import Path
+
+
+class LocalPolulate(Enum):
+    FOLDER = "backups"
+    FILE_NAME = "backup_prod_v1.json"
+    URL_JSON = "https://storage.googleapis.com/basedosdados-dev/backend_backup/backup_prod_v1.json"
+    PATH_FILE = Path(FOLDER) / FILE_NAME

backend/apps/core/management/commands/dump_v1.py

@@ -0,0 +1,23 @@
+# -*- coding: utf-8 -*-
+
+from django.core.management import call_command
+from django.core.management.base import BaseCommand
+from loguru import logger
+
+from .constants import LocalPolulate
+from .utils import remove_connection_accounts
+
+
+class Command(BaseCommand):
+    help = "Dump APP v1, avoiding APP accout"
+
+    def handle(self, *args, **options) -> None:
+        LocalPolulate.PATH_FILE.value.parent.mkdir(parents=True, exist_ok=True)
+
+        logger.info("Fazendo Dump")
+        call_command("dumpdata", "v1", indent=2, output=str(LocalPolulate.PATH_FILE.value))
+
+        logger.info("Removendo conexão com dos dados de V1 com Account")
+        remove_connection_accounts(LocalPolulate.PATH_FILE.value)
+
+        logger.info("Processo concluído com sucesso!")

backend/apps/core/management/commands/local_populate.py

@@ -0,0 +1,28 @@
+# -*- coding: utf-8 -*-
+from django.core.management import call_command
+from django.core.management.base import BaseCommand
+from loguru import logger
+
+from .constants import LocalPolulate
+from .utils import download_backup
+
+
+class Command(BaseCommand):
+    help = "Populate local database with initial data from prod"
+
+    def handle(self, *args, **options) -> None:
+        download_backup(LocalPolulate.URL_JSON.value, LocalPolulate.PATH_FILE.value)
+
+        logger.info("Purge previous database if exists")
+        call_command("flush", interactive=False)
+
+        logger.info("Migrate development database")
+        call_command("migrate")
+
+        logger.info("Load data!")
+        call_command("loaddata", str(LocalPolulate.PATH_FILE.value))
+
+        logger.info("Atualizar index!")
+        call_command("update_index")
+
+        logger.info("Processo concluído com sucesso!")

backend/apps/core/management/commands/utils.py

@@ -0,0 +1,76 @@
+# -*- coding: utf-8 -*-
+import json
+from pathlib import Path
+
+import requests
+
+
+def download_backup(url_arquivo: str, caminho_salvar: str) -> None:
+    """
+    Baixa um arquivo JSON grande usando chunks para economia de memória.
+
+    Args:
+        url_arquivo: URL do arquivo JSON
+        caminho_salvar: Caminho onde salvar o arquivo
+    """
+    # Configurações para controle de memória
+    tamanho_chunk = 1024 * 1024  # 1MB por chunk
+
+    try:
+        resposta = requests.get(url_arquivo, stream=True)
+
+        tamanho_total = int(resposta.headers.get("content-length", 0))
+
+        caminho_arquivo = Path(caminho_salvar)
+        caminho_arquivo.parent.mkdir(parents=True, exist_ok=True)
+
+        with open(caminho_salvar, "wb") as arquivo:
+            bytes_baixados = 0
+
+            for chunk in resposta.iter_content(chunk_size=tamanho_chunk):
+                if chunk:
+                    arquivo.write(chunk)
+                    bytes_baixados += len(chunk)
+
+                    print(
+                        f"\rBaixando: {bytes_baixados/tamanho_total*100:.2f}%", end="", flush=True
+                    )
+
+        print("\n\nDownload concluído!")
+
+    except Exception as e:
+        print(f"Erro ao baixar o arquivo: {str(e)}")
+
+
+def remove_connection_accounts(path: str) -> None:
+    """
+    Remove qualquer conexão com que o JSON gerado do APP v1 tenha com account
+
+    Args:
+        path: Caminho para o path JSON
+    """
+
+    try:
+        # Ler o path
+        with open(path, "r", encoding="utf-8") as json_file:
+            dados = json.load(json_file)
+
+        # Processar cada registro
+        for registro in dados:
+            if registro.get("model") == "v1.table":
+                registro.setdefault("fields", {}).update(
+                    {"published_by": [], "data_cleaned_by": []}
+                )
+            elif registro.get("model") == "v1.informationrequest":
+                registro.setdefault("fields", {})["started_by"] = None
+
+        # Salvar alterações
+        with open(path, "w", encoding="utf-8") as f:
+            json.dump(dados, f, indent=2, ensure_ascii=False)
+
+    except FileNotFoundError:
+        print(f"Erro: path '{path}' não encontrado")
+    except json.JSONDecodeError:
+        print(f"Erro: path '{path}' contém JSON inválido")
+    except Exception as e:
+        print(f"Erro inesperado ao processar '{path}': {str(e)}")