Passa al contenuto principale

Source

Flowlyze supporta due modalità di integrazione delle sorgenti: attiva e passiva. Entrambe consentono di acquisire, normalizzare e processare i dati in modo coerente all’interno della piattaforma.

Panoramica

Le modalità di ingresso sono due:

  • Attiva (pull): Flowlyze preleva i dati dalle sorgenti a intervalli o su richiesta.
  • Passiva (push): le sorgenti inviano eventi/messaggi a endpoint esposti da Flowlyze.

In entrambi i casi, i dati vengono acquisiti, normalizzati e instradati nelle stesse pipeline di trasformazione e validazione.

Modalità Attiva (pull)​

A intervalli programmati o su richiesta (via API o azione manuale in UI), Flowlyze preleva i dati dalla sorgente e li carica nel database interno. Il trattamento successivo segue le regole definite nel flusso (pipeline, trasformazioni, validazioni).

Sorgenti tipiche

  • Connettori API standard (REST/JSON).
  • Database relazionali o NoSQL.
  • Flat file (es. CSV) su storage locale o FTP/SFTP.

Esempi d’uso

  • Ogni notte alle 00:00: importazione di un file CSV da uno spazio FTP.
  • Ogni ora: lettura di una tabella da un database operativo.

Durante la fase di configurazione di una Source in Flowlyze è possibile definire, a livello generale, la ricorrenza dell’esecuzione tramite una cron expression. Questa funzionalità consente di pianificare in modo flessibile e preciso la frequenza con cui la sorgente deve essere attivata (ad esempio ogni minuto, ogni ora, ogni giorno o in orari specifici).

Inoltre, è possibile specificare il tipo di integrazione utilizzata, selezionando tra le diverse modalità supportate da Flowlyze (ad esempio API REST, database, file system, webhook, eccetera), così da adattare il comportamento della Source alle esigenze del flusso di integrazione.

Per facilitare l’utilizzo anche in contesti globali, i riferimenti temporali della Cron Expression sono sempre in UTC (quindi non tengono conto della localizzazione dell’utilizzatore, né della time zone predefinita scelta per il tenant).

Nell’immagine seguente è mostrata la configurazione di base di una sorgente schedulata ogni minuto.

Template piattaforma

Questa configurazione indica che la Source di tipo Http (chiamata verso API) verrà eseguita ogni minuto di ogni ora, ogni giorno.

Cos’è una Cron Expression​

Una cron expression è una stringa che definisce una pianificazione temporale ricorrente. È composta da cinque o sei campi (a seconda del sistema), ognuno dei quali rappresenta un intervallo temporale:

CampoDescrizioneValori possibili
1Minuti0–59
2Ore0–23
3Giorno del mese1–31
4Mese1–12 o nomi (JAN–DEC)
5Giorno della settimana0–6 o nomi (SUN–SAT)

Esempi di cron expression​

Cron expressionSignificato
* * * * *Ogni minuto
0 * * * *Ogni ora, al minuto 0
0 0 * * *Ogni giorno a mezzanotte
0 9 * * 1-5Ogni giorno lavorativo alle 09:00
*/15 * * * *Ogni 15 minuti
0 0 1 * *Il primo giorno di ogni mese a mezzanotte

Gestione dati incrementali​

Flowlyze mette a disposizione un sistema di memoria integrato che consente di implementare letture incrementali dalle sorgenti dati, ottimizzando così le performance dei flussi e riducendo la quantità di dati processati a ogni esecuzione.

Incrementale

Si identifica un campo discriminante (timestamp/ordinamento). Ad ogni run viene memorizzato il valore massimo processato; al successivo run si leggono solo i record successivi. Questo riduce in modo significativo il carico sulle sorgenti e sui flussi.

Il meccanismo si basa sull’identificazione di un campo discriminante nella sorgente (ad esempio una colonna di un database o un campo presente nel payload di una chiamata HTTP) che rappresenta la sequenzialità o l’ordine temporale dei dati.

