Python to Salesforce integration with SFDX CLI and subprocess

Our app purpose is salesforce metadata backup, so we need to implement retrieve metadata logic. In Salesforce we can work with metadata with Metadata API and one of possibility if to retrieve metadata with package.xml file that contains the description of the components we need. But we can simplify it a lot and don't need to use API directly. I plan to use SFDX CLI that offers us a lot of features that we can use with python subprocess lib.

More details regarding package.xml you can find here

https://developer.salesforce.com/docs/atlas.en-us.api_meta.meta/api_meta/manifest_samples.htm

There are different options to build this file. You can do it manually, or use one of automation tools (ex. https://packagebuilder.herokuapp.com/).

If you have your package.xml it is the time to add it to the app. We can store package.xml in Task level, so let's add new field to the Task model (with name package_xml). (You can reference this article to do DB migration - https://angular-python-salesforce.blogspot.com/2020/03/setup-db-postgres.html)

Next step is to add new logic to manage package.xml to frontend. Nothing new here. We just need to create new SLDS-Textarea component and wrap it to the modal. 

Now we have package.xml stored in regular text field in Task model. Before we start with regular backup logic, let's implement "Validation" for package.xml. And as this process can be too long and we can rich Heroku timeout limit we need to implement validation as async process. Validation can be done same way as regular Task Scheduled process and we can store results in TaskRun model. Let's do it this way.

./backend/controllers/task.py

@task_bp.route('/validate-package-xml', methods=['POST'])
@login_required
def validate_package_xml():

    task_dto = request.get_json()

    task = (Task.query.filter(Task.id == task_dto['id'],
                              Task.user_id == current_user.id)
                      .first_or_404())

    new_task_run = TaskRun()
    new_task_run.results = []
    new_task_run.started_at = datetime.now()
    new_task_run.status = 'queue'
    new_task_run.task_id = task.id

    db.session.add(new_task_run)
    db.session.commit()

    validate_package_xml_helper(task, new_task_run)

    return jsonify({'results': 'Validation started'})

./backend/scripts/task.py

def validate_package_xml_helper(task, task_run):
    if current_app.config['is_development_mode']:
        print('VALIDATE TASK LOCALY')
        import subprocess
        subprocess.Popen(['flask', 'task', 'validate_package_xml', str(task.id), str(task_run.id)])
    else:
        print('VALIDATE PACKAGE_XML IN PRODUCTION ...')


@task_scripts.cli.command("validate_package_xml")
@click.argument("task_id")
@click.argument("task_run_id")
def validate_package_xml(task_id, task_run_id):

    # EXAMPLE: flask task validate_package_xml <task_id> <task_run_id>

    print('PACKAGE.XML VALIDATOR')

    task = Task.query.filter_by(id=task_id).first()
    print('==== Task:')
    print(task)

    if task is None:
        sys.exit('Task not found')

    task_run = TaskRun.query.filter_by(id=task_run_id).first()
    print('==== TaskRun:')
    print(task_run)

    if task_run is None:
        sys.exit('TaskRun not found')

    task_run.status = 'processing'
    db.session.commit()

    try:
        with open('./sfdc/crds.txt', 'w') as f:
            f.write(f'force://{current_app.config["SFDC_CONNECTED_APP_KEY"]}:{current_app.config["SFDC_CONNECTED_APP_SECRET"]}:{task.sfdc["refresh_token"]}@{task.sfdc["instance_url"]}')  # NOQA

        with open('./sfdc/package.xml', 'w') as f:
            f.write(task.package_xml)  # NOQA

        import subprocess

        process = subprocess.Popen(['./bash_scripts/prepare_sfdc_project.sh'],
                                   stdout=subprocess.PIPE,
                                   stderr=subprocess.PIPE)
        out, err = process.communicate()
        errcode = process.returncode

        print('==============')
        print('ERRCODE:', errcode)
        print('STDOUT:', out)
        print('STDERR:', err)

        process = subprocess.Popen(['./bash_scripts/retrieve_sfdc.sh'],
                                   stdout=subprocess.PIPE,
                                   stderr=subprocess.PIPE)
        out, err = process.communicate()
        errcode = process.returncode

        print('==============')
        print('ERRCODE:', errcode)
        print('STDOUT:', out)
        print('STDERR:', err)

        try:
            out_ = json.loads(out)
        except:  # NOQA
            task_run.status = 'error'
            flag_modified(task_run, 'results')
            task_run.results = task_run.results.append(out)
            db.session.commit()
            return

        # errcode == 0
        if errcode == 0:
            if 'result' in out_ and out_['result']['success'] is True:
                task_run.status = 'success'
                task_run.results.append('*** Package.xml Validation: Success')
                if 'messages' in out_['result']:
                    for m in out_['result']['messages']:
                        if 'problem' in m:
                            task_run.results.append('WARNING: ' + m['problem'])
                        else:
                            task_run.results.append('INFO: ' + json.dumps(m))
            else:
                task_run.results.append('*** Package.xml Validation: Fail')
                task_run.status = 'fail'

        # errcode != 0
        else:
            task_run.status = 'error'
            if 'message' in out_:
                task_run.results.append(out_['message'])
            else:
                task_run.results = task_run.results.append(out)

        flag_modified(task_run, 'results')
        db.session.commit()

    except Exception as e:  # NOQA
        task_run.status = 'error'
        task_run.results.append(str(e))
        flag_modified(task_run, 'results')
        db.session.commit()

    print('PACKAGE.XML VALIDATOR FINISHED')

In these two methods there are a lot of magic that I need to explain.

- method validate_package_xml_helper is called from flask controller and it starts second method as flask script in separate process (in current implementation we support Development mode only and run bash script. In Production mode we will start separate Heroku Dyno, but I will implement it latter). We just run subprocess.Popen(...) so it doesn't block the thread and flask return response immediately.

- second method validate_package_xml works as flask script and has few important steps (I marked in different colors):

(first black) - just regular validation of input parameters. If Task and Task Run are available we change Task Run status to "processing" and run few bash scripts

(blue) - in special "sfdc" folder we need to create files for SFDC CLI. They are package.xml from Task and crds.txt file with auth data for org we want to connect. More information last file you can find here -  

https://developer.salesforce.com/docs/atlas.en-us.sfdx_cli_reference.meta/sfdx_cli_reference/cli_reference_force_auth.htm

(green) - we run prepare_sfdc_project.sh bash script to create empty sfdx project. It is required step to use sfdx cli.

cd ./sfdc

rm -rf ./sfdc_project

echo "*** Removed ./sfdc_project folder"

sfdx force:project:create --projectname sfdc_project --template standard

echo "*** Created ./sfdc_project project"

cd ./sfdc_project

sfdx force:auth:sfdxurl:store -f ../crds.txt -a M1

echo "*** Auth to SF with SfdxAuthUrl in crds.txt"

(orange) - we run retrieve_sfdc.sh bash script with sfdx command to retrieve metadata specified in package.xml and store it in ./mdapipkg folder of the empty project. 

#!/bin/bash

cd ./sfdc/sfdc_project

sfdx force:mdapi:retrieve -r ./mdapipkg -u M1 -k ../package.xml --json

(last black) - with magic parameter --json and process.communicate() we receive results in flask script method to parse it and write retrieve results to TaskRun record.

Last step - I we added new section to Task page with list of all TaskRuns and possibility to see TaskRun.results field in the modal.