Salesforce Integration Flow
Available in v44 and later. On earlier releases, the approach explained in section Before v44 does the same job without the output block.
The Salesforce Notifier output block does not write to Salesforce itself. For every payload of a completed submission it runs a flow you write, passing down the Salesforce credentials, the submission id and a reference to the rendered payload. Everything that happens in Salesforce is therefore ordinary Python in a flow you own: writing several objects, a related record, an upsert, or a value computed from two fields is a matter of editing that code, and so is deciding what counts as a duplicate or when to write nothing at all.
In v44 the picker lists this one as Salesforce Notifier Output Block, and the block that came before it as Salesforce Notifier (Deprecated) Output Block, which is removed in v46.0. Their identifiers are unchanged, so existing flows keep working until then.
The example
The sample is the integration flow itself, one file: download it.
salesforce_writer_flow.py
"""Integration flow driven by the Salesforce Notifier output block.
The Salesforce Notifier output block does not write to Salesforce itself. For every payload of a
completed submission it runs the flow you select in its "Integration Flow" picker, passing down the
Salesforce credentials, the submission id and a reference to the rendered payload. Everything that
happens in Salesforce is therefore ordinary Python you own and can change: this sample looks up a
Salesforce record by the fields it is about to write and creates it only when nothing matches.
To be selectable in that picker a flow needs two things, both below:
* the ``salesforce_integration`` role, and
* the inputs the output block passes in - ``submission_id``, ``payload_uuid``, ``username``,
``consumer_key``, ``private_key`` and ``sandbox_environment``.
The credentials are entered once, on the output block, and forwarded from there, so they are never
stored in this file.
Everything else is configured here, through the manifest defaults: which Salesforce object to
write, how document fields and submission metadata map onto its fields, and whether to run in dry
run mode. A supporting flow has no settings panel of its own, so a manifest default is the only
place these values can come from.
Prerequisites:
* ``simple-salesforce`` and its dependencies uploaded under **Flows -> Python Packages**, for the
same Python version this flow is built for. It is not a single wheel, and packages are tracked
per Python version.
* A Salesforce connected app set up for JWT bearer authentication.
Nothing here is tied to an IDP version, because the flow never touches IDP: it takes a payload
reference and writes to Salesforce. Import it as it is, then add the output block to whichever
document processing flow you already run::
HS_PYTHON_CODE_BLOCK_PYTHON_VERSION=3.13 hs-flows bundle salesforce_writer_flow.py \
-o salesforce_writer_flow.zip
See `Salesforce Integration Flow <../pages/salesforce_integration_flow.html>`_ for the full walkthrough.
"""
from uuid import UUID
from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsBlockInstance, HsTask
from flows_sdk.utils import workflow_input
IDENTIFIER = 'SALESFORCE_INTEGRATION_FLOW_EXAMPLE'
FLOW_UUID = UUID('9c1d4b7e-2f83-4a56-b0d1-7e5c8a3f42b6')
# Must match the output block's role filter, which is what narrows its "Integration Flow" picker.
SALESFORCE_INTEGRATION_ROLE = 'salesforce_integration'
class Inputs:
# Passed in by the Salesforce Notifier output block (hidden, never configured here).
SubmissionId = 'submission_id'
PayloadUuid = 'payload_uuid'
Username = 'username'
ConsumerKey = 'consumer_key'
PrivateKey = 'private_key'
SandboxEnvironment = 'sandbox_environment'
# Configured on this flow, through the manifest defaults below.
SfObject = 'sf_object'
MetadataFieldsMapping = 'metadata_fields_mapping'
DocumentFieldsMapping = 'document_fields_mapping'
DryRun = 'dry_run'
def _write_payload_to_salesforce(
submission_id: int,
payload_uuid: str,
username: str,
consumer_key: str,
private_key: str,
sandbox_environment: bool,
sf_object: str,
metadata_fields_mapping: dict,
document_fields_mapping: dict,
dry_run: bool,
_hs_block_instance: HsBlockInstance,
_hs_task: HsTask,
) -> dict:
import json
# `simple-salesforce` has to be installed for the platform's code-block interpreter, see the
# module docstring. The import lives inside the function because it runs on the block, not
# when the flow is built.
from simple_salesforce import ( # type: ignore[import-not-found]
Salesforce, # pants: no-infer-dep
)
# A Python code block ships only THIS function's source, so every helper has to be defined
# inside it. A module-level helper would raise NameError at runtime.
def log_info(msg: str, *args: object) -> None:
# `log()` applies {}-style formatting to the message, so values that may themselves
# contain braces (a dict, a SOQL string) are passed as ARGS rather than interpolated
# into `msg`: args are substituted, not re-parsed.
_hs_block_instance.log(msg, *args, level=HsBlockInstance.LogLevel.INFO)
def soql_literal(value: object) -> str:
# SOQL string literals are single-quoted; escape backslashes and single quotes.
escaped = str(value).replace('\\', '\\\\').replace("'", "\\'")
return f"'{escaped}'"
# The output block hands over a blob reference, one per rendered payload. The payload is the
# same JSON the platform renders for its output blocks: a top-level `metadata` (submission
# metadata) plus `output` (the serialized submission). With the "Full individual object" export
# type each payload covers one document, with `output.documents` narrowed to it.
blob = _hs_block_instance.fetch_blob(payload_uuid)
payload = json.loads(blob.content.decode('utf-8'))
metadata = payload.get('metadata') or {}
documents = (payload.get('output') or {}).get('documents') or []
def get_extracted_field(field_name: str) -> object:
# A field in a v5 payload looks like:
# {"name": "Name", "output_name": "name", "id": 1, "state": "complete",
# "transcription": {"raw": "Jane Doe", "normalized": "Jane Doe",
# "source": "machine_transcription", ...}, ...}
# so a field is matched on either the layout field name or its output name, and
# `transcription` is an object rather than a string. A plain string transcription is
# accepted too, since that is what the flat export types send.
for document in documents:
for field in document.get('document_fields') or []:
names = (field.get('name'), field.get('output_name'), field.get('field_name'))
if field_name not in names:
continue
transcription = field.get('transcription')
if isinstance(transcription, dict):
# Prefer the normalized value, which is the platform's canonical one:
# "12-0719651" normalizes to "120719651" and text is upper-cased. Use
# `raw` instead where the verbatim string matters.
return transcription.get('normalized') or transcription.get('raw')
return field.get('transcription_normalized') or transcription
return None
# Build the Salesforce record from the two mappings, source name -> Salesforce field API name.
record: dict = {}
for metadata_key, sf_field in (metadata_fields_mapping or {}).items():
if metadata_key not in metadata:
raise ValueError(
f'Submission metadata has no key {metadata_key!r}. '
f'Available keys: {sorted(metadata)}'
)
record[sf_field] = metadata[metadata_key]
for document_field_name, sf_field in (document_fields_mapping or {}).items():
value = get_extracted_field(document_field_name)
if value is None:
raise ValueError(
f'No extracted value for document field {document_field_name!r} in payload '
f'{payload_uuid} of submission {submission_id}.'
)
record[sf_field] = value
if not sf_object:
raise ValueError('sf_object is required (e.g. "Account").')
if not record:
raise ValueError(
'Nothing to write: both metadata_fields_mapping and document_fields_mapping are empty.'
)
# JWT bearer authentication with the connected app credentials the output block passes in:
# username is the JWT subject, consumer_key the issuer, private_key the RS256 signing key.
# A PEM pasted into a form field can arrive with literal "\n", so normalize it first.
private_key = private_key.replace('\\n', '\n')
domain = 'test' if sandbox_environment else 'login'
salesforce = Salesforce(
username=username,
consumer_key=consumer_key,
privatekey=private_key,
domain=domain,
)
# Look the record up by the very fields that are about to be written, AND'ed together, and
# create it only when nothing matches. That keeps a re-run from duplicating records.
where = ' AND '.join(f'{sf_field} = {soql_literal(v)}' for sf_field, v in record.items())
query = f'SELECT Id FROM {sf_object} WHERE {where}'
# Log the query before running it, so it is visible even if the query itself errors.
log_info('Salesforce lookup query: {}', query)
result = salesforce.query_all(query)
count = result.get('totalSize', len(result.get('records', [])))
log_info('Salesforce lookup on {} with {} -> {} record(s).', sf_object, record, count)
created_id = None
if dry_run:
log_info('Dry run: not creating a {} record. Would have written {}.', sf_object, record)
elif count == 0:
created = getattr(salesforce, sf_object).create(record)
created_id = created.get('id')
log_info('No existing {}; created record {} with {}.', sf_object, created_id, record)
else:
log_info('{} existing {} record(s) matched; skipping create.', count, sf_object)
return {
'submission_id': submission_id,
'payload_uuid': payload_uuid,
'result_count': count,
'lookup_fields': record,
'query': query,
'dry_run': dry_run,
'created': created_id is not None,
'salesforce_id': created_id,
}
def workflow() -> Flow:
write_to_salesforce = PythonBlock(
reference_name='write_payload_to_salesforce',
code=_write_payload_to_salesforce,
# No `python_version` here: the block compiles to the Python version of the venv this
# is built in, or to `HS_PYTHON_CODE_BLOCK_PYTHON_VERSION` if that is set. Pick the one
# your platform runs, as described in the walkthrough.
code_input={
'submission_id': workflow_input(Inputs.SubmissionId),
'payload_uuid': workflow_input(Inputs.PayloadUuid),
'username': workflow_input(Inputs.Username),
'consumer_key': workflow_input(Inputs.ConsumerKey),
'private_key': workflow_input(Inputs.PrivateKey),
'sandbox_environment': workflow_input(Inputs.SandboxEnvironment),
'sf_object': workflow_input(Inputs.SfObject),
'metadata_fields_mapping': workflow_input(Inputs.MetadataFieldsMapping),
'document_fields_mapping': workflow_input(Inputs.DocumentFieldsMapping),
'dry_run': workflow_input(Inputs.DryRun),
},
title='Write payload to Salesforce',
description='Fetches the rendered submission payload, maps its metadata and extracted '
'document fields to Salesforce field API names, and writes them to a Salesforce record '
'using simple-salesforce.',
)
return Flow(
uuid=FLOW_UUID,
owner_email='flows-sdk@hyperscience.com',
title='Salesforce Integration Flow - flows-sdk example',
description='Writes one rendered submission payload to Salesforce. Driven by the '
'Salesforce Notifier output block, which passes in the credentials, the submission id '
'and the payload reference.',
manifest=Manifest(
identifier=IDENTIFIER,
# 'supporting' marks this as a subflow, so the validator does not require values for
# the inputs the caller supplies. The salesforce role is what makes the output
# block's "Integration Flow" picker offer this flow.
roles=['supporting', SALESFORCE_INTEGRATION_ROLE],
input=[
# Passed in by the output block, hidden because they are not configured here.
Parameter(
name=Inputs.SubmissionId,
type='integer',
title='Submission ID',
ui={'hidden': True},
),
Parameter(
name=Inputs.PayloadUuid,
type='string',
title='Payload UUID',
ui={'hidden': True},
),
Parameter(
name=Inputs.Username,
type='string',
title='Associated Username',
ui={'hidden': True},
),
Parameter(
name=Inputs.ConsumerKey,
type='Password',
title='Consumer Key',
secret=True,
ui={'hidden': True},
),
Parameter(
name=Inputs.PrivateKey,
type='Password',
title='Private Key',
secret=True,
ui={'hidden': True},
),
Parameter(
name=Inputs.SandboxEnvironment,
type='boolean',
title='Sandbox Environment',
value=False,
ui={'hidden': True},
),
# Configured here, through these defaults. The output block passes a fixed input
# set, and a supporting flow has no settings panel, so a default is the only way
# to set them.
Parameter(
name=Inputs.SfObject,
type='string',
title='Salesforce Object',
description='API name of the Salesforce object to look up and create, '
'e.g. Account.',
value='Account',
),
Parameter(
name=Inputs.MetadataFieldsMapping,
type='json',
title='Metadata Field Mappings',
description='Map submission metadata keys to Salesforce field API names, e.g. '
'{"customer_id": "AccountNumber"}. Submission metadata is empty for documents '
'uploaded through the UI.',
value={},
optional=True,
),
Parameter(
name=Inputs.DocumentFieldsMapping,
type='json',
title='Document Field Mappings',
description='Map extracted document field names to Salesforce field API '
'names, e.g. {"Name": "Name"}. A mapped field with no extracted value fails '
'the flow rather than writing a partial record.',
# "Name" comes from the W-9 layout shipped with the Document Processing
# Starter and maps onto a standard Account field, so nothing has to be created
# in Salesforce first. It is printed on the sample document, so machine
# transcription reads it without a human keyer - unlike its "Address", which
# is handwritten and comes back illegible.
value={'Name': 'Name'},
),
Parameter(
name=Inputs.DryRun,
type='boolean',
title='Dry Run',
description='Run the Salesforce lookup and log what would be written, but do '
'not create the record. Turn this off to write for real.',
value=True,
),
],
),
blocks=[write_to_salesforce],
output={'result': write_to_salesforce.output()},
)
def entry_point_workflow() -> Flow:
return workflow()
if __name__ == '__main__':
export_flow(entry_point_workflow())
It stands in for vendor onboarding: a payee returns a Form W-9, and the flow creates that vendor’s
Salesforce Account object from it, mapping the name on line 1 of the form to Account.Name. It
looks the record up by the fields it is about to write and creates it only when nothing matches, and
it ships with dry run on, so you can follow it end to end without writing anything to your org.
The mapping is deliberately one field, because the point of the sample is the flow rather than the
mapping. Account.Name is standard, so nothing has to be created in Salesforce before the
walkthrough works. Matching on a name alone is not what you want in production, though: add the
field your org actually keys on, a tax id for instance, and the lookup narrows to it, as in
{"Name": "Name", "Employer Identification Number": "Tax_ID__c"}. Every mapped field becomes part
of the lookup, so a field that does not exist in your org fails the query rather than the write.
To onboard individuals as Contact records instead, change sf_object and map onto LastName,
which is required, rather than Name, which Salesforce derives and will not accept a write on.
The output block is added to a document processing flow you already have, in Flow Studio, which is where outputs are best defined. So there is no IDP flow in the sample, and nothing in it is tied to an IDP version.
Prerequisites
The Python version your platform runs. Everything below depends on it: the code block the flow is built for, and the wheels you upload. Look your release up in the table in Managing Third-Party Python Packages, and use that version wherever the steps say
3.13. Choosing the Python version explains how the flow picks one up.The
flows_sdkwheel installed, as in Getting Started.A document processing flow that extracts documents. The Document Processing Starter for your platform version, from Examples, set up as in Getting Started, is enough - it ships the Form W-9 layout release and the
form-w9-sample.pngused in the expected results below.A Salesforce connected app configured for JWT bearer authentication: its consumer key, the matching private key, and the username the app is authorized for.
Permission to edit flows and secrets, and a license that includes the enterprise system connectors. Without it the block is greyed out in the picker.
Setting it up
The commands below use Python 3.13 for the sake of the example. Substitute the version your release runs, from the table in Managing Third-Party Python Packages.
Upload
simple-salesforceand its dependencies under Flows -> Python Packages. It is not a single wheel:cryptography,pyjwt,zeep,lxml,requestsand others come with it, and every one of them has to be uploaded, built for Linux and for the same Python version the flow is built for.pip downloadfetches the whole set:pip download --python-version 3.13 --platform=manylinux2014_x86_64 \ --only-binary=:all: simple-salesforce
Package the integration flow, building it for the same Python version:
HS_PYTHON_CODE_BLOCK_PYTHON_VERSION=3.13 hs-flows bundle salesforce_writer_flow.py \ -o salesforce_writer_flow.zip
Import it at
https://{hyperscience-url}/flowswith Import Flow. Do this before the next step: the output block’s Integration Flow picker only lists flows that are already imported.Open your document processing flow in Flow Studio and click the Outputs block on the canvas. The panel on the right lists the flow’s connections and has an Add button.
Press Add, pick Salesforce Notifier Output Block under Connection Type, and press Add Connection.
In the connection’s settings, select Salesforce Integration Flow - flows-sdk example under Integration Flow, then fill in the associated username, consumer key and private key, ticking Sandbox Environment for a sandbox org, and press Test Connection. This is the only place the credentials are entered: the output block forwards them to the integration flow, so they are never stored in a file.
Set API Version to
v5and pick Full individual object under Export Type to get one integration flow run per extracted document. Neither has a default, and the flow cannot go live while one of them is empty.Turn the connection’s toggle on - a connection is added disabled - then Save the flow and set it to Live.
Submit
form-w9-sample.pngfrom the starter package athttps://{hyperscience-url}/submissions, or through the submission creation API.
Expected results
Open the completed submission and choose Actions -> View Flow Run. The output block has one
child flow run, one per extracted document. That run’s write_payload_to_salesforce block logs
Salesforce lookup query: SELECT Id FROM Account WHERE Name = 'CHRISTOPHER THOMAS KNIGHT'
Salesforce lookup on Account with {'Name': 'CHRISTOPHER THOMAS KNIGHT'} -> 0 record(s).
Dry run: not creating a Account record. Would have written {'Name': 'CHRISTOPHER THOMAS KNIGHT'}.
and returns
{
"result": {
"submission_id": 10,
"payload_uuid": "bf113b8c-5e69-4e08-92d1-0f4b0a304f7f",
"result_count": 0,
"lookup_fields": {"Name": "CHRISTOPHER THOMAS KNIGHT"},
"query": "SELECT Id FROM Account WHERE Name = 'CHRISTOPHER THOMAS KNIGHT'",
"dry_run": true,
"created": false,
"salesforce_id": null
}
}
The submission id and payload uuid are yours; the value is the sample W-9’s normalized
transcription, which is why the name is upper-cased. The sample maps only a field that is printed on
the document: a mapped field with no extracted value fails the flow rather than writing a partial
record, and this W-9’s handwritten Address comes back illegible.
To write for real, turn Dry Run off. The connection’s settings link to the selected flow: the
name shown under Integration Flow opens it in Flow Studio, where its own settings, Dry Run
among them, can be changed and saved without leaving the browser. The other route is to set
dry_run to False in salesforce_writer_flow.py and re-import, which is what you want when the
flow is kept in source, since the next import would otherwise put the old value back.
Either way the block looks the record up by the fields it is about to write and creates it only when nothing matches, so re-processing a submission does not duplicate records.
Choosing the Python version
The sample does not pin a Python version. A PythonBlock compiles to a code block of a specific
Python version, and which one it gets is decided at build time, in this order:
an explicit
python_version=argument on the block, which the sample deliberately omits;the
HS_PYTHON_CODE_BLOCK_PYTHON_VERSIONenvironment variable, if set;otherwise, the Python version of the virtual environment you build in.
The steps above set the environment variable, which works whichever venv you happen to be in. Building in a venv whose Python already matches your platform does the same thing without it.
Which versions your release runs, and which one it defaults to, is listed in
Managing Third-Party Python Packages. That same choice decides which
wheels you download for simple-salesforce: packages are tracked separately per Python version, so
a package installed for one is not available to a block running another.
Importing a flow that names a code block your platform does not have fails, so it is worth checking the built flow before importing it:
unzip -p salesforce_writer_flow.zip '*.json' | grep -o 'PYTHON_[0-9_]*CODE' | sort -u
Writing your own
A flow is offered in the output block’s Integration Flow picker when it declares the
salesforce_integration role and accepts the inputs the block passes in:
Input |
Description |
|---|---|
|
id of the submission being exported |
|
reference to one rendered payload, fetched with |
|
the credentials entered on the output block |
How many payloads there are, and therefore how many runs of your flow, follows the output block’s Export Type: one per extracted document for “Full individual object”, or a single payload for the whole submission. The payload itself is the same JSON the platform renders for its output blocks.
Anything else your flow needs comes from its own manifest defaults, as sf_object,
document_fields_mapping, metadata_fields_mapping and dry_run do in the sample: the output
block passes a fixed set of inputs, and a supporting flow has no settings panel of its own.
metadata_fields_mapping is the counterpart of document_fields_mapping for submission metadata,
which is what an API submission can attach to a submission and is empty for documents uploaded
through the UI.
Adding the output block in code instead
A connection added in Flow Studio lives in the platform’s copy of the flow, so re-importing that flow
from the SDK overwrites it. If you keep your document processing flow in source, add the block to its
IDPOutputsBlock there:
from flows_sdk.blocks import IOBlock
salesforce_notifier = IOBlock(
identifier='COB_SALESFORCE_NOTIFIER_2',
reference_name='salesforce_notifier',
title='Salesforce Notifier Output Block',
enabled=True,
input={
'submission': idp_flow.output('submission'),
# The `IDENTIFIER` of the integration flow: what the "Integration Flow" picker would set.
'integration_flow_identifier': 'SALESFORCE_INTEGRATION_FLOW_EXAMPLE',
'api_version': 'v5',
# One payload, and therefore one integration flow run, per extracted document. Use
# 'SUBMISSION' for a single payload covering the whole submission.
'export_type': 'DOCUMENTS',
},
)
outputs = idp_blocks.IDPOutputsBlock(
inputs={'submission': idp_flow.output('submission')},
blocks=[salesforce_notifier],
)
The credentials stay out of the file either way: enter them on the block in Flow Studio. Bundling
your flow together with the integration flow imports both at once, and
integration_flow_identifier resolves during the import whichever order they land in:
HS_PYTHON_CODE_BLOCK_PYTHON_VERSION=3.13 hs-flows bundle your_flow.py salesforce_writer_flow.py \
-o salesforce_flows.zip
Before v44
The updated Salesforce Notifier output block does not exist on earlier releases, but the same job can be done without it: a document processing flow of your own calls a writer subflow directly, and that subflow holds the Salesforce code. The result is the same lookup-then-create against Salesforce, wired by hand instead of through an output block.
That sample is two files: download it.
document_processing_flow.py
"""Document processing flow that writes to Salesforce, for platforms before v44.
Runs IDP Core and then calls the Salesforce Writer subflow (`salesforce_writer_flow.py`), handing
it the processed submission, the trigger (which carries the submission metadata) and the Salesforce
credentials.
The credentials are configured HERE, once. You iterate on the Salesforce logic in the subflow and
re-import only that, so this flow - and the credentials on it - stay untouched.
Written for Hyperscience v43. For an older release, change the `idp_v43` imports below to the
matching `idp_vXX` package so the flow references the IDP Core subflow that release installs, and
build in a venv whose Python matches the code block that release ships.
Import the writer subflow FIRST (this flow references it by identifier), then this one::
HS_PYTHON_CODE_BLOCK_PYTHON_VERSION=3.13 hs-flows bundle document_processing_flow.py \
-o document_processing_flow.zip
See `Salesforce Integration Flow <../pages/salesforce_integration_flow.html>`_, section "Before v44", for the
full walkthrough. On v44 and later, add the updated Salesforce Notifier output block to a
processing flow you already run instead, as `samples/salesforce_integration_flow` shows.
"""
from uuid import UUID
from flows_sdk.blocks import Block
from flows_sdk.flows import Flow, Parameter
from flows_sdk.implementations.idp_v43.idp_blocks import IDPCoreBlock
from flows_sdk.implementations.idp_v43.idp_values import (
IDPTriggers,
IDPWrapperManifest,
Settings,
get_idp_wf_config,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import workflow_input
# `WRITER_IDENTIFIER` of `salesforce_writer_flow.py`, declared here too so neither file has to
# import the other: a downloaded sample then runs from any directory. Keep the two in sync.
WRITER_IDENTIFIER = 'SALESFORCE_WRITER_EXAMPLE'
IDENTIFIER = 'SALESFORCE_DOCUMENT_PROCESSING_EXAMPLE'
FLOW_UUID = str(UUID('654a59c7-d04f-4e16-9abd-3a02962ec22a'))
class Inputs:
# Salesforce connected app credentials (JWT bearer flow), configured on THIS flow, once.
# Iterating on the subflow never re-imports them.
Username = 'username'
ConsumerKey = 'consumer_key'
PrivateKey = 'private_key'
SandboxEnvironment = 'sandbox_environment'
def workflow() -> Flow:
# Standard IDP Core document processing (classification / identification / transcription).
idp_core = IDPCoreBlock(get_idp_wf_config())
# Call the Salesforce Writer subflow. A block whose identifier is another flow's identifier is
# executed as a sub-workflow: the platform runs SALESFORCE_WRITER_EXAMPLE and waits for it.
# We pass the processed submission, the trigger (carries the metadata), and the credentials.
write_to_salesforce = Block(
identifier=WRITER_IDENTIFIER,
reference_name='salesforce_writer',
input={
'submission': idp_core.output('submission'),
'trigger': workflow_input('$trigger'),
'username': workflow_input(Inputs.Username),
'consumer_key': workflow_input(Inputs.ConsumerKey),
'private_key': workflow_input(Inputs.PrivateKey),
'sandbox_environment': workflow_input(Inputs.SandboxEnvironment),
},
title='Write to Salesforce',
description='Calls the Salesforce Writer subflow with the processed submission and '
'credentials.',
)
# Start from the standard IDP wrapper manifest (exposes Layout Release and the other IDP flow
# settings), then add the Salesforce credential inputs.
manifest = IDPWrapperManifest(IDENTIFIER)
manifest.input = list(manifest.input) + [
Parameter(
name=Inputs.Username,
type='string',
title='Associated Username',
description='Username associated with the Salesforce Connected App',
value='',
),
Parameter(
name=Inputs.ConsumerKey,
type='Password',
title='Consumer Key',
description='Consumer key used for authentication with Salesforce. Paired with the '
'Salesforce instance from which it was issued.',
secret=True,
value='',
),
Parameter(
name=Inputs.PrivateKey,
type='Password',
title='Private Key',
description='Private key used for authentication with Salesforce',
secret=True,
value='',
),
Parameter(
name=Inputs.SandboxEnvironment,
type='boolean',
title='Sandbox Environment',
description='Enable if the Salesforce environment is a sandbox',
value=False,
),
]
return Flow(
uuid=UUID(FLOW_UUID),
owner_email='flows-sdk@hyperscience.com',
title='Salesforce Document Processing - flows-sdk example',
description='Document processing flow that runs IDP Core and then calls the Salesforce '
'Writer subflow to write submission data to Salesforce.',
manifest=manifest,
input={Settings.LayoutReleaseUuid: None},
triggers=IDPTriggers(),
blocks=[idp_core, write_to_salesforce],
output={'submission': idp_core.output('submission')},
)
def entry_point_workflow() -> Flow:
return workflow()
if __name__ == '__main__':
export_flow(entry_point_workflow())
salesforce_writer_flow.py
"""Salesforce Writer subflow: the half of this example that does the Salesforce work.
The document processing flow (`document_processing_flow.py`) calls this flow as a subflow, passing
the processed submission, the trigger (which carries the submission metadata) and the Salesforce
credentials. Keeping the two apart is what lets you iterate: the credentials are configured on the
document processing flow, so re-importing this subflow never clears them.
This subflow owns the settings you tune most often - the target Salesforce object and the two field
mappings - and the Python that writes the record.
Import this subflow FIRST; the document processing flow references it by identifier.
HS_PYTHON_CODE_BLOCK_PYTHON_VERSION=3.13 hs-flows bundle salesforce_writer_flow.py \
-o salesforce_writer_flow.zip
See `Salesforce Integration Flow <../pages/salesforce_integration_flow.html>`_, section "Before v44", for the
full walkthrough. On v44 and later, prefer the updated Salesforce Notifier output block and
integration flow in `samples/salesforce_integration_flow`.
"""
from uuid import UUID
from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsBlockInstance, HsTask
from flows_sdk.utils import workflow_input
# The document processing flow calls this subflow by identifier. It declares the same string
# itself, so the two files stay independent: keep them in sync.
WRITER_IDENTIFIER = 'SALESFORCE_WRITER_EXAMPLE'
WRITER_FLOW_UUID = str(UUID('7c1d3e58-9b2a-4c6e-8f0d-1a2b3c4d5e6f'))
class Inputs:
# Passed in by the parent flow (not configured here).
Submission = 'submission'
Trigger = 'trigger'
Username = 'username'
ConsumerKey = 'consumer_key'
PrivateKey = 'private_key'
SandboxEnvironment = 'sandbox_environment'
# Configured on THIS subflow (non-secret; safe to reset on re-import, and have defaults).
SfObject = 'sf_object'
# Mappings of source name -> Salesforce field API name, as JSON objects:
# {"SOURCE_NAME": "SF_FIELD_API_NAME"}.
MetadataFieldsMapping = 'metadata_fields_mapping'
DocumentFieldsMapping = 'document_fields_mapping'
DryRun = 'dry_run'
def _write_metadata_to_salesforce(
trigger: dict,
submission: dict,
username: str,
consumer_key: str,
private_key: str,
sandbox_environment: bool,
sf_object: str,
metadata_fields_mapping: dict,
document_fields_mapping: dict,
dry_run: bool,
_hs_block_instance: HsBlockInstance,
_hs_task: HsTask,
) -> dict:
# `simple-salesforce` and its dependencies must be installed on the platform for this
# interpreter, under Flows -> Python Packages. The import lives inside the function because it
# runs on the block, not at flow-compile time.
from simple_salesforce import Salesforce # type: ignore[import-not-found] # pants: no-infer-dep
# Build the Salesforce record from the two mappings, source name -> Salesforce field.
record: dict = {}
# 1. Extract fields for the Submission metadata.
# Metadata rides on the trigger at $trigger.payload.api_params.metadata;
# the parent passes that trigger in as `trigger`.
# metadata_fields_mapping: {"metadata_key": "SF_FIELD_API_NAME"}
metadata = trigger.get('payload', {}).get('api_params', {}).get('metadata', {}) or {}
for metadata_key, sf_field in (metadata_fields_mapping or {}).items():
if metadata_key not in metadata:
raise ValueError(
f'Submission metadata has no key {metadata_key!r}. '
f'Available keys: {sorted(metadata)}'
)
record[sf_field] = metadata[metadata_key]
# 2. Extract document fields from the processed submission. The `submission` object is the
# output of IDP Core and carries the documents/pages/fields. Each extracted field looks like:
# { "field_name": "document_field", "output_name": "...",
# "transcription": "Acme Corporation", "transcription_normalized": "ACME CORPORATION" }
# NB: the Python Code block only ships THIS function's source, so any helper must be defined
# inside it (a module-level helper would raise NameError at runtime).
def get_extracted_field(field_name: str) -> str | None:
for document in submission.get('documents', []):
for field in document.get('document_fields', []) or []:
if field_name in (field.get('field_name'), field.get('output_name')):
# Prefer the normalized value; fall back to the raw transcription.
return field.get('transcription_normalized') or field.get('transcription')
return None
for document_field_name, sf_field in (document_fields_mapping or {}).items():
value = get_extracted_field(document_field_name)
if value is None:
raise ValueError(
f'No extracted value for document field {document_field_name!r} in the submission.'
)
record[sf_field] = value
if not sf_object:
raise ValueError('sf_object is required (e.g. "Account").')
if not record:
raise ValueError(
'Nothing to write: both metadata_fields_mapping and document_fields_mapping are empty.'
)
# JWT bearer auth with the connected app credentials the document processing flow passes in:
# username -> JWT subject, consumer_key -> JWT issuer, private_key -> RS256 signing key.
# A PEM pasted into a form field can arrive with literal "\n", so normalize it first.
private_key = private_key.replace('\\n', '\n')
domain = 'test' if sandbox_environment else 'login'
salesforce = Salesforce(
username=username,
consumer_key=consumer_key,
privatekey=private_key,
domain=domain,
)
# Look up existing records on the chosen object using the mapped fields as WHERE criteria
# (AND'd together). If none match, create the record; otherwise skip creation to avoid a
# duplicate. SOQL string literals are single-quoted, so escape backslashes and single quotes.
def soql_literal(value: object) -> str:
escaped = str(value).replace('\\', '\\\\').replace("'", "\\'")
return f"'{escaped}'"
# `log()` formats the message with `{}` style, so values that contain braces (a dict, a SOQL
# string) are passed as ARGS rather than interpolated into `msg`: args are substituted, not
# re-parsed. `level` is keyword-only; passed positionally it would be taken for an arg.
def log_info(msg: str, *args: object) -> None:
_hs_block_instance.log(msg, *args, level=HsBlockInstance.LogLevel.INFO)
where = ' AND '.join(f'{sf_field} = {soql_literal(v)}' for sf_field, v in record.items())
query = f'SELECT Id FROM {sf_object} WHERE {where}'
# Log the query BEFORE running it, so it is visible even if the query itself errors.
log_info('Salesforce lookup query: {}', query)
result = salesforce.query_all(query)
count = result.get('totalSize', len(result.get('records', [])))
log_info('Salesforce lookup on {} with {} -> {} record(s).', sf_object, record, count)
created_id = None
if dry_run:
# The lookup above still ran, so a dry run proves the credentials and the mapping without
# touching the org. Turn Dry Run off to write for real.
log_info(
'Dry run: not creating a {} record. Would have written {}.', sf_object, record
)
elif count == 0:
# No existing match - create the record.
created = getattr(salesforce, sf_object).create(record)
created_id = created.get('id')
log_info(
'No existing {}; created record {} with {}.', sf_object, created_id, record
)
else:
log_info('{} existing {} record(s) matched; skipping create.', count, sf_object)
return {
'result_count': count,
'lookup_fields': record,
'query': query,
'dry_run': dry_run,
'created': created_id is not None,
'salesforce_id': created_id,
}
def workflow() -> Flow:
write_to_salesforce = PythonBlock(
reference_name='write_metadata_to_salesforce',
code=_write_metadata_to_salesforce,
# No `python_version` here: the block compiles to the Python version of the venv this
# is built in, or to `HS_PYTHON_CODE_BLOCK_PYTHON_VERSION` if that is set. Pick the one
# your platform runs, as described in the walkthrough.
code_input={
'trigger': workflow_input(Inputs.Trigger),
'submission': workflow_input(Inputs.Submission),
'username': workflow_input(Inputs.Username),
'consumer_key': workflow_input(Inputs.ConsumerKey),
'private_key': workflow_input(Inputs.PrivateKey),
'sandbox_environment': workflow_input(Inputs.SandboxEnvironment),
'sf_object': workflow_input(Inputs.SfObject),
'metadata_fields_mapping': workflow_input(Inputs.MetadataFieldsMapping),
'document_fields_mapping': workflow_input(Inputs.DocumentFieldsMapping),
'dry_run': workflow_input(Inputs.DryRun),
},
title='Write metadata to Salesforce',
description='Maps submission metadata and extracted document fields to Salesforce field '
'API names and writes them to a Salesforce record using simple-salesforce.',
)
return Flow(
uuid=UUID(WRITER_FLOW_UUID),
owner_email='flows-sdk@hyperscience.com',
title='Salesforce Writer (Example Subflow)',
description='Writes submission data to Salesforce. Called as a subflow by the Salesforce '
'document processing example, which passes the Salesforce credentials in.',
manifest=Manifest(
# 'supporting' marks this as a subflow: the validator does not require values for its
# mandatory inputs (submission/trigger/credentials) because the parent passes them in.
identifier=WRITER_IDENTIFIER,
roles=['supporting'],
input=[
# Passed in by the parent (hidden - not user-configured on this subflow).
Parameter(
name=Inputs.Submission, type='object', title='Submission', ui={'hidden': True}
),
Parameter(name=Inputs.Trigger, type='object', title='Trigger', ui={'hidden': True}),
Parameter(
name=Inputs.Username,
type='string',
title='Associated Username',
ui={'hidden': True},
),
Parameter(
name=Inputs.ConsumerKey,
type='Password',
title='Consumer Key',
secret=True,
ui={'hidden': True},
),
Parameter(
name=Inputs.PrivateKey,
type='Password',
title='Private Key',
secret=True,
ui={'hidden': True},
),
Parameter(
name=Inputs.SandboxEnvironment,
type='boolean',
title='Sandbox Environment',
value=False,
ui={'hidden': True},
),
# Configured on this subflow (non-secret, with defaults).
Parameter(
name=Inputs.SfObject,
type='string',
title='Salesforce Object',
description='API name of the Salesforce object to create, e.g. Account.',
value='',
),
Parameter(
name=Inputs.MetadataFieldsMapping,
type='json',
title='Metadata Field Mappings',
description='Map submission metadata keys to Salesforce field API names, e.g. '
'{"sf_account_name": "Name"}.',
value={},
),
Parameter(
name=Inputs.DocumentFieldsMapping,
type='json',
title='Document Field Mappings',
description='Map extracted document field names to Salesforce field API names, '
'e.g. {"OWNER_NAME": "AccountNumber"}. Leave empty to write metadata only.',
value={},
optional=True,
),
Parameter(
name=Inputs.DryRun,
type='boolean',
title='Dry Run',
description='Run the Salesforce lookup and log what would be written, but do '
'not create the record. Turn this off to write for real.',
value=True,
),
],
),
blocks=[write_to_salesforce],
output={'result': write_to_salesforce.output()},
)
def entry_point_workflow() -> Flow:
return workflow()
if __name__ == '__main__':
export_flow(entry_point_workflow())
File |
What it is |
|---|---|
|
runs IDP Core, then calls the writer by identifier. Holds the Salesforce credentials |
|
the writer subflow: the Python that talks to Salesforce, plus the target object and the field mappings |
Splitting them is what makes the example usable: re-importing a flow overwrites the input values configured on it, credentials included. Keeping the credentials on the document processing flow and the volatile code in the subflow means you can rebuild and re-import the writer as often as you like without re-entering them. The parent resolves the subflow by identifier at runtime, so it does not need re-importing when the writer changes.
Like the v44 sample, this one can write submission metadata as well as extracted fields. It
reaches it differently, because it renders no payload: metadata supplied to
POST /api/v5/submissions arrives on the flow run trigger at
$trigger.payload.api_params.metadata, and the document processing flow passes the trigger down.
On v44 the same values arrive in the rendered payload’s top-level metadata, which is where the
v44 sample reads them. Either way, metadata_fields_mapping maps their keys onto Salesforce fields,
document_fields_mapping does the same for extracted document fields, and either may be empty.
Setting up the pre-v44 sample
As above, the commands use Python 3.13 for the sake of the example; substitute the version your release runs.
Upload
simple-salesforceand its dependencies under Flows -> Python Packages, exactly as in step 1 of Setting it up - same command, same caveat that it is not a single wheel and every dependency has to be uploaded for the Python version your release runs.Package the two flows, building them for that same version:
HS_PYTHON_CODE_BLOCK_PYTHON_VERSION=3.13 hs-flows bundle salesforce_writer_flow.py \ -o salesforce_writer_flow.zip HS_PYTHON_CODE_BLOCK_PYTHON_VERSION=3.13 hs-flows bundle document_processing_flow.py \ -o document_processing_flow.zip
Import the writer first at
https://{hyperscience-url}/flows, then the document processing flow, which references it by identifier.On the document processing flow, fill in the Salesforce username, consumer key and private key, assign a Layout Release, then Save and set it Live.
Open the Write to Salesforce block to set the Salesforce object and the two field mappings. These live on the writer, so changing them later means re-importing only that flow.
Submit a document. If you are mapping metadata, submit through the submission creation API and pass the
metadataobject there: it cannot be set from the UI, so a submission uploaded in the browser carries none. Mapping only document fields works either way. The writer’s block logs the SOQL query it runs and what it would write, and returnsresult_count,query,dry_run,created,salesforce_idandlookup_fields.Like the v44 sample, this one ships with Dry Run on: it authenticates and runs the lookup, so a first pass proves the credentials and the mappings without touching your org. Turn Dry Run off on the Write to Salesforce block to write for real.
Submitting the Document Processing Starter’s
form-w9-sample.pngwithmetadata={"vendor_reference": "VENDOR-42"},sf_objectAccount,document_fields_mapping{"Name": "Name"}andmetadata_fields_mapping{"vendor_reference": "AccountNumber"}logsSalesforce lookup query: SELECT Id FROM Account WHERE AccountNumber = 'VENDOR-42' AND Name = 'CHRISTOPHER THOMAS KNIGHT' Salesforce lookup on Account with {'AccountNumber': 'VENDOR-42', 'Name': 'CHRISTOPHER THOMAS KNIGHT'} -> 0 record(s). Dry run: not creating a Account record. Would have written {'AccountNumber': 'VENDOR-42', 'Name': 'CHRISTOPHER THOMAS KNIGHT'}.one value off the document and one off the submission metadata.
Adapting it to your release
Two things follow the platform version. The sample is written for v43:
document_processing_flow.pyimportsflows_sdk.implementations.idp_v43. Changeidp_v43to the package matching your release so the flow references theIDP_CORE_Vxxsubflow installed on it.the writer’s code block follows the Python version you build with, so build it for the release you are importing into, as Choosing the Python version describes.