Durante ogni esecuzione, Flowlyze memorizza il valore massimo (o altro criterio di aggregazione) ottenuto per quel campo all’interno del set di dati processati. Nell’esecuzione successiva, tale valore viene utilizzato come punto di riferimento per richiedere solo i nuovi record o quelli modificati dopo l’ultimo aggiornamento.

Esempio di comportamento​

Supponiamo che la sorgente contenga un campo last_update che rappresenta la data di ultima modifica di ogni record. Alla prima esecuzione, Flowlyze elabora tutti i dati disponibili e registra il valore frutto di un’operazione di aggregazione (generalmente il valore massimo di last_update trovato. All’esecuzione successiva, la piattaforma userà tale valore per interrogare la sorgente, richiedendo solo i dati con last_update successivo a quello memorizzato.

In questo modo, il sistema evita la rilettura di dati già processati, garantendo efficienza, coerenza e scalabilità del flusso.

Parametri di configurazione​

La configurazione del meccanismo incrementale è completamente personalizzabile e consente di definire i seguenti campi:

CampoDescrizione
Nome variabileIdentifica la variabile associata al campo incrementale. Può essere richiamata all’interno delle configurazioni tramite la sintassi `{{nome_variabile}}`.
Message JSON PathEspressione JSONPath che individua, all’interno del messaggio o del payload, il campo discriminante da utilizzare per la lettura incrementale.
AggregatorCriterio di aggregazione applicato al campo per determinare il valore da memorizzare (ad esempio max, min, ecc.). Nella maggior parte dei casi viene utilizzato max.
Current valueValore attuale memorizzato per il campo incrementale. Può essere impostato o modificato manualmente per forzare una ripartenza da un punto specifico.
OffsetIntero opzionale applicato al valore memorizzato al momento del recupero dati. Consente di anticipare o posticipare il punto di partenza della lettura incrementale. Valori positivi spostano il riferimento in avanti, valori negativi lo spostano indietro.
Offset e valore visualizzato

Il valore mostrato in interfaccia per Current value non include l'offset. L'offset viene applicato dinamicamente solo al recupero dati.

The displayed value does not include the offset yet. The offset will be dynamically applied when retrieving the data.

L'offset viene sommato al valore memorizzato in base al tipo del campo incrementale:

Tipo del valore memorizzatoEffetto dell'offset
Data/ora (DateTime)Viene aggiunto (o sottratto) l'equivalente in tick corrispondente all'intero configurato
Numero (intero, decimale, ecc.)Viene sommato algebricamente all'intero configurato

Se Offset è valorizzato ma il valore memorizzato non è né una data né un tipo numerico, il recupero dati fallisce con errore:

Cannot apply offset to field '{nome_variabile}'. The value is neither a DateTime nor a numeric type (Current type: {tipo}).

Esempio di utilizzo dell'offset​

Campo numerico — valore memorizzato 30, offset 5 → valore utilizzato per la query alla sorgente: 35.

Con offset -5 e valore memorizzato 30, il valore utilizzato sarebbe 25. Utile per rileggere una piccola finestra di record già elaborati.

Campo data/ora — valore memorizzato 2024-06-01T12:00:00, offset -36000000000 (equivalente a −1 ora in tick .NET) → valore utilizzato per la query: 2024-06-01T11:00:00. Un offset negativo su un timestamp consente di spostare indietro il punto di partenza e sovrapporre dati già processati.

Esempio di configurazione​

Nell’immagine seguente è riportato un esempio di configurazione che utilizza il campo last_update come discriminante per la lettura incrementale.

Source incremental

Esecuzione manuale​

Flowlyze consente di avviare manualmente un flusso, senza attendere il naturale innesco della sua pianificazione temporale. L'esecuzione manuale prevede due modalità distinte, a seconda della necessità di personalizzare o meno il comportamento del job a runtime.

Run Now (Esecuzione standard)​

È l'esecuzione immediata e diretta del flusso. Il job viene inserito istantaneamente nella coda di elaborazione mantenendo la configurazione della sorgente esattamente così com'è stata salvata nel sistema, senza alcuna modifica o aggiunta di parametri.

Run With Options (Esecuzione con opzioni)​

Consente di sovrascrivere o arricchire temporaneamente l'esecuzione del flusso passando dei parametri di runtime. Questa modalità permette di:

  • Pianificare un'esecuzione differita: impostando una data e un ora specifica nel futuro in cui far partire il flusso, entro un limite massimo.
  • Passare variabili istanza (instanceFields): una mappa di chiavi e valori dinamici utili a sovrascrivere o popolare la configurazione della sorgente (es. iniettare range di date o Id specifici per quel singolo run).

Run Now vs Run with Options​

ModalitàEndpoint APIBodyComportamento
Run NowPOST /api/jobs/{flowId}(vuoto)Accoda immediatamente l’esecuzione del job schedulato.
Run with OptionsPOST /api/jobs/run/{flowId}FlowRunOptionsAccoda subito oppure pianifica l’esecuzione; applica variabili istanza alla configurazione della sorgente.

Requisiti​

Per poter invocare correttamente un'esecuzione manuale (in entrambe le modalità), devono essere soddisfatti i seguenti requisiti:)

  • Configurazione del flusso: Il flusso deve avere una sorgente schedulata (modalità attiva/pull) e deve avere lo schedule abilitato (isScheduleEnable = true). In caso contrario, il sistema restituirà errore 400 Bad Request.
  • Autorizzazioni: L'utente deve disporre del permesso flows:run.
  • Contesto: Le chiamate HTTP devono includere il contesto del tenant (header x-tenant-id).

Utilizzo da interfaccia utente​

Nella pagina di dettaglio di un flusso, il pulsante Run espone un menu con due opzioni:

  1. Run Now — esecuzione immediata senza opzioni.
  2. Run Now with Options — apre una finestra di dialogo in cui l'utente può configurare:
  • Next Execution — data e ora opzionale per l’esecuzione differita.
  • Flow Run Options (instanceFields) — mappa chiave/valore di variabili da usare solo per quell’esecuzione.
Nota di usabilità

Le opzioni inserite nel pannello vengono memorizzate localmente nel browser (per singolo tenant e flusso) e riproposte automaticamente all'apertura successiva del dialogo per velocizzare i test ripetuti.

Utilizzo via API​

Endpoint Run Now (standard)​

ElementoValore
Endpointhttps://api.flowlyze.io/api/jobs/{flowId}
MetodoPOST
Path parameterflowId — identificativo del flusso
HeaderAuthorization: Bearer <token>, x-tenant-id: <tenant_id>, Content-Type: application/json
BodyNon richiesto (vuoto)
Esempio di chiamata Run Now​
curl -X POST \
  https://api.flowlyze.io/api/jobs/{flowId} \
  -H "Authorization: Bearer <TOKEN>" \
  -H "x-tenant-id: <TENANT_NAME>"

Endpoint Run with Options​

ElementoValore
Endpointhttps://api.flowlyze.io/api/jobs/run/{flowId}
MetodoPOST
Path parameterflowId — identificativo del flusso
HeaderAuthorization: Bearer <token>, x-tenant-id: <tenant_id>, Content-Type: application/json
BodyOggetto JSON FlowRunOptions (Tutti i campi interni sono opzionali).
Modello FlowRunOptions​
Campo (JSON)TipoObbligatorioDescrizione
nextExecutionstringNoData e ora UTC di esecuzione. Se omesso, il job viene accodato immediatamente.
instanceFieldsobject (Mappa stringa -> valore)NoVariabili di runtime iniettare nella sorgente. Sovrascrivono eventuali variabili globali.
Formato date

Il campo nextExecution accetta stringhe in formato ISO 8601 con indicazione dell'offset o in formato UTC nativo (es. 2026-06-23T14:30:00Z).

Esempio di chiamata Run with Options​
curl -X POST \
  https://api.flowlyze.io/api/jobs/run/{flowId} \
  -H "Authorization: Bearer <TOKEN>" \
  -H "x-tenant-id: <TENANT_NAME>"
  -d'{
      "nextExecution": "2026-06-23T22:00:00Z",
      "instanceFields": {
        "bodyOption": "Value of body option"
      }
    }'

