MS Dynamics 365 F&O - Getting started in Peliqan

# Entity browser for Microsoft Dynamics 365 F&O (Finance & Operations).
# Use this to find entity names and their primary keys, to add additional entities to the pipeline.
# Contact support@peliqan.io for questions.

dynamicsax_api = pq.connect('MS Dynamics 365 Finance and Operations (AX)')

reload = False

def load_entities():
    st.text("Loading all entities from Dynamics F&O, this will take a minute or so.")
    entities = dynamicsax_api.get('entities')
    all_entities = []
    for entity in entities:
        entity_name = entity["@Name"]
        pks = entity.get("Key",{}).get("PropertyRef", [])
        if not isinstance(pks, list):
            pks = [pks]
            
        all_pks = []
        for pk in pks:
            pk_name = pk["@Name"]
            all_pks.append(pk_name)
        
        all_entities.append(
            {
                "name": entity_name,
                "primary keys": ", ".join(all_pks)
            }
        )
    all_entities = sorted(all_entities, key=lambda x: x["name"])
    return all_entities

all_entities = pq.get_state()
if not all_entities or reload:
    all_entities = load_entities()
    pq.set_state(all_entities)
#st.table(all_entities)

search = st.text_input("Search (matches on start of entity name only):")
if search:
    filtered_entities = [e for e in all_entities if e["name"].lower().startswith(search.lower())]
    selected_entity = st.selectbox("Entities (start typing to filter)", options = filtered_entities, format_func=lambda x: x["name"])
    if selected_entity:
        st.subheader("Entity name")
        st.write(selected_entity["name"])
        st.subheader("Entity primary keys")
        st.write(selected_entity["primary keys"])

# Bulk export from Microsoft Dynamics 365 F&O Finance & Operations, and import into the Peliqan data warehouse.
# This script will fetch all entities from Dynamics F&O with their PKs on the first run (metadata).
#
# Create a DMF Export project in F&O first (DMF = Data Management Framework).
# Use Source data format "XML-Element".
#
# Use list_projects() to find correct name of Export DMF Project in Dynamics F&O.
# Set correct project name and legal entity below:

export_project_name = "My export 2"
legal_entity_name = "DAT"
export_format = "XML" # or "CSV"
export_schema = "dynamics f&o bulk export"

###########################

import time
import zipfile
import io
import os
import requests
import pandas as pd
import re
import xml.etree.ElementTree as ET
from typing import Iterator, Dict, List


dynamicsax_api = pq.connect('MS Dynamics 365 Finance and Operations (AX)')
dbconn = pq.dbconnect(pq.DW_NAME)

extract_dir = f"/etc/shared_temp/dynamics_exported_data_{pq.INTERFACE_ID}"
reload_entities_from_api = False
start_new_export_run = True # Set to false to import already downloaded files


def main():
    st.title("Microsoft Dynamics 365 F&O Bulk Export")

    if start_new_export_run:
        execution_id = start_export_run(export_project_name, legal_entity_name)
    
        execution_status = ""
        while not "Succeeded" in execution_status: # "Succeeded" or "PartiallySucceeded"
            time.sleep(10)
            execution_status = get_export_run_status(execution_id)
    
        export_url = get_export_url(execution_id)
        download_files(export_url)

    if export_format == "CSV":
        import_data_csv()
    elif export_format == "XML":
        import_data_xml()
    else:
        st.warning("Unknown export format")

def load_entities_pks():
    pks_all_entities = pq.get_state()
    if not pks_all_entities or reload_entities_from_api:
        st.text("Loading and caching all entities with their PKs from Dynamics F&O, this will take a minute or so.")
        
        def make_plural(word):
            match = re.search(r"(.*?)(V[1-4])$", word) # Special case: entities ending with V1, V2, V3, V4
            if match:
                base, version = match.groups()
                return make_plural(base) + version
            lower_word = word.lower()
            if lower_word == "person":
                return "people"
            if lower_word.endswith(("s", "sh", "ch", "x", "z")):
                return word + "es"
            if lower_word.endswith("y") and len(word) > 1 and word[-2].lower() not in "aeiou":
                return word[:-1] + "ies"
            if lower_word.endswith("f"):
                return word[:-1] + "ves"
            if lower_word.endswith("fe"):
                return word[:-2] + "ves"
            return word + "s"
    
        entities = dynamicsax_api.get('entities')
        pks_all_entities = {}
        for entity in entities:
            entity_name = entity["@Name"]
            entity_name_plural = make_plural(entity_name)
            pks = entity.get("Key",{}).get("PropertyRef", [])
            if not isinstance(pks, list):
                pks = [pks]
            all_pks = []
            for pk in pks:
                pk_name = pk["@Name"]
                if pk_name.upper() not in ["DATAAREAID"]:
                    all_pks.append(pk_name.upper())
            pks_all_entities[entity_name_plural.lower()] = all_pks
        pq.set_state(pks_all_entities)
    
    return pks_all_entities


def list_projects():
    projects = dynamicsax_api.get('dmfprojects')
    export_projects = []
    for project in projects:
        if project["OperationType"] == "Export":
            export_projects.append(project["Name"])
    st.table(export_projects)


def start_export_run(export_project_name, legal_entity_name):
    params = {
        'reExecute': False,
        'executionId': "",
        'packageName': "MyExportPackage.zip",
        'legalEntityId': legal_entity_name,
        'definitionGroupId': export_project_name
    }
    result = dynamicsax_api.add('exportrun', params)
    if result["status"] == "success":
        
        execution_id = result["detail"]["value"]
        st.write(f"Export run started, execution id: {execution_id}")
        return execution_id
    else:
        st.warning("Could not start export run")
        st.json(result)
        exit()


