Automatically sync Custom Fields and Property Setters in Frappe sites
import os

import frappe
from frappe.modules.import_file import import_file_by_path

from sync_customizations.utils import (
    write_document_file,
    delete_document_file,
    get_folder,
    get_filters,
    get_doctype_field,
    get_file_name,
)
from sync_customizations.constants import DOCTYPES_TO_SYNC


IMPORT_FLAG = "sc_importing_customizations"


def export_doc(doc, method=None):
    if (
        doc.doctype == "Custom Field" and doc.is_system_generated
    ) or frappe.local.flags.get(IMPORT_FLAG, False):
        return

    frappe.db.after_commit.add(lambda: write_document_file(doc))


def delete_doc(doc, method=None):
    frappe.db.after_commit.add(lambda: delete_document_file(doc))


def on_save_customization(doc, method=None):
    frappe.db.after_commit.add(
        lambda: export_customizations(filter_by_doctype=doc.doc_type)
    )


def export_customizations(filter_by_doctype=None):
    for doctype in DOCTYPES_TO_SYNC:
        folder = get_folder(doctype)
        filters = get_filters(doctype)
        files_to_keep = set(
            get_file_name(doc.name) for doc in frappe.get_all(doctype, filters=filters)
        )
        files_to_delete = [
            fname for fname in os.listdir(folder) if fname not in files_to_keep
        ]

        for file_name in files_to_delete:
            os.remove(os.path.join(folder, file_name))

        if filter_by_doctype:
            doctype_field = get_doctype_field(doctype)
            filters[doctype_field] = filter_by_doctype

        for doc in frappe.get_all(doctype, filters=filters, fields="*"):
            doc = frappe.get_doc({"doctype": doctype, **doc})
            write_document_file(doc)


def import_customizations():
    flags = frappe.local.flags
    try:
        flags[IMPORT_FLAG] = True
        _import_customizations()

    finally:
        flags.pop(IMPORT_FLAG, None)


def _import_customizations():
    for doctype in DOCTYPES_TO_SYNC:
        docnames = set()

        folder = get_folder(doctype)

        for doc_filename in os.listdir(folder):
            if not doc_filename.endswith(".json"):
                continue

            import_file_by_path(os.path.join(folder, doc_filename))
            docnames.add(os.path.splitext(doc_filename)[0])

        frappe.db.commit()
        print(f"{doctype}s imported successfully...")

        docs_to_delete = frappe.get_all(
            doctype,
            filters={"name": ("not in", docnames), **get_filters(doctype)},
            pluck="name",
        )
        if docs_to_delete:
            print(f"Detected stale {doctype}s:\n{', '.join(docs_to_delete)}")
            print("\nDeleting...", end=" ")

            frappe.db.delete(doctype, {"name": ("in", docs_to_delete)})
            frappe.db.commit()
            print("Done!\n")

    frappe.clear_cache()

sync_customizations/api.py
resilient-tech/sync-customizations
import click

import frappe
from frappe.commands import get_site, pass_context


@click.command("sc-export")
@pass_context
def export_customizations(context):
    """Export supported customizations to JSON files"""

    from sync_customizations.api import export_customizations

    site = get_site(context)
    frappe.init(site=site)
    frappe.connect()

    export_customizations()
    print("Customizations exported successfully!")


commands = [export_customizations]

sync_customizations/commands.py
resilient-tech/sync-customizations
EXPORT_DIR_NAME = "customizations"
DOCTYPES_TO_SYNC = ("Custom Field", "Property Setter")

sync_customizations/constants.py
resilient-tech/sync-customizations
from sync_customizations.constants import DOCTYPES_TO_SYNC

app_name = "sync_customizations"
app_title = "Sync Customizations"
app_publisher = "Resilient Tech"
app_description = (
    "Automatically sync Custom Fields and Property Setters in Frappe sites"
)
app_email = "admin@resilient.tech"
app_license = "mit"

before_migrate = "sync_customizations.api.import_customizations"

doc_events = {
    DOCTYPES_TO_SYNC: {
        "on_update": "sync_customizations.api.export_doc",
        "on_trash": "sync_customizations.api.delete_doc",
    },
    "Customize Form": {
        "save_customization": "sync_customizations.api.on_save_customization",
    },
}

sync_customizations/hooks.py
resilient-tech/sync-customizations
import os

import frappe
from frappe.modules.export_file import strip_default_fields
from frappe.utils import get_bench_path

from sync_customizations.constants import EXPORT_DIR_NAME


def write_document_file(doc):
    doc_export = doc.as_dict(no_nulls=True)
    for key in ("idx", "__unsaved"):
        doc_export.pop(key, None)

    doc.run_method("before_export", doc_export)
    doc_export = strip_default_fields(doc, doc_export)

    with open(get_file_path(doc), "w+") as txtfile:
        txtfile.write(frappe.as_json(doc_export))


def get_filters(doctype):
    return {"is_system_generated": 0} if doctype == "Custom Field" else {}


def get_doctype_field(doctype):
    return "dt" if doctype == "Custom Field" else "doc_type"


def delete_document_file(doc):
    try:
        os.remove(get_file_path(doc))
    except OSError:
        pass


def get_export_dir_path():
    if frappe.conf.customizations_dir:
        return os.path.join(get_bench_path(), frappe.conf.customizations_dir)

    return frappe.get_site_path(EXPORT_DIR_NAME)

def get_folder(doctype):
    folder_path = os.path.join(get_export_dir_path(), frappe.scrub(doctype))
    frappe.create_folder(folder_path)
    return folder_path


def get_file_path(doc):
    return os.path.join(get_folder(doc.doctype), get_file_name(doc.name))


def get_file_name(docname):
    return f"{docname}.json"

sync_customizations/utils.py
resilient-tech/sync-customizations