Parametri in dettaglio​

nextExecution — esecuzione differita​

Quando nextExecution è valorizzato, Flowlyze non esegue subito il flusso: registra un job Hangfire pianificato per l’istante indicato.

Limite massimo

La data di esecuzione non può superare un intervallo massimo rispetto all’orario corrente (default: 24 ore). Se nextExecution eccede tale limite, viene automaticamente ridotta al valore massimo consentito.

Se nextExecution non è impostato, il job viene accodato immediatamente (comportamento equivalente a Run Now).

instanceFields — variabili di runtime​

Le variabili istanza sono coppie chiave/valore fornite al momento del run manuale. Vengono unite alle variabili globali del tenant e utilizzate per la sostituzione Handlebars nei campi della configurazione della sorgente, con la stessa sintassi già usata altrove in Flowlyze: {{nome_variabile}}.

Esempio d’uso

Supponiamo una sorgente HTTP con URL configurato come:

https://api.example.com/orders?from={{dataDa}}&to={{dataA}}

Con una chiamata Run with Options che imposta:

{
  "instanceFields": {
    "dataDa": "2025-06-01",
    "dataA": "2025-06-23"
  }
}

al momento dell’esecuzione l’URL effettivo diventerà:

https://api.example.com/orders?from=2025-06-01&to=2025-06-23
Precedenza