def get_export_run_status(execution_id):
    params = {
        'executionId': execution_id
    }
    execution_status = dynamicsax_api.get('exportstatus', params)
    st.write(f"Execution status: {execution_status}")
    return execution_status


def get_export_url(execution_id):
    params = {
        'executionId': execution_id
    }
    export_url = dynamicsax_api.get('exportdownloadurl', params)
    st.text(f"Export URL: {export_url}")
    return export_url


def download_files(url):
    os.makedirs(extract_dir, exist_ok=True)
    
    response = requests.get(url)
    response.raise_for_status()
    
    with zipfile.ZipFile(io.BytesIO(response.content)) as zf:
        zf.extractall(extract_dir)
    
    st.write(f"Files extracted to: {os.path.abspath(extract_dir)}")


def import_data_csv():
    pks_all_entities = load_entities_pks()
    for filename in os.listdir(extract_dir):
        st.subheader(f"File: {filename}")
        file_path = os.path.join(extract_dir, filename)
        if filename.lower().endswith(".csv"):
            df = pd.read_csv(file_path)
            st.write(df.head())
            if len(df)>0:
                entity_name_plural = filename.replace(".csv", "").replace(" ", "") # e.g. CustomersV3
                table_name = filename.replace(".csv", "").lower().replace(" ", "_")
                if entity_name_plural.lower() in pks_all_entities:
                    pks = pks_all_entities[entity_name_plural.lower()]
                    st.info(f"Importing data from file: entity name {entity_name_plural}, PKs: {pks}")
                    result = dbconn.write(export_schema, table_name, records = df, pk = pks)
                    status = result["status"]
                    if status.lower() != "success":
                        st.error("Error writing to DWH")
                        st.write(result)
                else:
                    st.error(f"Cannot find PKs of entity {entity_name_plural}, skipping import of this file.")
            else:
                st.warning(f"No data in file, skipping.")
        else:
            st.warning(f"Not a CSV data file, skipping.")
        os.remove(file_path)


def read_xml_in_chunks(file_path: str, chunk_size: int = 1000) -> Iterator[List[Dict]]:
    # Processes all direct children of the Document root tag.
    context = ET.iterparse(file_path, events=('start', 'end'))
    chunk = []
    depth = 0
    current_record = None
    for event, elem in context:
        if event == 'start':
            depth += 1
            # Depth 2 means direct child of Document (depth 1)
            if depth == 2:
                current_record = elem
        elif event == 'end':
            # Process complete records (direct children of Document)
            if depth == 2 and current_record is not None:
                # Convert XML element to dictionary
                row_dict = {}
                # Add attributes if any
                if elem.attrib:
                    row_dict.update(elem.attrib)
                # Add child elements
                for child in elem:
                    if child.text and child.text.strip():
                        row_dict[child.tag] = child.text.strip()
                    elif len(child) > 0:
                        # Handle nested elements
                        for nested in child:
                            if nested.text and nested.text.strip():
                                row_dict[f"{child.tag}_{nested.tag}"] = nested.text.strip()
                
                # Add direct text if element has no children
                if len(elem) == 0 and elem.text and elem.text.strip():
                    row_dict['value'] = elem.text.strip()
                if row_dict:  # Only add non-empty records
                    chunk.append(row_dict)
                # Clear element to free memory
                elem.clear()
                current_record = None
                # Yield chunk when it reaches the desired size
                if len(chunk) >= chunk_size:
                    yield chunk
                    chunk = []
            depth -= 1
    # Yield remaining records
    if chunk:
        yield chunk
    # Clean up
    del context


def import_data_xml():
    chunk_size = 1000
    pks_all_entities = load_entities_pks()
    
    for filename in os.listdir(extract_dir):
        st.subheader(f"File: {filename}")
        file_path = os.path.join(extract_dir, filename)

        if filename == "Manifest.xml" or filename == "PackageHeader.xml":
            pass  
        elif filename.lower().endswith(".xml"):
            try:
                # Process first chunk to show preview
                total_records = 0
                
                entity_name_plural = filename.replace(".xml", "").replace(" ", "")
                table_name = filename.replace(".xml", "").lower().replace(" ", "_")
                
                if entity_name_plural.lower() not in pks_all_entities:
                    st.error(f"Cannot find PKs of entity {entity_name_plural}, skipping import of this file.")
                    continue
                
                pks = pks_all_entities[entity_name_plural.lower()]
                st.info(f"Importing data from file: entity name {entity_name_plural}, PKs: {pks}")
                
                # Process file in chunks
                for chunk_idx, chunk in enumerate(read_xml_in_chunks(file_path, chunk_size)):
                    if chunk_idx == 0:
                        # Show preview of first chunk
                        st.write(f"Preview (first {min(5, len(chunk))} records):")
                        st.write(chunk[:5])
                    
                    total_records += len(chunk)
                    
                    # Convert chunk to DataFrame for writing
                    df_chunk = pd.DataFrame(chunk)
                    result = dbconn.write(export_schema, table_name, records=df_chunk, pk=pks)
                    
                    status = result["status"]
                    if status.lower() != "success":
                        st.error(f"Error writing chunk {chunk_idx + 1} to DWH")
                        st.write(result)
                        break
                    
                    # Show progress
                    st.write(f"Processed chunk {chunk_idx + 1}: {len(chunk)} records")
                
                if total_records == 0:
                    st.warning(f"No data in file, skipping.")
                else:
                    st.success(f"Successfully imported {total_records} total records")
                    
            except ET.ParseError as e:
                st.error(f"Error parsing XML file: {e}")
            except Exception as e:
                st.error(f"Error processing file: {e}")
        else:
            st.warning(f"Not an XML data file, skipping.")
        
        try:
            os.remove(file_path)
        except Exception as e:
            st.warning(f"Could not remove file: {e}")


main()