Le variabili in instanceFields vengono aggiunte al dizionario delle variabili del tenant per quella singola esecuzione. In caso di chiavi duplicate, il valore fornito in instanceFields ha priorità su quello globale.

Modalità Passiva (push)​

Flowlyze espone un endpoint al quale i sistemi esterni inviano i dati (singoli record o batch). È il modello ideale per eventi in tempo quasi reale.

Sicurezza Webhook

Proteggi l’endpoint con API Key e limita il path per tenant. Verifica i limiti di rate e firma dei messaggi quando disponibili.

Esempio d’uso

  • Un e-commerce invia una variazione ordine tramite webhook verso l’endpoint REST di Flowlyze.

Di seguito un’immagine che rappresenta l’attivazione di un endpoint, con path “test” e apikey “my-secret-apikey”

Template piattaforma

Modalità combinate​

Le due modalità possono coesistere. La stessa sorgente può:

  • inviare eventi in push (webhook) per gli aggiornamenti,
  • essere sincronizzata in pull a cadenza regolare per garantire allineamento e consistenza.

Processo unificato​

Indipendentemente dalla modalità di ingresso (attiva o passiva), tutti i dati:

  1. Vengono acquisiti nel database interno di Flowlyze.
  2. Sono aggregati e normalizzati (schema, formati, codifiche).
  3. Seguono lo stesso flusso di trasformazione, arricchimento, validazione e consegna verso le destinazioni previste.

Questo approccio assicura uniformità operativa, tracciabilità e scalabilità del trattamento dati.

Http Ingestion (Push)​

Endpoint V2​

L’endpoint V2 consente di definire il modello dati lato client, svincolando il chiamante dal dover rispettare uno schema preconfigurato. In pratica, chiunque disponga di URL (Uniform Resource Locator) e API Key può scegliere liberamente il formato del payload, entro le opzioni supportate.

Il parser viene scelto automaticamente in base al MIME type dichiarato nell’header `Content-Type`:

  • JSON → `application/json`
  • XML → `application/xml, text/xml`
  • Form → `application/x-www-form-urlencoded, multipart/form-data`

Invio in formato JSON​

Nel caso in cui sia indicato il parametro x-fl-selector viene utilizzato il sotto oggetto puntato Selettore opzionale x-fl-selector

Se è presente il query parameter HTTP x-fl-selector, l’endpoint applica l’espressione JSONPath indicata per individuare il sotto-oggetto (o l’elenco) da processare. Se l’header non è presente, viene utilizzato l’oggetto alla root del body JSON.L’endpoint identifica automaticamente se il payload rappresenta un singolo messaggio (oggetto) oppure un insieme di messaggi (array). Le espressioni JSONPath possono selezionare elementi annidati e liste.

Esempi (JSON)​
#Input (schematico)Content-Typex-fl-selectorInterpretazione
1`[{}, {}, {}]`application/json(omesso)L’intero array è un batch: 3 messaggi.
2`{ "data": [ {}, {}, {} ] }`application/jsondataSeleziona data: batch di 3 messaggi.
3`[ { "children": [ {}, {}, {} ] }, { "children": [ {}, {}, {} ] }, { "children": [ {}, {}, {} ] } ]`application/json$.*.children[*]Seleziona e appiattisce tutti i children: 9 messaggi.

Nota: gli esempi sono volutamente compatti e privi di campi applicativi; in produzione i singoli {} rappresentano record/oggetti completi.

Se il parsing del payload fallisce, Flowlyze crea comunque un messaggio di fallback che contiene un campo content con il contenuto originale ricevuto. Questo garantisce tracciabilità e possibilità di ritentare l’elaborazione a valle.

Invio batchId​

L’endpoint consente l’invio di un Batch ID per richiesta. I parametri possono essere forniti tramite header oppure tramite query string.

Parametri supportati​

Nome parametroTipoObbligatorioDescrizione
batchIdstringSìIdentificativo univoco del batch (ObjectId). È consentito un solo valore per richiesta.
processableBatchbooleanNoIndica se il batch deve essere processato.

Regole di elaborazione​

  • Se processableBatch = true → lo stato del batch viene impostato a PROCESSING
  • Se processableBatch = false o non fornito → lo stato del batch viene impostato a DATA_LOADING

Endpoint V1​

L’endpoint V1, in fase di dismissione, si basa su una struttura dati definita a priori. Ciò significa che il flusso deve essere configurato anticipatamente per interpretare correttamente il payload in ingresso. All’interno del flusso è necessario specificare:

Deprecato

L’endpoint V1 è deprecato e in dismissione. Si consiglia la migrazione verso V2 per flessibilità e robustezza.

  • Consider as single object: se l’endpoint riceve un singolo oggetto
  • path: percorso univoco della url (ultima parte del percorso https://in.flowlyze.io/api/wh/{tenant_id}/{path})
  • apikey: chiave univoca da passare in query string oppure in header sotto con nome di parametro x-api-key

Esempio: https://in.flowlyze.io/api/wh/flowlyze-demo/test for tenant flowlyze-demo and path test.

Parametri richiesti nel flusso​
Consider as single object​

Indica se l’endpoint riceve un singolo oggetto JSON per richiesta (true) oppure un insieme di oggetti (array).

  • true → il payload è un singolo record.
  • false → il payload è una lista di record.
Path​

Percorso univoco in coda alla URL che identifica l’istanza del webhook all’interno del tenant. Esempio: in /flowlyze-demo/test, il path è test.

API Key​

Chiave univoca da fornire per l’autenticazione:

  • In query string: ?x-api-key=<CHIAVE>
  • Oppure via HTTP header: x-api-key: <CHIAVE>
Esempio di URL​

Per il tenant flowlyze-demo e path test: https://in.flowlyze.io/api/wh/flowlyze-demo/test

Autenticazione con query string: https://in.flowlyze.io/api/wh/flowlyze-demo/prova?x-api-key=<CHIAVE>

Autenticazione con HTTP header (consigliato), esempio curl:

curl -X POST \
  https://in.flowlyze.io/api/wh/flowlyze-demo/prova \
  -H "x-api-key: <CHIAVE>" \
  -H "Content-Type: application/json" \
  -d '{ "example": "value" }'

Stato di vita e migrazione​

L’endpoint V1 è in dismissione. Per una maggiore flessibilità di schema (selettori, formati, autodiscovery del batch), si raccomanda la migrazione a Endpoint V2, che permette la definizione del modello dati lato client e supporta selezione tramite JSONPath.

Connettori​